Replication task generation delay (#3465)

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",
 	HistoryEnableRPCReplication:                           "history.EnableRPCReplication",
 	HistoryEnableKafkaReplication:                         "history.EnableKafkaReplication",
 	HistoryEnableCleanupReplicationTask:                   "history.EnableCleanupReplicationTask",
@@ -853,6 +854,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
 	// HistoryEnableRPCReplication is the feature flag for RPC replication
 	HistoryEnableRPCReplication
 	// HistoryEnableKafkaReplication is the migration flag for Kafka replication

service/history/config/config.go

@@ -232,6 +232,8 @@ type Config struct {
 	ReplicationTaskProcessorStartWaitJitterCoefficient dynamicconfig.FloatPropertyFnWithShardIDFilter
 	ReplicationTaskProcessorHostQPS                    dynamicconfig.FloatPropertyFn
 	ReplicationTaskProcessorShardQPS                   dynamicconfig.FloatPropertyFn
+	ReplicationTaskGenerationQPS                       dynamicconfig.FloatPropertyFn
+
 	// TODO: those two flags are for migration. Consider remove them after the migration complete
 	EnableRPCReplication         dynamicconfig.BoolPropertyFn
 	EnableKafkaReplication       dynamicconfig.BoolPropertyFn
@@ -448,6 +450,7 @@ func New(dc *dynamicconfig.Collection, numberOfShards int, storeType string, isA
 		EnableRPCReplication:                               dc.GetBoolProperty(dynamicconfig.HistoryEnableRPCReplication, false),
 		EnableKafkaReplication:                             dc.GetBoolProperty(dynamicconfig.HistoryEnableKafkaReplication, true),
 		EnableCleanupReplicationTask:                       dc.GetBoolProperty(dynamicconfig.HistoryEnableCleanupReplicationTask, true),
+		ReplicationTaskGenerationQPS:                       dc.GetFloat64Property(dynamicconfig.ReplicationTaskGenerationQPS, 100),
 
 		EnableConsistentQuery:                 dc.GetBoolProperty(dynamicconfig.EnableConsistentQuery, true),
 		EnableConsistentQueryByDomain:         dc.GetBoolPropertyFilteredByDomain(dynamicconfig.EnableConsistentQueryByDomain, false),

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/service/history/execution"
 	"github.com/uber/cadence/service/history/shard"
 	"github.com/uber/cadence/service/history/task"
@@ -62,6 +63,7 @@ type (
 		queueAckMgr
 
 		lastShardSyncTimestamp time.Time
+		rateLimiter            *quotas.DynamicRateLimiter
 	}
 )
 
@@ -141,7 +143,9 @@ func newReplicatorQueueProcessor(
 	)
 	processor.queueAckMgr = queueAckMgr
 	processor.queueProcessorBase = queueProcessorBase
-
+	processor.rateLimiter = quotas.NewDynamicRateLimiter(func() float64 {
+		return config.ReplicationTaskGenerationQPS()
+	})
 	return processor
 }
 
@@ -449,6 +453,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