Repository: OpenAtomFoundation/pikiwidb (language: cpp)

Issue:
If a binlog task blocks the slave for a long period, the master may resume the increment replication from an incorrect  start position after timeout reconnection
### Is this a regression?

No

### Description

**在增量同步的过程中，如果某条BinlogTask阻塞了Slave很长时间（超过20s的超时时间），会造成主从断联，当主从再次试图重建连接时，Slave可能会提供一个错误的Binlog Start Point给Master来进行续传**。

**针对这一Case的展开描述**:
**场景**：假设Slave在增量同步阶段，发生超时转为了TryConnect状态，在发出TrySync请求时，如果相应的BinlogWorker正好还在执行一个上次连接期间的Binlog任务（将该Binlog任务定义为任务A) 。 那么，因为任务A的存在，Slave是否可能会发送了不正确的续传起始Offset给Master ？ 换句话说：虽然最终任务A会在Slave落盘，但是Slave是否可能告知了Master从任务A（所对应的Binlog）开始的位置发送Binlog，导致Master会将任务A对应的Binlog再发一次，且Slave会消费两次任务A所对应的Binlog ？  先做回答: 会出现这种问题。具体梳理在下面：

**背景补充和定义**：TrySync请求携带的offset(定义为offset A)主要作用是给Master判断是否能做增量同步，真正决定Master初始化发送窗口起始位置（续传Binlog的起始位置）的，是slave在收到TrySync的Resp后发出的一条特殊BinlogACK (is_first_send=true)，该ACK的offst start等于end(定义为Offset B)，且该Offset B就是slave在发出这条特殊Ack时的最新落盘点。在大多数情况下，其实Offset B应该会和Offset A相同，但是在Edge Case 1中，由于发出TrySync时任务A正在执行且最终会执行完毕，Offset B的正确预期应当是  Offset A + BinlogSizeOf(任务A), 或者说Binlog的起始续传位置应该就在任务A对应的Binlogs的后面。

**现有代码逻辑中**：Offset B可能不会等于正确预期，假如BinlogWork执行任务A执行了比较久，而在任务A的执行期间Slave完成了：TrySync Req发送 ->TrySync Resp处理->在TrySync Resp的处理函数末尾会获取Slave最新的Binlog落盘点（Offset B），发出特殊BinlogAck(is_first_send=true)来告知Master从哪里开始重发Binlog。 如果在获取Offset B的时候，任务A还没落盘，或者落盘了但是还没有将最新的offset更新到version_对象中(Offset B就是从version_对象中取)，就会导致获取的OffsetB就等于Offset A(但正确的应该Offset B应该是 Offset A + Offset A + BinlogSizeOf(任务A))，于是Master会把任务A所对应的那批Binlog再发一遍。

**后果**：Master会把任务A内部的BInlog再发一次，Slave等于消费了2批同样的Binlog，一次是在上一次Session的末尾（阻塞的那个任务A），一次是在新Session的第一个Binlog任务中。可能会导致主从数据不一致，且在同步没有滞后/完全同步的情况下，Slave的Binlog文件会比Master的对应BInlog文件长一点，在Slave切走这个Binlog文件之前如果发生了断点重连，主从建联可能会报错 binlog offset is not a start point of master, 但是如果slave在处于该Binlog文件期间完全没有断开过，可能消费到了窗口末尾时Master端会报出奇怪的SyncWin Corruption。(另外，在线上实例的主从超时故障中，确实也出现过binlog offset is not a start point of master)


**During incremental synchronization, if a BinlogTask blocks the Slave for a long time (exceeding the 20-second timeout), when the Master and Slave attempt to reconnect, the Slave may provide an incorrect Binlog Start Point to the Master for resuming the process.**

