config: Enable data sharing in default event store config

coordinator/controller_drain.go

@@ -225,6 +225,7 @@ func (c *Controller) DrainNode(ctx context.Context, target node.ID) (int, error)
 		zap.Int("dispatcherCountOnTarget", observation.dispatcherCountOnTarget),
 		zap.Int("targetInflightDrainMoveCount", observation.targetInflightDrainMoveCount),
 		zap.Int("pendingStatusCount", observation.pendingStatusCount),
+		zap.Int("logServiceDispatcherCount", observation.logServiceDispatcherCount),
 		zap.Int("remaining", observation.remaining))
 	return ensureDrainRemainingNonZero(observation.remaining), nil
 }
@@ -259,6 +260,7 @@ func (c *Controller) observeRemovedActiveDrainTarget(target node.ID, epoch uint6
 		zap.Int("dispatcherCountOnTarget", observation.dispatcherCountOnTarget),
 		zap.Int("targetInflightDrainMoveCount", observation.targetInflightDrainMoveCount),
 		zap.Int("pendingStatusCount", observation.pendingStatusCount),
+		zap.Int("logServiceDispatcherCount", observation.logServiceDispatcherCount),
 		zap.Int("remaining", observation.remaining))
 	return ensureDrainRemainingNonZero(observation.remaining)
 }
@@ -310,10 +312,11 @@ type drainNodeObservation struct {
 	// pendingStatusCount is the number of running changefeeds not converged to the active target epoch.
 	pendingStatusCount int
 	// remaining is the max of all workload dimensions used by drain completion gating.
-	remaining        int
-	nodeState        drain.State
-	drainingObserved bool
-	stoppingObserved bool
+	remaining                 int
+	nodeState                 drain.State
+	drainingObserved          bool
+	stoppingObserved          bool
+	logServiceDispatcherCount int
 }
 
 func (c *Controller) observeDrainNode(target node.ID, epoch uint64) drainNodeObservation {
@@ -323,12 +326,14 @@ func (c *Controller) observeDrainNode(target node.ID, epoch uint64) drainNodeObs
 	}
 	observation.dispatcherCountOnTarget, observation.targetInflightDrainMoveCount = c.aggregateDrainTargetProgress(target, epoch)
 	observation.pendingStatusCount = c.collectDrainPendingStatus(target, epoch)
+	observation.logServiceDispatcherCount = c.drainController.GetLogServiceDispatcherCount(target)
 	observation.remaining = drainRemainingEstimate(
 		observation.maintainersOnTarget,
 		observation.inflightOpsInvolvingTarget,
 		observation.dispatcherCountOnTarget,
 		observation.targetInflightDrainMoveCount,
 		observation.pendingStatusCount,
+		observation.logServiceDispatcherCount,
 	)
 
 	_, observation.drainingObserved, observation.stoppingObserved = c.drainController.GetStatus(target)
@@ -1096,13 +1101,15 @@ func drainRemainingEstimate(
 	dispatcherCountOnTarget int,
 	targetInflightDrainMoveCount int,
 	pendingStatusCount int,
+	logServiceDispatcherCount int,
 ) int {
 	return max(
 		maintainersOnTarget,
 		inflightOpsInvolvingTarget,
 		dispatcherCountOnTarget,
 		targetInflightDrainMoveCount,
 		pendingStatusCount,
+		logServiceDispatcherCount,
 	)
 }
 

coordinator/controller_drain_test.go

@@ -154,6 +154,54 @@ func TestDrainNodeCompletesAfterCompletionObserved(t *testing.T) {
 	require.Equal(t, epoch, c.drainSession.epoch)
 }
 
