1#pragma once
2
3#include <Poco/Event.h>
4#include <common/logger_useful.h>
5#include <Core/BackgroundSchedulePool.h>
6#include <Core/Types.h>
7#include <thread>
8#include <atomic>
9
10
11namespace DB
12{
13
14class StorageReplicatedMergeTree;
15
16
17/** Initializes ZK session.
18 * Exposes ephemeral nodes. It sets the node values that are required for replica detection.
19 * Starts participation in the leader selection. Starts all background threads.
20 * Then monitors whether the session has expired. And if it expired, it will reinitialize it.
21 */
22class ReplicatedMergeTreeRestartingThread
23{
24public:
25 ReplicatedMergeTreeRestartingThread(StorageReplicatedMergeTree & storage_);
26
27 void start() { task->activateAndSchedule(); }
28
29 void wakeup() { task->schedule(); }
30
31 void shutdown();
32
33private:
34 StorageReplicatedMergeTree & storage;
35 String log_name;
36 Logger * log;
37 std::atomic<bool> need_stop {false};
38
39 /// The random data we wrote into `/replicas/me/is_active`.
40 String active_node_identifier;
41
42 BackgroundSchedulePool::TaskHolder task;
43 Int64 check_period_ms; /// The frequency of checking expiration of session in ZK.
44 bool first_time = true; /// Activate replica for the first time.
45 time_t prev_time_of_check_delay = 0;
46 bool startup_completed = false;
47
48 void run();
49
50 /// Start or stop background threads. Used for partial reinitialization when re-creating a session in ZooKeeper.
51 bool tryStartup(); /// Returns false if ZooKeeper is not available.
52
53 /// Note in ZooKeeper that this replica is currently active.
54 void activateReplica();
55
56 /// Delete the parts for which the quorum has failed (for the time when the replica was inactive).
57 void removeFailedQuorumParts();
58
59 /// If there is an unreachable quorum, and we have a part, then add this replica to the quorum.
60 void updateQuorumIfWeHavePart();
61
62 void partialShutdown();
63};
64
65
66}
67