**Expanded description of this case**:
**Scenario**: Suppose during the incremental synchronization phase, the Slave times out and switches to TryConnect state. When sending a TrySync request, if the corresponding BinlogWorker happens to still be executing a Binlog task from the previous connection period (let's define this Binlog task as Task A), then due to the existence of Task A, could the Slave possibly send an incorrect resumption starting Offset to the Master? In other words, even though Task A will eventually be written to the Slave’s disk, could the Slave inform the Master to send the Binlog from the starting point of Task A, causing the Master to resend the Binlog corresponding to Task A and resulting in the Slave consuming the Binlog for Task A twice? The preliminary answer is: yes, this issue can occur. The detailed explanation is below:

**Background and definition**: The offset carried by the TrySync request (defined as Offset A) primarily serves for the Master to determine whether incremental synchronization can be performed. The actual determination of the starting position of the Master’s initial sending window (the starting position for resuming the Binlog) is done by a special BinlogACK (with is_first_send=true) sent by the Slave after receiving the TrySync response. The start of this ACK’s offset is equal to the end (defined as Offset B), and this Offset B is the latest disk write point of the Slave when sending this special Ack. In most cases, Offset B should be the same as Offset A, but in Edge Case 1, since Task A is executing when the TrySync is sent and will eventually complete, the correct expectation for Offset B should be Offset A + BinlogSizeOf(Task A), or in other words, the starting position for resuming the Binlog should be right after the Binlogs corresponding to Task A.

**In the current code logic**: Offset B may not equal the correct expectation. If the BinlogWork executing Task A takes a long time, and during the execution of Task A, the Slave completes: sending TrySync Req -> processing TrySync Resp -> obtaining the latest Binlog write point (Offset B) at the end of the TrySync Resp processing function, and sending a special BinlogAck (is_first_send=true) to inform the Master where to start resending the Binlog. If Task A has not been written to disk or the latest offset has not been updated to the version_ object (Offset B is taken from the version_ object) when obtaining Offset B, then the obtained Offset B will equal Offset A (but the correct Offset B should be Offset A + BinlogSizeOf(Task A)), thus causing the Master to resend the batch of Binlogs corresponding to Task A.

**Consequence**: The Master will resend the Binlogs within Task A, causing the Slave to consume two batches of the same Binlog, once at the end of the last session (the blocking Task A) and once in the first Binlog task of the new session. This can lead to data inconsistency between the Master and Slave. Additionally, if there is no lag in synchronization or if synchronization is complete, the Slave’s Binlog file may be slightly longer than the corresponding Binlog file on the Master. If a breakpoint reconnection occurs before the Slave switches from this Binlog file, the Master-Slave connection might report an error: binlog offset is not a start point of master. However, if the Slave never disconnects during this Binlog file, a strange SyncWin Corruption error might be reported on the Master side when the window reaches the end. (Furthermore, in the actual Master-Slave timeout failures in online instances, the binlog offset is not a start point of master error has indeed occurred.)


### Please provide a link to a minimal reproduction of the bug

_No response_

### Screenshots or videos


### Please provide the version you discovered this bug in (check about page for version information)

_No response_

### Anything else?

_No response_


Discussion / hints:

Fix patch:
diff --git a/conf/pika.conf b/conf/pika.conf
index 055208dae5..a99c30b30d 100644
--- a/conf/pika.conf
+++ b/conf/pika.conf
@@ -34,10 +34,17 @@ slow-cmd-thread-pool-size : 1
 # Slow cmd list e.g. hgetall, mset
 slow-cmd-list :
 
-# The number of sync-thread for data replication from master, those are the threads work on slave nodes
-# and are used to execute commands sent from master node when replicating.
+# The number of threads to write DB in slaveNode when replicating.
+# It's preferable to set slave's sync-thread-num value close to master's thread-pool-size.
 sync-thread-num : 6
 
+# The num of threads to write binlog in slaveNode when replicating,
+# each DB cloud only bind to one sync-binlog-thread to write binlog in maximum
+#[NOTICE] It's highly recommended to set sync-binlog-thread-num equal to conf item 'database'(then each DB cloud have a exclusive thread to write binlog),
+# eg. if you use 8 DBs(databases_ is 8), sync-binlog-thread-num is preferable to be 8
+# Valid range of sync-binlog-thread-num is [1, databases], the final value of it is Min(sync-binlog-thread-num, databases)
+sync-binlog-thread-num : 1
+
 # Directory to store log files of Pika, which contains multiple types of logs,
 # Including: INFO, WARNING, ERROR log, as well as binglog(write2fine) file which
 # is used for replication.
@@ -101,6 +108,8 @@ instance-mode : classic
 # The default database id is DB 0. You can select a different one on
 # a per-connection by using SELECT. The db id range is [0, 'databases' value -1].
 # The value range of this parameter is [1, 8].
+# [NOTICE] It's RECOMMENDED to set sync-binlog-thread-num equal to DB num(databases),
+# if you've changed the value of databases, remember to check if the value of sync-binlog-thread-num is proper.
 databases : 1
 
 # The number of followers of a master. Only [0, 1, 2, 3, 4] is valid at present.
diff --git a/include/pika_conf.h b/include/pika_conf.h
index 2a5ed25212..e0cb81062d 100644
--- a/include/pika_conf.h
+++ b/include/pika_conf.h
@@ -65,6 +65,10 @@ class PikaConf : public pstd::BaseConf {
     std::shared_lock l(rwlock_);
     return sync_thread_num_;
   }
+  int sync_binlog_thread_num() {
+    std::shared_lock l(rwlock_);
+    return sync_binlog_thread_num_;
+  }
   std::string log_path() {
     std::shared_lock l(rwlock_);
     return log_path_;
@@ -793,6 +797,7 @@ class PikaConf : public pstd::BaseConf {
   int slow_cmd_thread_pool_size_ = 0;
   std::unordered_set<std::string> slow_cmd_set_;
   int sync_thread_num_ = 0;
+  int sync_binlog_thread_num_ = 0;
   int expire_dump_days_ = 3;
   int db_sync_speed_ = 0;
   std::string slaveof_;
diff --git a/include/pika_define.h b/include/pika_define.h
index 4d6bf6a846..91ddffeb9f 100644
--- a/include/pika_define.h
+++ b/include/pika_define.h
@@ -30,6 +30,8 @@
 #define PIKA_SERVER_ID_MAX 65535
 
 class PikaServer;
+/* Global Const */
+constexpr int MAX_DB_NUM = 8;
 
 /* Port shift */
 const int kPortShiftRSync = 1000;
diff --git a/include/pika_repl_client.h b/include/pika_repl_client.h
index aa20b83c78..49eb2c9b82 100644
--- a/include/pika_repl_client.h
+++ b/include/pika_repl_client.h
@@ -65,6 +65,7 @@ class PikaReplClient {
   pstd::Status Close(const std::string& ip, int port);
 
   void Schedule(net::TaskFunc func, void* arg);
+  void ScheduleByDBName(net::TaskFunc func, void* arg, const std::string& db_name);
   void ScheduleWriteBinlogTask(const std::string& db_name, const std::shared_ptr<InnerMessage::InnerResponse>& res,
                                const std::shared_ptr<net::PbConn>& conn, void* res_private_data);
   void ScheduleWriteDBTask(const std::shared_ptr<Cmd>& cmd_ptr, const LogOffset& offset, const std::string& db_name);
@@ -80,13 +81,15 @@ class PikaReplClient {
   pstd::Status SendRemoveSlaveNode(const std::string& ip, uint32_t port, const std::string& db_name, const std::string& local_ip);
 
  private:
-  size_t GetHashIndex(const std::string& key, bool upper_half);
-  void UpdateNextAvail() { next_avail_ = (next_avail_ + 1) % static_cast<int32_t>(bg_workers_.size()); }
+  size_t GetBinlogWorkerIndexByDBName(const std::string &db_name);
+  size_t GetHashIndexByKey(const std::string& key);
+  void UpdateNextAvail() { next_avail_ = (next_avail_ + 1) % static_cast<int32_t>(write_binlog_workers_.size()); }
 
   std::unique_ptr<PikaReplClientThread> client_thread_;
   int next_avail_ = 0;
   std::hash<std::string> str_hash;
-  std::vector<std::unique_ptr<PikaReplBgWorker>> bg_workers_;
+  std::vector<std::unique_ptr<PikaReplBgWorker>> write_binlog_workers_;
+  std::vector<std::unique_ptr<PikaReplBgWorker>> write_db_workers_;
 };
 
 #endif
diff --git a/include/pika_rm.h b/include/pika_rm.h
index 9cc0c5ef38..b3e310ddf4 100644
--- a/include/pika_rm.h
+++ b/include/pika_rm.h
@@ -182,6 +182,7 @@ class PikaReplicaManager {
                                const std::shared_ptr<InnerMessage::InnerResponse>& res,
                                const std::shared_ptr<net::PbConn>& conn, void* res_private_data);
   void ScheduleWriteDBTask(const std::shared_ptr<Cmd>& cmd_ptr, const LogOffset& offset, const std::string& db_name);
+  void ScheduleReplClientBGTaskByDBName(net::TaskFunc , void* arg, const std::string &db_name);
   void ReplServerRemoveClientConn(int fd);
   void ReplServerUpdateClientConnMap(const std::string& ip_port, int fd);
 
diff --git a/src/pika_admin.cc b/src/pika_admin.cc
index 6760ac247b..15d49cdbae 100644
--- a/src/pika_admin.cc
+++ b/src/pika_admin.cc
@@ -942,6 +942,7 @@ void InfoCmd::InfoServer(std::string& info) {
   tmp_stream << "tcp_port:" << g_pika_conf->port() << "\r\n";
   tmp_stream << "thread_num:" << g_pika_conf->thread_num() << "\r\n";
   tmp_stream << "sync_thread_num:" << g_pika_conf->sync_thread_num() << "\r\n";
+  tmp_stream << "sync_binlog_thread_num:" << g_pika_conf->sync_binlog_thread_num() << "\r\n";
   tmp_stream << "uptime_in_seconds:" << (current_time_s - g_pika_server->start_time_s()) << "\r\n";
   tmp_stream << "uptime_in_days:" << (current_time_s / (24 * 3600) - g_pika_server->start_time_s() / (24 * 3600) + 1)
              << "\r\n";
@@ -1501,6 +1502,12 @@ void ConfigCmd::ConfigGet(std::string& ret) {
     EncodeNumber(&config_body, g_pika_conf->sync_thread_num());
   }
 
+  if (pstd::stringmatch(pattern.data(), "sync-binlog-thread-num", 1) != 0) {
+     elements += 2;
+     EncodeString(&config_body, "sync-binlog-thread-num");
+     EncodeNumber(&config_body, g_pika_conf->sync_binlog_thread_num());
+    }
+
   if (pstd::stringmatch(pattern.data(), "log-path", 1) != 0) {
     elements += 2;
     EncodeString(&config_body, "log-path");
diff --git a/src/pika_conf.cc b/src/pika_conf.cc
index adc1b26d9f..e2e0d4f885 100644
--- a/src/pika_conf.cc
+++ b/src/pika_conf.cc
@@ -176,7 +176,7 @@ int PikaConf::Load() {
 
   if (classic_mode_.load()) {
     GetConfInt("databases", &databases_);
-    if (databases_ < 1 || databases_ > 8) {
+    if (databases_ < 1 || databases_ > MAX_DB_NUM) {
       LOG(FATAL) << "config databases error, limit [1 ~ 8], the actual is: " << databases_;
     }
     for (int idx = 0; idx < databases_; ++idx) {
@@ -185,6 +185,15 @@ int PikaConf::Load() {
   }
   default_db_ = db_structs_[0].db_name;
 
+  // sync_binlog_thread_num_ must be set after the setting of databases_
+  GetConfInt("sync-binlog-thread-num", &sync_binlog_thread_num_);
+  if (sync_binlog_thread_num_ <= 0) {
+      sync_binlog_thread_num_ = databases_;
+  } else {
+      // final value is MIN(sync_binlog_thread_num, databases_)
+      sync_binlog_thread_num_ = sync_binlog_thread_num_ > databases_ ?  databases_ : sync_binlog_thread_num_;
+  }
+
   int tmp_replication_num = 0;
   GetConfInt("replication-num", &tmp_replication_num);
   if (tmp_replication_num > 4 || tmp_replication_num < 0) {
diff --git a/src/pika_repl_client.cc b/src/pika_repl_client.cc
index 2754eb2f6e..352fbdf7e5 100644
--- a/src/pika_repl_client.cc
+++ b/src/pika_repl_client.cc
@@ -27,8 +27,11 @@ extern std::unique_ptr<PikaReplicaManager> g_pika_rm;
 PikaReplClient::PikaReplClient(int cron_interval, int keepalive_timeout)  {
   client_thread_ = std::make_unique<PikaReplClientThread>(cron_interval, keepalive_timeout);
   client_thread_->set_thread_name("PikaReplClient");
-  for (int i = 0; i < 2 * g_pika_conf->sync_thread_num(); ++i) {
-    bg_workers_.push_back(std::make_unique<PikaReplBgWorker>(PIKA_SYNC_BUFFER_SIZE));
+  for (int i = 0; i < g_pika_conf->sync_binlog_thread_num(); i++) {
+      write_binlog_workers_.emplace_back(std::make_unique<PikaReplBgWorker>(PIKA_SYNC_BUFFER_SIZE));
+  }
+  for (int i = 0; i < g_pika_conf->sync_thread_num(); ++i) {
+    write_db_workers_.emplace_back(std::make_unique<PikaReplBgWorker>(PIKA_SYNC_BUFFER_SIZE));
   }
 }
 
@@ -43,49 +46,77 @@ int PikaReplClient::Start() {
     LOG(FATAL) << "Start ReplClient ClientThread Error: " << res
                << (res == net::kCreateThreadError ? ": create thread error " : ": other error");
   }
-  for (auto & bg_worker : bg_workers_) {
-    res = bg_worker->StartThread();
+  for (auto & binlog_worker : write_binlog_workers_) {
+    res = binlog_worker->StartThread();
     if (res != net::kSuccess) {
-      LOG(FATAL) << "Start Pika Repl Worker Thread Error: " << res
+      LOG(FATAL) << "Start Pika Repl Write Binlog Worker Thread Error: " << res
                  << (res == net::kCreateThreadError ? ": create thread error " : ": other error");
     }
   }
+  for (auto & db_worker : write_db_workers_) {
+        res = db_worker->StartThread();
+        if (res != net::kSuccess) {
+            LOG(FATAL) << "Start Pika Repl Write DB Worker Thread Error: " << res
+                       << (res == net::kCreateThreadError ? ": create thread error " : ": other error");
+        }
+    }
   return res;
 }
 
 int PikaReplClient::Stop() {
   client_thread_->StopThread();
-  for (auto & bg_worker : bg_workers_) {
-    bg_worker->StopThread();
+  for (auto & binlog_worker : write_binlog_workers_) {
+    binlog_worker->StopThread();
+  }
+  for (auto &db_worker: write_db_workers_) {
+    db_worker->StopThread();
   }
   return 0;
 }
 
 void PikaReplClient::Schedule(net::TaskFunc func, void* arg) {
-  bg_workers_[next_avail_]->Schedule(func, arg);
+  write_binlog_workers_[next_avail_]->Schedule(func, arg);
   UpdateNextAvail();
 }
 
+void PikaReplClient::ScheduleByDBName(net::TaskFunc func, void* arg, const std::string& db_name) {
+  size_t index = GetBinlogWorkerIndexByDBName(db_name);
+  write_binlog_workers_[index]->Schedule(func, arg);
+};
+
 void PikaReplClient::ScheduleWriteBinlogTask(const std::string& db_name,
                                              const std::shared_ptr<InnerMessage::InnerResponse>& res,
                                              const std::shared_ptr<net::PbConn>& conn, void* res_private_data) {
-  size_t index = GetHashIndex(db_name, true);
-  auto task_arg = new ReplClientWriteBinlogTaskArg(res, conn, res_private_data, bg_workers_[index].get());
-  bg_workers_[index]->Schedule(&PikaReplBgWorker::HandleBGWorkerWriteBinlog, static_cast<void*>(task_arg));
+  size_t index = GetBinlogWorkerIndexByDBName(db_name);
+  auto task_arg = new ReplClientWriteBinlogTaskArg(res, conn, res_private_data, write_binlog_workers_[index].get());
+  write_binlog_workers_[index]->Schedule(&PikaReplBgWorker::HandleBGWorkerWriteBinlog, static_cast<void*>(task_arg));
 }
 
 void PikaReplClient::ScheduleWriteDBTask(const std::shared_ptr<Cmd>& cmd_ptr, const LogOffset& offset,
                                          const std::string& db_name) {
   const PikaCmdArgsType& argv = cmd_ptr->argv();
   std::string dispatch_key = argv.size() >= 2 ? argv[1] : argv[0];
-  size_t index = GetHashIndex(dispatch_key, false);
+  size_t index = GetHashIndexByKey(dispatch_key);
   auto task_arg = new ReplClientWriteDBTaskArg(cmd_ptr, offset, db_name);
-  bg_workers_[index]->Schedule(&PikaReplBgWorker::HandleBGWorkerWriteDB, static_cast<void*>(task_arg));
+  write_db_workers_[index]->Schedule(&PikaReplBgWorker::HandleBGWorkerWriteDB, static_cast<void*>(task_arg));
+}
+
+size_t PikaReplClient::GetBinlogWorkerIndexByDBName(const std::string &db_name) {
+    char db_num_c = db_name.back();
+    int32_t db_num = db_num_c - '0';
+    //Valid range of db_num is [0, MAX_DB_NUM)
+    if (db_num < 0 || db_num >= MAX_DB_NUM) {
+        LOG(ERROR)
+                << "Corruption in consuming binlog: the last char of the db_name(extracted from binlog) is not a valid db num, the extracted db_num is "
+                << db_num_c << " while write_binlog_workers.size() is " << write_binlog_workers_.size();
+        if (db_num < 0) { assert(false && "db_num invalid, check if the db_name in the request is valid, also check the ERROR Log of Pika."); }
+    }
+    return db_num % write_binlog_workers_.size();
 }
 
-size_t PikaReplClient::GetHashIndex(const std::string& key, bool upper_half) {
-  size_t hash_base = bg_workers_.size() / 2;
-  return (str_hash(key) % hash_base) + (upper_half ? 0 : hash_base);
+size_t PikaReplClient::GetHashIndexByKey(const std::string& key) {
+  size_t hash_base = write_db_workers_.size();
+  return (str_hash(key) % hash_base);
 }
 
 Status PikaReplClient::Write(const std::string& ip, const int port, const std::string& msg) {
diff --git a/src/pika_repl_client_conn.cc b/src/pika_repl_client_conn.cc
index 672648d64d..6c8e001bf1 100644
--- a/src/pika_repl_client_conn.cc
+++ b/src/pika_repl_client_conn.cc
@@ -63,9 +63,12 @@ int PikaReplClientConn::DealMessage() {
       break;
     }
     case InnerMessage::kTrySync: {
+      const std::string& db_name = response->try_sync().slot().db_name();
+      //TrySync resp must contain db_name
+      assert(!db_name.empty());
       auto task_arg =
           new ReplClientTaskArg(response, std::dynamic_pointer_cast<PikaReplClientConn>(shared_from_this()));
-      g_pika_rm->ScheduleReplClientBGTask(&PikaReplClientConn::HandleTrySyncResponse, static_cast<void*>(task_arg));
+      g_pika_rm->ScheduleReplClientBGTaskByDBName(&PikaReplClientConn::HandleTrySyncResponse, static_cast<void*>(task_arg), db_name);
       break;
     }
     case InnerMessage::kBinlogSync: {
@@ -193,7 +196,6 @@ void PikaReplClientConn::HandleTrySyncResponse(void* arg) {
     LOG(WARNING) << "TrySync Failed: " << reply;
     return;
   }
-
   const InnerMessage::InnerResponse_TrySync& try_sync_response = response->try_sync();
   const InnerMessage::Slot& db_response = try_sync_response.slot();
   std::string db_name = db_response.db_name();
diff --git a/src/pika_rm.cc b/src/pika_rm.cc
index 3445e9fea6..a996463bc2 100644
--- a/src/pika_rm.cc
+++ b/src/pika_rm.cc
@@ -659,6 +659,10 @@ void PikaReplicaManager::ScheduleReplClientBGTask(net::TaskFunc func, void* arg)
   pika_repl_client_->Schedule(func, arg);
 }
 
+void PikaReplicaManager::ScheduleReplClientBGTaskByDBName(net::TaskFunc func, void* arg, const std::string &db_name) {
+  pika_repl_client_->ScheduleByDBName(func, arg, db_name);
+}
+
 void PikaReplicaManager::ScheduleWriteBinlogTask(const std::string& db,
                                                  const std::shared_ptr<InnerMessage::InnerResponse>& res,
                                                  const std::shared_ptr<net::PbConn>& conn, void* res_private_data) {


Test patch:
diff --git a/tests/integration/replication_test.go b/tests/integration/replication_test.go
index 51a8ad00bd..764e8bcca5 100644
--- a/tests/integration/replication_test.go
+++ b/tests/integration/replication_test.go
@@ -558,6 +558,7 @@ var _ = Describe("should replication ", func() {
 
 			log.Println("randomSpopstore test start")
 			execute(&ctx, clientMaster, 4, randomSpopstroeThread)
+			time.Sleep(10 * time.Second)
 			master_spopstore_set := clientMaster.SMembers(ctx, "set1")
 			Expect(master_spopstore_set.Err()).NotTo(HaveOccurred())
 			slave_spopstore_set := clientSlave.SMembers(ctx, "set1")