blob: 02103272a1f5a29f186b0c7ac2ba7e4c4fd1bc35 (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
|
#pragma once
#include <Poco/Event.h>
#include <Core/BackgroundSchedulePool.h>
#include <base/types.h>
#include <thread>
#include <atomic>
#include <Common/logger_useful.h>
namespace DB
{
class StorageReplicatedMergeTree;
/** Initializes ZK session.
* Exposes ephemeral nodes. It sets the node values that are required for replica detection.
* Starts participation in the leader selection. Starts all background threads.
* Then monitors whether the session has expired. And if it expired, it will reinitialize it.
*/
class ReplicatedMergeTreeRestartingThread
{
public:
explicit ReplicatedMergeTreeRestartingThread(StorageReplicatedMergeTree & storage_);
void start(bool schedule = true)
{
LOG_TRACE(log, "Starting restating thread, schedule: {}", schedule);
if (schedule)
task->activateAndSchedule();
else
task->activate();
}
void wakeup() { task->schedule(); }
void shutdown(bool part_of_full_shutdown);
void run();
private:
StorageReplicatedMergeTree & storage;
String log_name;
Poco::Logger * log;
std::atomic<bool> need_stop {false};
/// The random data we wrote into `/replicas/me/is_active`.
String active_node_identifier;
BackgroundSchedulePool::TaskHolder task;
Int64 check_period_ms; /// The frequency of checking expiration of session in ZK.
UInt32 consecutive_check_failures = 0; /// How many consecutive checks have failed
bool first_time = true; /// Activate replica for the first time.
/// Restarts table if needed, returns false if it failed to restart replica.
bool runImpl();
/// Start or stop background threads. Used for partial reinitialization when re-creating a session in ZooKeeper.
bool tryStartup(); /// Returns false if ZooKeeper is not available.
/// Note in ZooKeeper that this replica is currently active.
void activateReplica();
/// Delete the parts for which the quorum has failed (for the time when the replica was inactive).
void removeFailedQuorumParts();
/// If there is an unreachable quorum, and we have a part, then add this replica to the quorum.
void updateQuorumIfWeHavePart();
void partialShutdown(bool part_of_full_shutdown = false);
/// Set readonly mode for table
void setReadonly(bool on_shutdown = false);
/// Disable readonly mode for table
void setNotReadonly();
};
}
|
