Add feature flags for RPC replication migration (cadence-workflow#3216)

common/service/dynamicconfig/constants.go

@@ -99,6 +99,7 @@ var keys = map[Key]string{
 	EnableClientVersionCheck:              "frontend.enableClientVersionCheck",
 	ValidSearchAttributes:                 "frontend.validSearchAttributes",
 	SendRawWorkflowHistory:                "frontend.sendRawWorkflowHistory",
+	FrontendEnableRPCReplication:          "frontend.enableRPCReplication",
 	SearchAttributesNumberOfKeysLimit:     "frontend.searchAttributesNumberOfKeysLimit",
 	SearchAttributesSizeOfValueLimit:      "frontend.searchAttributesSizeOfValueLimit",
 	SearchAttributesTotalSizeLimit:        "frontend.searchAttributesTotalSizeLimit",
@@ -232,6 +233,8 @@ var keys = map[Key]string{
 	ReplicationTaskProcessorNoTaskInitialWait:              "history.ReplicationTaskProcessorNoTaskInitialWait",
 	ReplicationTaskProcessorCleanupInterval:                "history.ReplicationTaskProcessorCleanupInterval",
 	ReplicationTaskProcessorCleanupJitterCoefficient:       "history.ReplicationTaskProcessorCleanupJitterCoefficient",
+	HistoryEnableRPCReplication:                            "history.EnableRPCReplication",
+	HistoryEnableKafkaReplication:                          "history.EnableKafkaReplication",
 	EnableConsistentQuery:                                  "history.EnableConsistentQuery",
 	EnableConsistentQueryByDomain:                          "history.EnableConsistentQueryByDomain",
 	MaxBufferedQueryCount:                                  "history.MaxBufferedQueryCount",
@@ -251,6 +254,7 @@ var keys = map[Key]string{
 	WorkerReplicationTaskMaxRetryDuration:           "worker.replicationTaskMaxRetryDuration",
 	WorkerReplicationTaskContextDuration:            "worker.replicationTaskContextDuration",
 	WorkerReReplicationContextTimeout:               "worker.workerReReplicationContextTimeout",
+	WorkerEnableRPCReplication:                      "worker.enableWorkerRPCReplication",
 	WorkerIndexerConcurrency:                        "worker.indexerConcurrency",
 	WorkerESProcessorNumOfWorkers:                   "worker.ESProcessorNumOfWorkers",
 	WorkerESProcessorBulkActions:                    "worker.ESProcessorBulkActions",
@@ -382,6 +386,8 @@ const (
 	ValidSearchAttributes
 	// SendRawWorkflowHistory is whether to enable raw history retrieving
 	SendRawWorkflowHistory
+	// FrontendEnableRPCReplication is a feature flag for rpc replication
+	FrontendEnableRPCReplication
 	// SearchAttributesNumberOfKeysLimit is the limit of number of keys
 	SearchAttributesNumberOfKeysLimit
 	// SearchAttributesSizeOfValueLimit is the size limit of each value
@@ -657,6 +663,8 @@ const (
 	WorkerReplicationTaskContextDuration
 	// WorkerReReplicationContextTimeout is the context timeout for end to end  re-replication process
 	WorkerReReplicationContextTimeout
+	// WorkerEnableRPCReplication is the feature flag for RPC replication
+	WorkerEnableRPCReplication
 	// WorkerIndexerConcurrency is the max concurrent messages to be processed at any given time
 	WorkerIndexerConcurrency
 	// WorkerESProcessorNumOfWorkers is num of workers for esProcessor
@@ -718,6 +726,10 @@ const (
 	ReplicationTaskProcessorCleanupInterval
 	// ReplicationTaskProcessorCleanupJitterCoefficient is the jitter for cleanup timer
 	ReplicationTaskProcessorCleanupJitterCoefficient
+	// HistoryEnableRPCReplication is the feature flag for RPC replication
+	HistoryEnableRPCReplication
+	// HistoryEnableKafkaReplication is the migration flag for Kafka replication
+	HistoryEnableKafkaReplication
 	// EnableConsistentQuery indicates if consistent query is enabled for the cluster
 	EnableConsistentQuery
 	// EnableConsistentQueryByDomain indicates if consistent query is enabled for a domain

host/dynamicconfig.go

@@ -44,6 +44,10 @@ var (
 		dynamicconfig.ReplicationTaskFetcherErrorRetryWait:          50 * time.Millisecond,
 		dynamicconfig.ReplicationTaskProcessorErrorRetryWait:        time.Millisecond,
 		dynamicconfig.EnableConsistentQueryByDomain:                 true,
+		dynamicconfig.FrontendEnableRPCReplication:                  true,
+		dynamicconfig.HistoryEnableRPCReplication:                   true,
+		dynamicconfig.HistoryEnableKafkaReplication:                 false,
+		dynamicconfig.WorkerEnableRPCReplication:                    true,
 	}
 )
 

service/frontend/service.go

@@ -93,6 +93,8 @@ type Config struct {
 	VisibilityArchivalQueryMaxPageSize dynamicconfig.IntPropertyFn
 
 	SendRawWorkflowHistory dynamicconfig.BoolPropertyFnWithDomainFilter
+
+	EnableRPCReplication dynamicconfig.BoolPropertyFn
 }
 
 // NewConfig returns new service config with default values
@@ -132,6 +134,7 @@ func NewConfig(dc *dynamicconfig.Collection, numHistoryShards int, enableReadFro
 		VisibilityArchivalQueryMaxPageSize:  dc.GetIntProperty(dynamicconfig.VisibilityArchivalQueryMaxPageSize, 10000),
 		DisallowQuery:                       dc.GetBoolPropertyFnWithDomainFilter(dynamicconfig.DisallowQuery, false),
 		SendRawWorkflowHistory:              dc.GetBoolPropertyFnWithDomainFilter(dynamicconfig.SendRawWorkflowHistory, false),
+		EnableRPCReplication:                dc.GetBoolProperty(dynamicconfig.FrontendEnableRPCReplication, false),
 	}
 }
 
@@ -222,7 +225,9 @@ func (s *Service) Start() {
 	clusterMetadata := s.GetClusterMetadata()
 	if clusterMetadata.IsGlobalDomainEnabled() {
 		consumerConfig := clusterMetadata.GetReplicationConsumerConfig()
-		if consumerConfig != nil && consumerConfig.Type == config.ReplicationConsumerTypeRPC {
+		if consumerConfig != nil &&
+			consumerConfig.Type == config.ReplicationConsumerTypeRPC &&
+			s.config.EnableRPCReplication() {
 			replicationMessageSink = s.GetDomainReplicationQueue()
 		} else {
 			var err error

service/history/config/config.go

@@ -203,6 +203,9 @@ type Config struct {
 	ReplicationTaskProcessorNoTaskRetryWait          dynamicconfig.DurationPropertyFn
 	ReplicationTaskProcessorCleanupInterval          dynamicconfig.DurationPropertyFn
 	ReplicationTaskProcessorCleanupJitterCoefficient dynamicconfig.FloatPropertyFn
+	// TODO: those two flags are for migration. Consider remove them after the migration complete
+	EnableRPCReplication   dynamicconfig.BoolPropertyFn
+	EnableKafkaReplication dynamicconfig.BoolPropertyFn
 
 	// The following are used by consistent query
 	EnableConsistentQuery         dynamicconfig.BoolPropertyFn
@@ -366,6 +369,8 @@ func New(dc *dynamicconfig.Collection, numberOfShards int, storeType string, isA
 		ReplicationTaskProcessorNoTaskRetryWait:          dc.GetDurationProperty(dynamicconfig.ReplicationTaskProcessorNoTaskInitialWait, 2*time.Second),
 		ReplicationTaskProcessorCleanupInterval:          dc.GetDurationProperty(dynamicconfig.ReplicationTaskProcessorCleanupInterval, 1*time.Minute),
 		ReplicationTaskProcessorCleanupJitterCoefficient: dc.GetFloat64Property(dynamicconfig.ReplicationTaskProcessorCleanupJitterCoefficient, 0.15),
+		EnableRPCReplication:                             dc.GetBoolProperty(dynamicconfig.HistoryEnableRPCReplication, false),
+		EnableKafkaReplication:                           dc.GetBoolProperty(dynamicconfig.HistoryEnableKafkaReplication, true),
 
 		EnableConsistentQuery:                 dc.GetBoolProperty(dynamicconfig.EnableConsistentQuery, true),
 		EnableConsistentQueryByDomain:         dc.GetBoolPropertyFnWithDomainFilter(dynamicconfig.EnableConsistentQueryByDomain, false),

service/history/historyEngine.go

@@ -309,7 +309,9 @@ func (e *historyEngineImpl) Start() {
 	e.timerProcessor.Start()
 
 	clusterMetadata := e.shard.GetClusterMetadata()
-	if e.replicatorProcessor != nil && clusterMetadata.GetReplicationConsumerConfig().Type != sconfig.ReplicationConsumerTypeRPC {
+	if e.replicatorProcessor != nil &&
+		clusterMetadata.GetReplicationConsumerConfig().Type != sconfig.ReplicationConsumerTypeRPC &&
+		e.config.EnableKafkaReplication() {
 		e.replicatorProcessor.Start()
 	}
 

service/history/replication/task_fetcher.go

@@ -93,7 +93,7 @@ func NewTaskFetchers(
 ) TaskFetchers {
 
 	var fetchers []TaskFetcher
-	if consumerConfig.Type == serviceConfig.ReplicationConsumerTypeRPC {
+	if consumerConfig.Type == serviceConfig.ReplicationConsumerTypeRPC && config.EnableRPCReplication() {
 		for clusterName, info := range clusterMetadata.GetAllClusterInfo() {
 			if !info.Enabled {
 				continue

service/worker/replicator/replicator.go

@@ -76,6 +76,7 @@ type (
 		ReplicationTaskMaxRetryDuration    dynamicconfig.DurationPropertyFn
 		ReplicationTaskContextTimeout      dynamicconfig.DurationPropertyFn
 		ReReplicationContextTimeout        dynamicconfig.DurationPropertyFnWithDomainIDFilter
+		EnableRPCReplication               dynamicconfig.BoolPropertyFn
 	}
 )
 
@@ -123,7 +124,7 @@ func (r *Replicator) Start() error {
 		}
 
 		if clusterName != currentClusterName {
-			if replicationConsumerConfig.Type == config.ReplicationConsumerTypeRPC {
+			if replicationConsumerConfig.Type == config.ReplicationConsumerTypeRPC && r.config.EnableRPCReplication() {
 				processor := newDomainReplicationMessageProcessor(
 					clusterName,
 					r.logger.WithTags(tag.ComponentReplicationTaskProcessor, tag.SourceCluster(clusterName)),

service/worker/service.go

@@ -119,6 +119,7 @@ func NewConfig(params *service.BootstrapParams) *Config {
 			ReplicationTaskMaxRetryDuration:    dc.GetDurationProperty(dynamicconfig.WorkerReplicationTaskMaxRetryDuration, 15*time.Minute),
 			ReplicationTaskContextTimeout:      dc.GetDurationProperty(dynamicconfig.WorkerReplicationTaskContextDuration, 30*time.Second),
 			ReReplicationContextTimeout:        dc.GetDurationPropertyFilteredByDomainID(dynamicconfig.WorkerReReplicationContextTimeout, 0*time.Second),
+			EnableRPCReplication:               dc.GetBoolProperty(dynamicconfig.WorkerEnableRPCReplication, false),
 		},
 		ArchiverConfig: &archiver.Config{
 			ArchiverConcurrency:           dc.GetIntProperty(dynamicconfig.WorkerArchiverConcurrency, 50),