enable shard linger with time limit config (#4696)

common/dynamicconfig/constants.go

@@ -503,15 +503,14 @@ const (
 	AcquireShardInterval = "history.acquireShardInterval"
 	// AcquireShardConcurrency is number of goroutines that can be used to acquire shards in the shard controller.
 	AcquireShardConcurrency = "history.acquireShardConcurrency"
-	// ShardLingerEnabled configures if the shard controller will temporarily
-	// delay closing shards after a membership update, awaiting a shard
-	// ownership lost error from persistence. Not recommended with persistence
-	// layers that are missing AssertShardOwnership support.
-	ShardLingerEnabled = "history.shardLingerEnabled"
 	// ShardLingerOwnershipCheckQPS is the frequency to perform shard ownership
 	// checks while a shard is lingering.
 	ShardLingerOwnershipCheckQPS = "history.shardLingerOwnershipCheckQPS"
-	// ShardLingerTimeLimit is the upper bound on how long a shard can linger.
+	// ShardLingerTimeLimit configures if and for how long the shard controller
+	// will temporarily delay closing shards after a membership update, awaiting a
+	// shard ownership lost error from persistence. Not recommended with
+	// persistence layers that are missing AssertShardOwnership support.
+	// If set to zero, shards will not delay closing.
 	ShardLingerTimeLimit = "history.shardLingerTimeLimit"
 	// HistoryClientOwnershipCachingEnabled configures if history clients try to cache
 	// shard ownership information, instead of checking membership for each request.

service/history/configs/config.go

@@ -82,7 +82,6 @@ type Config struct {
 	RangeSizeBits                uint
 	AcquireShardInterval         dynamicconfig.DurationPropertyFn
 	AcquireShardConcurrency      dynamicconfig.IntPropertyFn
-	ShardLingerEnabled           dynamicconfig.BoolPropertyFn
 	ShardLingerOwnershipCheckQPS dynamicconfig.IntPropertyFn
 	ShardLingerTimeLimit         dynamicconfig.DurationPropertyFn
 
@@ -367,9 +366,8 @@ func NewConfig(
 		RangeSizeBits:                20, // 20 bits for sequencer, 2^20 sequence number for any range
 		AcquireShardInterval:         dc.GetDurationProperty(dynamicconfig.AcquireShardInterval, time.Minute),
 		AcquireShardConcurrency:      dc.GetIntProperty(dynamicconfig.AcquireShardConcurrency, 10),
-		ShardLingerEnabled:           dc.GetBoolProperty(dynamicconfig.ShardLingerEnabled, false),
 		ShardLingerOwnershipCheckQPS: dc.GetIntProperty(dynamicconfig.ShardLingerOwnershipCheckQPS, 4),
-		ShardLingerTimeLimit:         dc.GetDurationProperty(dynamicconfig.ShardLingerTimeLimit, 3*time.Second),
+		ShardLingerTimeLimit:         dc.GetDurationProperty(dynamicconfig.ShardLingerTimeLimit, 0),
 
 		HistoryClientOwnershipCachingEnabled: dc.GetBoolProperty(dynamicconfig.HistoryClientOwnershipCachingEnabled, false),
 

service/history/shard/controller_impl.go

@@ -49,6 +49,10 @@ import (
 	"go.temporal.io/server/service/history/configs"
 )
 
+const (
+	shardLingerMaxTimeLimit = 5 * time.Second
+)
+
 var (
 	invalidShardIdLowerBound = serviceerror.NewInvalidArgument("shard Id cannot be equal or lower than zero")
 	invalidShardIdUpperBound = serviceerror.NewInvalidArgument("shard Id cannot be larger than max shard count")
@@ -346,8 +350,10 @@ func (c *ControllerImpl) endLinger(shard ControllableContext) {
 
 func (c *ControllerImpl) doLinger(ctx context.Context, shard ControllableContext) {
 	startTime := time.Now()
-	timeout := util.Min(c.config.ShardLingerTimeLimit(), shardIOTimeout)
-	ctx, cancel := context.WithTimeout(ctx, timeout)
+	// Enforce a max limit to ensure we close the shard in a reasonable time,
+	// and to indirectly limit the number of lingering shards.
+	timeLimit := util.Min(c.config.ShardLingerTimeLimit(), shardLingerMaxTimeLimit)
+	ctx, cancel := context.WithTimeout(ctx, timeLimit)
 	defer cancel()
 
 	qps := c.config.ShardLingerOwnershipCheckQPS()
@@ -391,7 +397,7 @@ func (c *ControllerImpl) acquireShards(ctx context.Context) {
 		if err := c.ownership.verifyOwnership(shardID); err != nil {
 			if IsShardOwnershipLostError(err) {
 				// current host is not owner of shard, unload it if it is already loaded.
-				if c.config.ShardLingerEnabled() {
+				if c.config.ShardLingerTimeLimit() > 0 {
 					c.shardLingerThenClose(ctx, shardID)
 				} else {
 					c.CloseShardByID(shardID)

service/history/shard/controller_test.go

@@ -45,6 +45,7 @@ import (
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/cluster"
 	"go.temporal.io/server/common/convert"
+	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/log/tag"
 	"go.temporal.io/server/common/membership"
@@ -741,13 +742,8 @@ func (s *controllerSuite) Test_GetOrCreateShard_InvalidShardID() {
 func (s *controllerSuite) TestShardLingerTimeout() {
 	shardID := int32(1)
 	s.config.NumberOfShards = 1
-	s.config.ShardLingerEnabled = func() bool {
-		return true
-	}
 	timeLimit := 1 * time.Second
-	s.config.ShardLingerTimeLimit = func() time.Duration {
-		return timeLimit
-	}
+	s.config.ShardLingerTimeLimit = dynamicconfig.GetDurationPropertyFn(timeLimit)
 
 	historyEngines := make(map[int32]*MockEngine)
 	mockEngine := NewMockEngine(s.controller)
@@ -789,17 +785,11 @@ func (s *controllerSuite) TestShardLingerTimeout() {
 func (s *controllerSuite) TestShardLingerSuccess() {
 	shardID := int32(1)
 	s.config.NumberOfShards = 1
-	s.config.ShardLingerEnabled = func() bool {
-		return true
-	}
 	timeLimit := 1 * time.Second
+	s.config.ShardLingerTimeLimit = dynamicconfig.GetDurationPropertyFn(timeLimit)
+
 	checkQPS := 5
-	s.config.ShardLingerTimeLimit = func() time.Duration {
-		return timeLimit
-	}
-	s.config.ShardLingerOwnershipCheckQPS = func() int {
-		return checkQPS
-	}
+	s.config.ShardLingerOwnershipCheckQPS = dynamicconfig.GetIntPropertyFn(checkQPS)
 
 	historyEngines := make(map[int32]*MockEngine)
 	mockEngine := NewMockEngine(s.controller)