Update worker replication config (#3373)

temporalio · Sep 26, 2020 · d516b40 · d516b40
1 parent 1b00d51
commit d516b40
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 1 deletion.
diff --git a/common/service/dynamicconfig/constants.go b/common/service/dynamicconfig/constants.go
@@ -272,6 +272,7 @@ var keys = map[Key]string{
 	WorkerReplicationTaskContextDuration:            "worker.replicationTaskContextDuration",
 	WorkerReReplicationContextTimeout:               "worker.workerReReplicationContextTimeout",
 	WorkerEnableRPCReplication:                      "worker.enableWorkerRPCReplication",
+	WorkerEnableKafkaReplication:                    "worker.enableKafkaReplication",
 	WorkerIndexerConcurrency:                        "worker.indexerConcurrency",
 	WorkerESProcessorNumOfWorkers:                   "worker.ESProcessorNumOfWorkers",
 	WorkerESProcessorBulkActions:                    "worker.ESProcessorBulkActions",
@@ -701,6 +702,8 @@ const (
 	WorkerReReplicationContextTimeout
 	// WorkerEnableRPCReplication is the feature flag for RPC replication
 	WorkerEnableRPCReplication
+	// 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
@@ -81,6 +81,7 @@ type (
 		ReplicationTaskContextTimeout      dynamicconfig.DurationPropertyFn
 		ReReplicationContextTimeout        dynamicconfig.DurationPropertyFnWithNamespaceIDFilter
 		EnableRPCReplication               dynamicconfig.BoolPropertyFn
+		EnableKafkaReplication             dynamicconfig.BoolPropertyFn
 	}
 )
 

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