+func TestDrainNodeUsesDefaultLogServiceDispatcherCount(t *testing.T) {
+	c, drainController, target := newDrainTestController(t)
+	setDrainProtocolVersion(c, target, heartbeatpb.CurrentDrainProtocolVersion)
+	cf := addRunningChangefeed(c, "cf1", node.ID("other"), 100)
+
+	remaining, err := c.DrainNode(context.Background(), target)
+	require.NoError(t, err)
+	require.Equal(t, 1, remaining)
+
+	_, epoch, ok := c.getDispatcherDrainTarget()
+	require.True(t, ok)
+	setChangefeedDrainStatus(cf, target, epoch, 0, 0)
+
+	// Old nodes do not report the log service dispatcher count. Its protobuf
+	// default is zero, so the missing field must not block drain completion.
+	drainController.ObserveSetNodeLivenessResponse(target, &heartbeatpb.SetNodeLivenessResponse{
+		Applied:   heartbeatpb.NodeLiveness_STOPPING,
+		NodeEpoch: 1,
+	})
+	remaining, err = c.DrainNode(context.Background(), target)
+	require.NoError(t, err)
+	require.Equal(t, 0, remaining)
+}
+
+func TestDrainNodeWaitsForReportedLogServiceDispatchers(t *testing.T) {
+	c, drainController, target := newDrainTestController(t)
+	setDrainProtocolVersion(c, target, heartbeatpb.CurrentDrainProtocolVersion)
+	cf := addRunningChangefeed(c, "cf1", node.ID("other"), 100)
+
+	remaining, err := c.DrainNode(context.Background(), target)
+	require.NoError(t, err)
+	require.Equal(t, 1, remaining)
+
+	_, epoch, ok := c.getDispatcherDrainTarget()
+	require.True(t, ok)
+	setChangefeedDrainStatus(cf, target, epoch, 0, 0)
+
+	setTargetStoppingHeartbeat(drainController, target, 2)
+	remaining, err = c.DrainNode(context.Background(), target)
+	require.NoError(t, err)
+	require.Equal(t, 2, remaining)
+
+	setTargetStoppingHeartbeat(drainController, target, 0)
+	remaining, err = c.DrainNode(context.Background(), target)
+	require.NoError(t, err)
+	require.Equal(t, 0, remaining)
+}
+
 func TestDrainNodeDispatcherCountBlocksCompletion(t *testing.T) {
 	c, drainController, target := newDrainTestController(t)
 	setDrainProtocolVersion(c, target, heartbeatpb.CurrentDrainProtocolVersion)
@@ -1171,11 +1219,19 @@ func setTargetStoppingObserved(
 	drainController *drain.Controller,
 	target node.ID,
 ) {
-	resp := &heartbeatpb.SetNodeLivenessResponse{
-		Applied:   heartbeatpb.NodeLiveness_STOPPING,
-		NodeEpoch: 1,
-	}
-	drainController.ObserveSetNodeLivenessResponse(target, resp)
+	setTargetStoppingHeartbeat(drainController, target, 0)
+}
+
+func setTargetStoppingHeartbeat(
+	drainController *drain.Controller,
+	target node.ID,
+	logServiceDispatcherCount uint32,
+) {
+	drainController.ObserveHeartbeat(target, &heartbeatpb.NodeHeartbeat{
+		Liveness:                  heartbeatpb.NodeLiveness_STOPPING,
+		NodeEpoch:                 1,
+		LogServiceDispatcherCount: logServiceDispatcherCount,
+	})
 }
 
 func drainMessageChannel(ch chan *messaging.TargetMessage) {

coordinator/drain/controller.go

@@ -64,6 +64,10 @@ type nodeState struct {
 	lastSeen    time.Time
 	nodeEpoch   uint64
 	liveness    heartbeatpb.NodeLiveness
+
+	// logServiceDispatcherCount is valid only after a STOPPING heartbeat has
+	// been observed for the current node epoch.
+	logServiceDispatcherCount int
 }
 
 type drainTargetSchedulerGate struct {
@@ -185,6 +189,10 @@ func (c *Controller) ObserveHeartbeat(nodeID node.ID, hb *heartbeatpb.NodeHeartb
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	c.observeLivenessLocked(nodeID, hb.NodeEpoch, hb.Liveness)
+	st := c.ensureNodeStateLocked(nodeID)
+	if hb.NodeEpoch == st.nodeEpoch && hb.Liveness == heartbeatpb.NodeLiveness_STOPPING {
+		st.logServiceDispatcherCount = int(hb.GetLogServiceDispatcherCount())
+	}
 	c.observeTargetSchedulerAckLocked(nodeID, hb)
 }
 
@@ -415,6 +423,20 @@ func (c *Controller) GetStatus(nodeID node.ID) (drainRequested, drainingObserved
 	return st.drainRequested, st.drainingObserved, st.stoppingObserved
 }
 
+// GetLogServiceDispatcherCount returns the dispatcher count reported by a
+// STOPPING heartbeat for the current node epoch. Nodes that do not report the
+// field use the protobuf default value zero.
+func (c *Controller) GetLogServiceDispatcherCount(nodeID node.ID) int {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	st, ok := c.nodes[nodeID]
+	if !ok {
+		return 0
+	}
+	return st.logServiceDispatcherCount
+}
+
 // GetDrainProtocolVersion returns the bootstrap-observed drain capability for a node.
 func (c *Controller) GetDrainProtocolVersion(nodeID node.ID) (uint32, bool) {
 	c.mu.Lock()

coordinator/drain/controller_test.go

@@ -126,6 +126,38 @@ func TestDrainControllerResetObservedStateForNewEpoch(t *testing.T) {
 	c.mu.Unlock()
 }
 
+func TestDrainControllerTracksStoppingLogServiceDispatcherCountByEpoch(t *testing.T) {
+	c := NewController(messaging.NewMockMessageCenter())
+	target := node.ID("n1")
+
+	c.ObserveSetNodeLivenessResponse(target, &heartbeatpb.SetNodeLivenessResponse{
+		Applied:   heartbeatpb.NodeLiveness_STOPPING,
+		NodeEpoch: 42,
+	})
+	require.Zero(t, c.GetLogServiceDispatcherCount(target))
+
+	c.ObserveHeartbeat(target, &heartbeatpb.NodeHeartbeat{
+		Liveness:                  heartbeatpb.NodeLiveness_STOPPING,
+		NodeEpoch:                 42,
+		LogServiceDispatcherCount: 2,
+	})
+	require.Equal(t, 2, c.GetLogServiceDispatcherCount(target))
+
+	c.ObserveHeartbeat(target, &heartbeatpb.NodeHeartbeat{
+		Liveness:  heartbeatpb.NodeLiveness_ALIVE,
+		NodeEpoch: 43,
+	})
+	require.Zero(t, c.GetLogServiceDispatcherCount(target))
+
+	// A delayed heartbeat from the old process must not satisfy the new epoch.
+	c.ObserveHeartbeat(target, &heartbeatpb.NodeHeartbeat{
+		Liveness:                  heartbeatpb.NodeLiveness_STOPPING,
+		NodeEpoch:                 42,
+		LogServiceDispatcherCount: 0,
+	})
+	require.Zero(t, c.GetLogServiceDispatcherCount(target))
+}
+
 func TestDrainControllerSkipStoppingForNewEpochWithoutDraining(t *testing.T) {
 	mc := messaging.NewMockMessageCenter()
 	c := NewController(mc)