Replication task generation delay (cadence-workflow#3465)

fengwk · Aug 20, 2020 · 4ac545c · 4ac545c
1 parent 0325ee5
commit 4ac545c
Show file tree

Hide file tree

Showing 4 changed files with 15 additions and 5 deletions.
diff --git a/common/service/dynamicconfig/constants.go b/common/service/dynamicconfig/constants.go
@@ -266,6 +266,7 @@ var keys = map[Key]string{
 	ReplicationTaskProcessorStartWaitJitterCoefficient:    "history.ReplicationTaskProcessorStartWaitJitterCoefficient",
 	ReplicationTaskProcessorHostQPS:                       "history.ReplicationTaskProcessorHostQPS",
 	ReplicationTaskProcessorShardQPS:                      "history.ReplicationTaskProcessorShardQPS",
+	ReplicationTaskGenerationQPS:                          "history.ReplicationTaskGenerationQPS",
 	EnableConsistentQuery:                                 "history.EnableConsistentQuery",
 	EnableConsistentQueryByDomain:                         "history.EnableConsistentQueryByDomain",
 	MaxBufferedQueryCount:                                 "history.MaxBufferedQueryCount",
@@ -868,6 +869,8 @@ const (
 	ReplicationTaskProcessorHostQPS
 	// ReplicationTaskProcessorShardQPS is the qps of task processing rate limiter on shard level
 	ReplicationTaskProcessorShardQPS
+	//ReplicationTaskGenerationQPS is the wait time between each replication task generation qps
+	ReplicationTaskGenerationQPS
 	// EnableConsistentQuery indicates if consistent query is enabled for the cluster
 	EnableConsistentQuery
 	// EnableConsistentQueryByDomain indicates if consistent query is enabled for a domain

diff --git a/service/frontend/adminHandler_test.go b/service/frontend/adminHandler_test.go
@@ -92,8 +92,8 @@ func (s *adminHandlerSuite) SetupTest() {
 		},
 	}
 	config := &Config{
-		EnableAdminProtection:        dynamicconfig.GetBoolPropertyFn(false),
-		EnableGracefulFailover:       dynamicconfig.GetBoolPropertyFn(false),
+		EnableAdminProtection:  dynamicconfig.GetBoolPropertyFn(false),
+		EnableGracefulFailover: dynamicconfig.GetBoolPropertyFn(false),
 	}
 	s.handler = NewAdminHandler(s.mockResource, params, config)
 	s.handler.Start()
@@ -553,8 +553,8 @@ 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),
 	}
 
 	type test struct {

diff --git a/service/history/config/config.go b/service/history/config/config.go
@@ -234,6 +234,7 @@ type Config struct {
 	ReplicationTaskProcessorStartWaitJitterCoefficient dynamicconfig.FloatPropertyFnWithShardIDFilter
 	ReplicationTaskProcessorHostQPS                    dynamicconfig.FloatPropertyFn
 	ReplicationTaskProcessorShardQPS                   dynamicconfig.FloatPropertyFn
+	ReplicationTaskGenerationQPS                       dynamicconfig.FloatPropertyFn
 
 	// The following are used by consistent query
 	EnableConsistentQuery         dynamicconfig.BoolPropertyFn
@@ -445,6 +446,7 @@ func New(dc *dynamicconfig.Collection, numberOfShards int, storeType string, isA
 		ReplicationTaskProcessorStartWaitJitterCoefficient: dc.GetFloat64PropertyFilteredByShardID(dynamicconfig.ReplicationTaskProcessorStartWaitJitterCoefficient, 0.9),
 		ReplicationTaskProcessorHostQPS:                    dc.GetFloat64Property(dynamicconfig.ReplicationTaskProcessorHostQPS, 1500),
 		ReplicationTaskProcessorShardQPS:                   dc.GetFloat64Property(dynamicconfig.ReplicationTaskProcessorShardQPS, 5),
+		ReplicationTaskGenerationQPS:                       dc.GetFloat64Property(dynamicconfig.ReplicationTaskGenerationQPS, 100),
 
 		EnableConsistentQuery:                 dc.GetBoolProperty(dynamicconfig.EnableConsistentQuery, true),
 		EnableConsistentQueryByDomain:         dc.GetBoolPropertyFilteredByDomain(dynamicconfig.EnableConsistentQueryByDomain, false),

diff --git a/service/history/replicatorQueueProcessor.go b/service/history/replicatorQueueProcessor.go
@@ -38,6 +38,7 @@ import (
 	"github.com/uber/cadence/common/messaging"
 	"github.com/uber/cadence/common/metrics"
 	"github.com/uber/cadence/common/persistence"
+	"github.com/uber/cadence/common/quotas"
 	"github.com/uber/cadence/common/service/dynamicconfig"
 	"github.com/uber/cadence/service/history/execution"
 	"github.com/uber/cadence/service/history/shard"
@@ -63,6 +64,7 @@ type (
 		queueAckMgr
 
 		lastShardSyncTimestamp time.Time
+		rateLimiter            *quotas.DynamicRateLimiter
 	}
 )
 
@@ -142,7 +144,9 @@ func newReplicatorQueueProcessor(
 	)
 	processor.queueAckMgr = queueAckMgr
 	processor.queueProcessorBase = queueProcessorBase
-
+	processor.rateLimiter = quotas.NewDynamicRateLimiter(func() float64 {
+		return config.ReplicationTaskGenerationQPS()
+	})
 	return processor
 }
 
@@ -451,6 +455,7 @@ func (p *replicatorQueueProcessorImpl) getTasks(
 	var replicationTasks []*replicator.ReplicationTask
 	readLevel := lastReadTaskID
 	for _, taskInfo := range taskInfoList {
+		_ = p.rateLimiter.Wait(ctx)
 		var replicationTask *replicator.ReplicationTask
 		op := func() error {
 			var err error