Rpc replication ack level update (uber#3266)

fydexx · May 19, 2020 · 3aab131 · 3aab131
1 parent af7583c
commit 3aab131
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 2 deletions.
diff --git a/service/history/historyEngine.go b/service/history/historyEngine.go
@@ -316,8 +316,8 @@ func (e *historyEngineImpl) Start() {
 
 	clusterMetadata := e.shard.GetClusterMetadata()
 	if e.replicatorProcessor != nil &&
-		clusterMetadata.GetReplicationConsumerConfig().Type != sconfig.ReplicationConsumerTypeRPC &&
-		e.config.EnableKafkaReplication() {
+		(clusterMetadata.GetReplicationConsumerConfig().Type != sconfig.ReplicationConsumerTypeRPC ||
+		e.config.EnableKafkaReplication()) {
 		e.replicatorProcessor.Start()
 	}
 

diff --git a/service/history/replicatorQueueProcessor.go b/service/history/replicatorQueueProcessor.go
@@ -330,6 +330,15 @@ func (p *replicatorQueueProcessorImpl) readTasks(readLevel int64) ([]task.Info,
 func (p *replicatorQueueProcessorImpl) updateAckLevel(ackLevel int64) error {
 	err := p.shard.UpdateReplicatorAckLevel(ackLevel)
 
+	// TODO: Remove this after enabled the rpc replication
+	clusterMetadata := p.shard.GetClusterMetadata()
+	for name, cluster := range clusterMetadata.GetAllClusterInfo() {
+		if !cluster.Enabled || clusterMetadata.GetCurrentClusterName() == name{
+			continue
+		}
+		p.shard.UpdateClusterReplicationLevel(name, ackLevel)
+	}
+
 	// this is a hack, since there is not dedicated ticker on the queue processor
 	// to periodically send out sync shard message, put it here
 	now := clock.NewRealTimeSource().Now()