Add flag for replication cleanup process (#3241)

common/persistence/domainReplicationQueue.go

@@ -101,7 +101,7 @@ func (q *domainReplicationQueueImpl) Start() {
 }
 
 func (q *domainReplicationQueueImpl) Stop() {
-	if !atomic.CompareAndSwapInt32(&q.status, common.DaemonStatusInitialized, common.DaemonStatusStarted) {
+	if !atomic.CompareAndSwapInt32(&q.status, common.DaemonStatusInitialized, common.DaemonStatusStopped) {
 		return
 	}
 	close(q.done)

common/service/dynamicconfig/constants.go

@@ -100,6 +100,9 @@ var keys = map[Key]string{
 	FrontendThrottledLogRPS:               "frontend.throttledLogRPS",
 	EnableClientVersionCheck:              "frontend.enableClientVersionCheck",
 	ValidSearchAttributes:                 "frontend.validSearchAttributes",
+	SendRawWorkflowHistory:                "frontend.sendRawWorkflowHistory",
+	FrontendEnableRPCReplication:          "frontend.enableRPCReplication",
+	FrontendEnableCleanupReplicationTask:  "frontend.enableCleanupReplicationTask",
 	SearchAttributesNumberOfKeysLimit:     "frontend.searchAttributesNumberOfKeysLimit",
 	SearchAttributesSizeOfValueLimit:      "frontend.searchAttributesSizeOfValueLimit",
 	SearchAttributesTotalSizeLimit:        "frontend.searchAttributesTotalSizeLimit",
@@ -225,7 +228,7 @@ var keys = map[Key]string{
 	MutableStateChecksumInvalidateBefore:                  "history.mutableStateChecksumInvalidateBefore",
 	ReplicationEventsFromCurrentCluster:                   "history.ReplicationEventsFromCurrentCluster",
 	EnableDropStuckTaskByDomainID:                         "history.DropStuckTaskByDomain",
-	SkipReapplicationByDomainId:                             "history.SkipReapplicationByDomainId",
+	SkipReapplicationByDomainId:                           "history.SkipReapplicationByDomainId",
 
 	WorkerPersistenceMaxQPS:                         "worker.persistenceMaxQPS",
 	WorkerPersistenceGlobalMaxQPS:                   "worker.persistenceGlobalMaxQPS",
@@ -368,6 +371,12 @@ const (
 	FrontendMaxBadBinaries
 	// ValidSearchAttributes is legal indexed keys that can be used in list APIs
 	ValidSearchAttributes
+	// SendRawWorkflowHistory is whether to enable raw history retrieving
+	SendRawWorkflowHistory
+	// FrontendEnableRPCReplication is a feature flag for rpc replication
+	FrontendEnableRPCReplication
+	// FrontendEnableCleanupReplicationTask is a feature flag for rpc replication cleanup
+	FrontendEnableCleanupReplicationTask
 	// SearchAttributesNumberOfKeysLimit is the limit of number of keys
 	SearchAttributesNumberOfKeysLimit
 	// SearchAttributesSizeOfValueLimit is the size limit of each value
@@ -677,6 +686,12 @@ 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
+	// HistoryEnableCleanupReplicationTask is the migration flag for Kafka replication
+	HistoryEnableCleanupReplicationTask
 	// EnableConsistentQuery indicates if consistent query is enabled for the cluster
 	EnableConsistentQuery
 	// EnableConsistentQueryByDomain indicates if consistent query is enabled for a domain

service/frontend/adminHandler.go

@@ -114,7 +114,9 @@ func (adh *AdminHandler) RegisterHandler() {
 // Start starts the handler
 func (adh *AdminHandler) Start() {
 	// Start domain replication queue cleanup
-	adh.Resource.GetDomainReplicationQueue().Start()
+	if adh.config.EnableCleanupReplicationTask() {
+		adh.Resource.GetDomainReplicationQueue().Start()
+	}
 }
 
 // Stop stops the handler

service/frontend/adminHandler_test.go

@@ -92,7 +92,8 @@ func (s *adminHandlerSuite) SetupTest() {
 		},
 	}
 	config := &Config{
-		EnableAdminProtection: dynamicconfig.GetBoolPropertyFn(false),
+		EnableAdminProtection:        dynamicconfig.GetBoolPropertyFn(false),
+		EnableCleanupReplicationTask: dynamicconfig.GetBoolPropertyFn(false),
 	}
 	s.handler = NewAdminHandler(s.mockResource, params, config)
 	s.handler.Start()
@@ -552,8 +553,9 @@ func (s *adminHandlerSuite) Test_AddSearchAttribute_Permission() {
 	ctx := context.Background()
 	handler := s.handler
 	handler.config = &Config{
-		EnableAdminProtection: dynamicconfig.GetBoolPropertyFn(true),
-		AdminOperationToken:   dynamicconfig.GetStringPropertyFn(common.DefaultAdminOperationToken),
+		EnableAdminProtection:        dynamicconfig.GetBoolPropertyFn(true),
+		AdminOperationToken:          dynamicconfig.GetStringPropertyFn(common.DefaultAdminOperationToken),
+		EnableCleanupReplicationTask: dynamicconfig.GetBoolPropertyFn(false),
 	}
 
 	type test struct {

service/frontend/service.go

@@ -91,6 +91,11 @@ type Config struct {
 
 	// VisibilityArchival system protection
 	VisibilityArchivalQueryMaxPageSize dynamicconfig.IntPropertyFn
+
+	SendRawWorkflowHistory dynamicconfig.BoolPropertyFnWithDomainFilter
+
+	EnableRPCReplication         dynamicconfig.BoolPropertyFn
+	EnableCleanupReplicationTask dynamicconfig.BoolPropertyFn
 }
 
 // NewConfig returns new service config with default values
@@ -129,6 +134,9 @@ func NewConfig(dc *dynamicconfig.Collection, numHistoryShards int, enableReadFro
 		MinRetentionDays:                    dc.GetIntProperty(dynamicconfig.MinRetentionDays, domain.MinRetentionDays),
 		VisibilityArchivalQueryMaxPageSize:  dc.GetIntProperty(dynamicconfig.VisibilityArchivalQueryMaxPageSize, 10000),
 		DisallowQuery:                       dc.GetBoolPropertyFnWithDomainFilter(dynamicconfig.DisallowQuery, false),
+		SendRawWorkflowHistory:              dc.GetBoolPropertyFnWithDomainFilter(dynamicconfig.SendRawWorkflowHistory, false),
+		EnableRPCReplication:                dc.GetBoolProperty(dynamicconfig.FrontendEnableRPCReplication, false),
+		EnableCleanupReplicationTask:        dc.GetBoolProperty(dynamicconfig.FrontendEnableCleanupReplicationTask, true),
 	}
 }
 

service/history/replicationTaskProcessor.go

@@ -207,10 +207,12 @@ func (p *ReplicationTaskProcessorImpl) cleanupReplicationTaskLoop() {
 			timer.Stop()
 			return
 		case <-timer.C:
-			err := p.cleanupAckedReplicationTasks()
-			if err != nil {
-				p.logger.Error("Failed to clean up replication messages.", tag.Error(err))
-				p.metricsClient.Scope(metrics.ReplicationTaskCleanupScope).IncCounter(metrics.ReplicationTaskCleanupFailure)
+			if p.config.EnableCleanupReplicationTask() {
+				err := p.cleanupAckedReplicationTasks()
+				if err != nil {
+					p.logger.Error("Failed to clean up replication messages.", tag.Error(err))
+					p.metricsClient.Scope(metrics.ReplicationTaskCleanupScope).IncCounter(metrics.ReplicationTaskCleanupFailure)
+				}
 			}
 			timer.Reset(backoff.JitDuration(
 				p.config.ShardSyncMinInterval(),

service/history/service.go

@@ -193,6 +193,8 @@ type Config struct {
 	ReplicationTaskProcessorCleanupInterval          dynamicconfig.DurationPropertyFn
 	ReplicationTaskProcessorCleanupJitterCoefficient dynamicconfig.FloatPropertyFn
 
+	EnableCleanupReplicationTask dynamicconfig.BoolPropertyFn
+
 	// The following are used by consistent query
 	EnableConsistentQuery         dynamicconfig.BoolPropertyFn
 	EnableConsistentQueryByDomain dynamicconfig.BoolPropertyFnWithDomainFilter
@@ -207,7 +209,7 @@ type Config struct {
 	ReplicationEventsFromCurrentCluster dynamicconfig.BoolPropertyFnWithDomainFilter
 
 	EnableDropStuckTaskByDomainID dynamicconfig.BoolPropertyFnWithDomainIDFilter
-	SkipReapplicationByDomainId dynamicconfig.BoolPropertyFnWithDomainIDFilter
+	SkipReapplicationByDomainId   dynamicconfig.BoolPropertyFnWithDomainIDFilter
 }
 
 const (
@@ -331,6 +333,7 @@ func NewConfig(dc *dynamicconfig.Collection, numberOfShards int, storeType strin
 		ReplicationTaskProcessorNoTaskRetryWait:          dc.GetDurationProperty(dynamicconfig.ReplicationTaskProcessorNoTaskInitialWait, 2*time.Second),
 		ReplicationTaskProcessorCleanupInterval:          dc.GetDurationProperty(dynamicconfig.ReplicationTaskProcessorCleanupInterval, 1*time.Minute),
 		ReplicationTaskProcessorCleanupJitterCoefficient: dc.GetFloat64Property(dynamicconfig.ReplicationTaskProcessorCleanupJitterCoefficient, 0.15),
+		EnableCleanupReplicationTask:                     dc.GetBoolProperty(dynamicconfig.HistoryEnableCleanupReplicationTask, true),
 
 		EnableConsistentQuery:                 dc.GetBoolProperty(dynamicconfig.EnableConsistentQuery, true),
 		EnableConsistentQueryByDomain:         dc.GetBoolPropertyFnWithDomainFilter(dynamicconfig.EnableConsistentQueryByDomain, false),
@@ -342,7 +345,7 @@ func NewConfig(dc *dynamicconfig.Collection, numberOfShards int, storeType strin
 		ReplicationEventsFromCurrentCluster: dc.GetBoolPropertyFnWithDomainFilter(dynamicconfig.ReplicationEventsFromCurrentCluster, false),
 
 		EnableDropStuckTaskByDomainID: dc.GetBoolPropertyFnWithDomainIDFilter(dynamicconfig.EnableDropStuckTaskByDomainID, false),
-		SkipReapplicationByDomainId: dc.GetBoolPropertyFnWithDomainIDFilter(dynamicconfig.SkipReapplicationByDomainId, false),
+		SkipReapplicationByDomainId:   dc.GetBoolPropertyFnWithDomainIDFilter(dynamicconfig.SkipReapplicationByDomainId, false),
 	}
 
 	return cfg