Update worker replication config (#3373)

cadence-workflow · Aug 1, 2020 · 6e9fea4 · 6e9fea4
1 parent 98e14e9
commit 6e9fea4
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 11 deletions.
diff --git a/common/service/dynamicconfig/constants.go b/common/service/dynamicconfig/constants.go
@@ -273,7 +273,7 @@ var keys = map[Key]string{
 	WorkerReplicationTaskContextDuration:             "worker.replicationTaskContextDuration",
 	WorkerReReplicationContextTimeout:                "worker.workerReReplicationContextTimeout",
 	WorkerEnableRPCReplication:                       "worker.enableWorkerRPCReplication",
-	WorkerEnableHistoryReplication:                   "worker.enableWorkerHistoryReplication",
+	WorkerEnableKafkaReplication:                     "worker.enableKafkaReplication",
 	WorkerIndexerConcurrency:                         "worker.indexerConcurrency",
 	WorkerESProcessorNumOfWorkers:                    "worker.ESProcessorNumOfWorkers",
 	WorkerESProcessorBulkActions:                     "worker.ESProcessorBulkActions",
@@ -720,8 +720,8 @@ const (
 	WorkerReReplicationContextTimeout
 	// WorkerEnableRPCReplication is the feature flag for RPC replication
 	WorkerEnableRPCReplication
-	// WorkerEnableHistoryReplication is the feature flag for history replication
-	WorkerEnableHistoryReplication
+	// WorkerEnableKafkaReplication is the feature flag for kafka replication
+	WorkerEnableKafkaReplication
 	// WorkerIndexerConcurrency is the max concurrent messages to be processed at any given time
 	WorkerIndexerConcurrency
 	// WorkerESProcessorNumOfWorkers is num of workers for esProcessor

diff --git a/service/worker/replicator/replicator.go b/service/worker/replicator/replicator.go
@@ -80,7 +80,7 @@ type (
 		ReplicationTaskContextTimeout      dynamicconfig.DurationPropertyFn
 		ReReplicationContextTimeout        dynamicconfig.DurationPropertyFnWithDomainIDFilter
 		EnableRPCReplication               dynamicconfig.BoolPropertyFn
-		EnableHistoryReplication           dynamicconfig.BoolPropertyFn
+		EnableKafkaReplication             dynamicconfig.BoolPropertyFn
 	}
 )
 
@@ -146,11 +146,9 @@ func (r *Replicator) Start() error {
 		}
 	}
 
-	if r.config.EnableHistoryReplication() {
-		for _, processor := range r.processors {
-			if err := processor.Start(); err != nil {
-				return err
-			}
+	for _, processor := range r.processors {
+		if err := processor.Start(); err != nil {
+			return err
 		}
 	}
 	for _, domainProcessor := range r.domainProcessors {

diff --git a/service/worker/service.go b/service/worker/service.go
@@ -121,7 +121,7 @@ func NewConfig(params *service.BootstrapParams) *Config {
 			ReplicationTaskContextTimeout:      dc.GetDurationProperty(dynamicconfig.WorkerReplicationTaskContextDuration, 30*time.Second),
 			ReReplicationContextTimeout:        dc.GetDurationPropertyFilteredByDomainID(dynamicconfig.WorkerReReplicationContextTimeout, 0*time.Second),
 			EnableRPCReplication:               dc.GetBoolProperty(dynamicconfig.WorkerEnableRPCReplication, false),
-			EnableHistoryReplication:           dc.GetBoolProperty(dynamicconfig.WorkerEnableRPCReplication, true),
+			EnableKafkaReplication:             dc.GetBoolProperty(dynamicconfig.WorkerEnableKafkaReplication, false),
 		},
 		ArchiverConfig: &archiver.Config{
 			ArchiverConcurrency:           dc.GetIntProperty(dynamicconfig.WorkerArchiverConcurrency, 50),
@@ -189,7 +189,7 @@ func (s *Service) Start() {
 		s.startIndexer()
 	}
 
-	if s.GetClusterMetadata().IsGlobalDomainEnabled() {
+	if s.GetClusterMetadata().IsGlobalDomainEnabled() && s.config.ReplicationCfg.EnableKafkaReplication() {
 		s.startReplicator()
 	}
 	if s.GetArchivalMetadata().GetHistoryConfig().ClusterConfiguredForArchival() {