config: Enable data sharing in default event store config

coordinator/controller.go

@@ -1163,6 +1163,8 @@ func (c *Controller) RemoveNode(id node.ID) {
 			observation.drainingObserved,
 			observation.stoppingObserved,
 			observation.remaining,
+			observation.logServiceDispatcherCount,
+			observation.logServiceDispatcherCountObserved,
 		)
 	}
 

coordinator/controller_drain.go

@@ -205,6 +205,8 @@ func (c *Controller) DrainNode(ctx context.Context, target node.ID) (int, error)
 		observation.drainingObserved,
 		observation.stoppingObserved,
 		observation.remaining,
+		observation.logServiceDispatcherCount,
+		observation.logServiceDispatcherCountObserved,
 	)
 
 	if completionObserved {
@@ -225,6 +227,8 @@ 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.Uint32("logServiceDispatcherCount", observation.logServiceDispatcherCount),
+		zap.Bool("logServiceDispatcherCountObserved", observation.logServiceDispatcherCountObserved),
 		zap.Int("remaining", observation.remaining))
 	return ensureDrainRemainingNonZero(observation.remaining), nil
 }
@@ -240,6 +244,8 @@ func (c *Controller) observeRemovedActiveDrainTarget(target node.ID, epoch uint6
 		observation.drainingObserved,
 		observation.stoppingObserved,
 		observation.remaining,
+		observation.logServiceDispatcherCount,
+		observation.logServiceDispatcherCountObserved,
 	)
 	if completionObserved {
 		log.Info("drain completion observed for removed active target",
@@ -259,6 +265,8 @@ 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.Uint32("logServiceDispatcherCount", observation.logServiceDispatcherCount),
+		zap.Bool("logServiceDispatcherCountObserved", observation.logServiceDispatcherCountObserved),
 		zap.Int("remaining", observation.remaining))
 	return ensureDrainRemainingNonZero(observation.remaining)
 }
@@ -310,10 +318,12 @@ 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         uint32
+	logServiceDispatcherCountObserved bool
 }
 
 func (c *Controller) observeDrainNode(target node.ID, epoch uint64) drainNodeObservation {
@@ -332,6 +342,7 @@ func (c *Controller) observeDrainNode(target node.ID, epoch uint64) drainNodeObs
 	)
 
 	_, observation.drainingObserved, observation.stoppingObserved = c.drainController.GetStatus(target)
+	observation.logServiceDispatcherCount, observation.logServiceDispatcherCountObserved = c.drainController.GetLogServiceDispatcherCount(target)
 	observation.nodeState = c.drainController.GetState(target)
 	return observation
 }
@@ -1082,11 +1093,16 @@ func isBestEffortDrainComplete(
 	drainingObserved bool,
 	stoppingObserved bool,
 	remaining int,
+	logServiceDispatcherCount uint32,
+	logServiceDispatcherCountObserved bool,
 ) bool {
 	if nodeState == drain.StateUnknown || !drainingObserved {
 		return false
 	}
-	return stoppingObserved && remaining == 0
+	return stoppingObserved &&
+		remaining == 0 &&
+		logServiceDispatcherCountObserved &&
+		logServiceDispatcherCount == 0
 }
 
 // drainRemainingEstimate uses the larger workload dimension to avoid obvious double counting.

coordinator/controller_drain_test.go

@@ -154,6 +154,39 @@ func TestDrainNodeCompletesAfterCompletionObserved(t *testing.T) {
 	require.Equal(t, epoch, c.drainSession.epoch)
 }
 
+func TestDrainNodeWaitsForLogServiceDispatchers(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)
+
+	// A STOPPING response alone does not contain the log service dispatcher count.
+	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, 1, remaining)
+
+	setTargetStoppingHeartbeat(drainController, target, 2)
+	remaining, err = c.DrainNode(context.Background(), target)
+	require.NoError(t, err)
+	require.Equal(t, 1, 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 +1204,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,11 @@ 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         uint32
+	logServiceDispatcherCountObserved bool
 }
 
 type drainTargetSchedulerGate struct {
@@ -185,6 +190,11 @@ 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 = hb.GetLogServiceDispatcherCount()
+		st.logServiceDispatcherCountObserved = true
+	}
 	c.observeTargetSchedulerAckLocked(nodeID, hb)
 }
 
@@ -415,6 +425,19 @@ 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.
+func (c *Controller) GetLogServiceDispatcherCount(nodeID node.ID) (uint32, bool) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	st, ok := c.nodes[nodeID]
+	if !ok || !st.logServiceDispatcherCountObserved {
+		return 0, false
+	}
+	return st.logServiceDispatcherCount, true
+}
+
 // 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,43 @@ 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,
+	})
+	_, observed := c.GetLogServiceDispatcherCount(target)
+	require.False(t, observed)
+
+	c.ObserveHeartbeat(target, &heartbeatpb.NodeHeartbeat{
+		Liveness:                  heartbeatpb.NodeLiveness_STOPPING,
+		NodeEpoch:                 42,
+		LogServiceDispatcherCount: 2,
+	})
+	count, observed := c.GetLogServiceDispatcherCount(target)
+	require.True(t, observed)
+	require.Equal(t, uint32(2), count)
+
+	c.ObserveHeartbeat(target, &heartbeatpb.NodeHeartbeat{
+		Liveness:  heartbeatpb.NodeLiveness_ALIVE,
+		NodeEpoch: 43,
+	})
+	_, observed = c.GetLogServiceDispatcherCount(target)
+	require.False(t, observed)
+
+	// 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,
+	})
+	_, observed = c.GetLogServiceDispatcherCount(target)
+	require.False(t, observed)
+}
+
 func TestDrainControllerSkipStoppingForNewEpochWithoutDraining(t *testing.T) {
 	mc := messaging.NewMockMessageCenter()
 	c := NewController(mc)

downstreamadapter/eventcollector/event_collector.go

@@ -415,7 +415,7 @@ func (c *EventCollector) groupHeartbeat() map[node.ID]*event.DispatcherHeartbeat
 	group := func(target node.ID, dispatcherID common.DispatcherID, checkpointTs uint64, epoch uint64) {
 		heartbeat, ok := groupedHeartbeats[target]
 		if !ok {
-			heartbeat = event.NewDispatcherHeartbeat()
+			heartbeat = event.NewDispatcherHeartbeatWithVersion(c.eventServiceProtocolVersion(target))
 			groupedHeartbeats[target] = heartbeat
 		}
 		heartbeat.AddDispatcherProgress(dispatcherID, checkpointTs, epoch)
@@ -439,6 +439,17 @@ func (c *EventCollector) groupHeartbeat() map[node.ID]*event.DispatcherHeartbeat
 	return groupedHeartbeats
 }
 
+func (c *EventCollector) eventServiceProtocolVersion(target node.ID) int {
+	if target == c.serverId {
+		return node.CurrentMessagingProtocolVersion
+	}
+	info := c.mc.GetNodeInfo(target)
+	if info != nil && info.MessagingProtocolVersion >= node.CurrentMessagingProtocolVersion {
+		return node.CurrentMessagingProtocolVersion
+	}
+	return node.LegacyMessagingProtocolVersion
+}
+
 func (c *EventCollector) processDSFeedback(ctx context.Context) error {
 	log.Info("Start process feedback from dynamic stream")
 	defer log.Info("Stop process feedback from dynamic stream")
@@ -727,7 +738,8 @@ func (c *EventCollector) newCongestionControlMessages() map[node.ID]*event.Conge
 	// build congestion control messages for each node
 	result := make(map[node.ID]*event.CongestionControl)
 	for nodeID, changefeedDispatchers := range nodeDispatcherMemory {
-		congestionControl := event.NewCongestionControlWithVersion(event.CongestionControlVersion2)
+		protocolVersion := c.eventServiceProtocolVersion(nodeID)
+		congestionControl := event.NewCongestionControlWithVersion(protocolVersion)
 
 		for changefeedID, dispatcherMemory := range changefeedDispatchers {
 			if len(dispatcherMemory) == 0 {
@@ -739,13 +751,18 @@ func (c *EventCollector) newCongestionControlMessages() map[node.ID]*event.Conge
 			if !ok {
 				continue
 			}
-			congestionControl.AddAvailableMemoryWithDispatchersAndUsageAndReleaseCount(
-				changefeedID.ID(),
-				totalAvailable,
-				changefeedUsageRatio[changefeedID],
-				dispatcherMemory,
-				getAndResetMemoryReleaseCount(changefeedID),
-			)
+			if protocolVersion >= event.CongestionControlVersion2 {
+				congestionControl.AddAvailableMemoryWithDispatchersAndUsageAndReleaseCount(
+					changefeedID.ID(),
+					totalAvailable,
+					changefeedUsageRatio[changefeedID],
+					dispatcherMemory,
+					getAndResetMemoryReleaseCount(changefeedID),
+				)
+			} else {
+				congestionControl.AddAvailableMemoryWithDispatchers(
+					changefeedID.ID(), totalAvailable, dispatcherMemory)
+			}
 		}
 
 		if len(congestionControl.GetAvailables()) > 0 {

downstreamadapter/eventcollector/event_collector_test.go

@@ -272,11 +272,8 @@ func TestRemoveLastDispatcher(t *testing.T) {
 }
 
 func TestGroupHeartbeatUsesEpochAndClamp(t *testing.T) {
-	ctx := context.Background()
 	serverInfo := node.NewInfo("127.0.0.1:18300", "")
-	mc := messaging.NewMessageCenter(ctx, serverInfo.ID, config.NewDefaultMessageCenterConfig(serverInfo.AdvertiseAddr), nil)
-	mc.Run(ctx)
-	defer mc.Close()
+	mc := messaging.NewMockMessageCenter()
 	appcontext.SetService(appcontext.MessageCenter, mc)
 
 	c := New(serverInfo.ID)
@@ -320,12 +317,28 @@ func TestGroupHeartbeatUsesEpochAndClamp(t *testing.T) {
 
 	remoteHeartbeat := grouped[remoteID]
 	require.NotNil(t, remoteHeartbeat)
+	require.Equal(t, commonEvent.DispatcherHeartbeatVersion1, remoteHeartbeat.Version)
+	require.Len(t, remoteHeartbeat.DispatcherProgressesLegacy, 1)
+	require.Equal(t, remoteDispatcher.id, remoteHeartbeat.DispatcherProgressesLegacy[0].DispatcherID)
+	require.Equal(t, uint64(210), remoteHeartbeat.DispatcherProgressesLegacy[0].CheckpointTs)
+
+	controlMessages := c.newCongestionControlMessages()
+	require.Equal(t, commonEvent.CongestionControlVersion2, controlMessages[serverInfo.ID].GetVersion())
+	require.Equal(t, commonEvent.CongestionControlVersion1, controlMessages[remoteID].GetVersion())
+
+	mc.OnNodeChanges(map[node.ID]*node.Info{
+		remoteID: {
+			ID:                       remoteID,
+			MessagingProtocolVersion: node.CurrentMessagingProtocolVersion,
+		},
+	})
+	grouped = c.groupHeartbeat()
+	remoteHeartbeat = grouped[remoteID]
 	require.Equal(t, commonEvent.DispatcherHeartbeatVersion2, remoteHeartbeat.Version)
 	require.Len(t, remoteHeartbeat.DispatcherProgresses, 1)
-	require.Equal(t, uint8(commonEvent.DispatcherProgressVersion1), remoteHeartbeat.DispatcherProgresses[0].Version)
-	require.Equal(t, remoteDispatcher.id, remoteHeartbeat.DispatcherProgresses[0].DispatcherID)
-	require.Equal(t, uint64(210), remoteHeartbeat.DispatcherProgresses[0].CheckpointTs)
 	require.Equal(t, uint64(5), remoteHeartbeat.DispatcherProgresses[0].Epoch)
+	controlMessages = c.newCongestionControlMessages()
+	require.Equal(t, commonEvent.CongestionControlVersion2, controlMessages[remoteID].GetVersion())
 }
 
 func TestGroupHeartbeatResetThenHandshake(t *testing.T) {