logcoordinator: ignore uninitialized subscriptions for reuse candidates (#3313) (#3337)

* add more test

* fix

* fix

* remove test

---------

Co-authored-by: lidezhu <[email protected]>

Author: ti-chi-bot

logservice/coordinator/coordinator.go

@@ -344,9 +344,14 @@ func (c *logCoordinator) getCandidateNodes(requestNodeID node.ID, span *heartbea
 		var maxResolvedTs, subID uint64
 		found := false
 		for _, subsState := range subStates.GetSubscriptions() {
+			// Only consider subscriptions which meet the following conditions:
+			// 1. subscription's span covers the requested span
+			// 2. subscription's checkpointTs <= request startTs
+			// 3. subscription's checkpointTs < resolvedTs (meaning the subscription has finished incremental scan)
 			if bytes.Compare(subsState.Span.StartKey, span.StartKey) <= 0 &&
 				bytes.Compare(span.EndKey, subsState.Span.EndKey) <= 0 &&
-				subsState.CheckpointTs <= startTs {
+				subsState.CheckpointTs <= startTs &&
+				subsState.CheckpointTs < subsState.ResolvedTs {
 				if !found || subsState.ResolvedTs > maxResolvedTs {
 					maxResolvedTs = subsState.ResolvedTs
 					subID = subsState.SubID

logservice/coordinator/coordinator_test.go

@@ -51,6 +51,7 @@ func TestGetCandidateNodes(t *testing.T) {
 	tableID2 := int64(101)
 	span1 := common.TableIDToComparableSpan(common.DefaultKeyspaceID, tableID1)
 	span2 := common.TableIDToComparableSpan(common.DefaultKeyspaceID, tableID2)
+	startTs := uint64(100)
 
 	// initialize event store states
 	coordinator.updateEventStoreState(nodeID1, &logservicepb.EventStoreState{
@@ -129,22 +130,23 @@ func TestGetCandidateNodes(t *testing.T) {
 			},
 		},
 	})
+	require.Len(t, coordinator.eventStoreStates.m, 3)
 
 	// check get candidates
 	{
-		nodes := coordinator.getCandidateNodes(nodeID1, &span1, uint64(100))
+		nodes := coordinator.getCandidateNodes(nodeID1, &span1, startTs)
 		assert.Equal(t, []string{nodeID2.String()}, nodes)
 	}
 	{
-		nodes := coordinator.getCandidateNodes(nodeID3, &span1, uint64(100))
+		nodes := coordinator.getCandidateNodes(nodeID3, &span1, startTs)
 		assert.Equal(t, []string{nodeID2.String(), nodeID1.String()}, nodes)
 	}
 	{
-		nodes := coordinator.getCandidateNodes(nodeID1, &span2, uint64(100))
+		nodes := coordinator.getCandidateNodes(nodeID1, &span2, startTs)
 		assert.Equal(t, []string{nodeID3.String(), nodeID2.String()}, nodes)
 	}
 	{
-		nodes := coordinator.getCandidateNodes(nodeID3, &span2, uint64(100))
+		nodes := coordinator.getCandidateNodes(nodeID3, &span2, startTs)
 		assert.Equal(t, []string{nodeID2.String()}, nodes)
 	}
 
@@ -164,7 +166,7 @@ func TestGetCandidateNodes(t *testing.T) {
 		},
 	})
 	{
-		nodes := coordinator.getCandidateNodes(nodeID3, &span1, uint64(100))
+		nodes := coordinator.getCandidateNodes(nodeID3, &span1, startTs)
 		assert.Equal(t, []string{nodeID1.String(), nodeID2.String()}, nodes)
 	}
 
@@ -190,18 +192,49 @@ func TestGetCandidateNodes(t *testing.T) {
 		},
 	})
 	{
-		nodes := coordinator.getCandidateNodes(nodeID3, &span1, uint64(100))
+		nodes := coordinator.getCandidateNodes(nodeID3, &span1, startTs)
 		assert.Equal(t, []string{nodeID2.String(), nodeID1.String()}, nodes)
 	}
 
 	// remove node1 and check again
 	delete(coordinator.nodes.m, nodeID1)
 	{
-		nodes := coordinator.getCandidateNodes(nodeID3, &span1, uint64(100))
+		nodes := coordinator.getCandidateNodes(nodeID3, &span1, startTs)
 		assert.Equal(t, []string{nodeID2.String()}, nodes)
 	}
 }
 
+func TestGetCandidateNodesIgnoreResolvedEqCheckpoint(t *testing.T) {
+	coordinator := newLogCoordinatorForTest()
+
+	nodeID1 := node.ID("node-1")
+	nodeID2 := node.ID("node-2")
+	coordinator.nodes.m[nodeID1] = &node.Info{ID: nodeID1}
+	coordinator.nodes.m[nodeID2] = &node.Info{ID: nodeID2}
+
+	tableID := int64(200)
+	span := common.TableIDToComparableSpan(common.DefaultKeyspaceID, tableID)
+	startTs := uint64(600)
+
+	coordinator.updateEventStoreState(nodeID2, &logservicepb.EventStoreState{
+		TableStates: map[int64]*logservicepb.TableState{
+			tableID: {
+				Subscriptions: []*logservicepb.SubscriptionState{
+					{
+						SubID:        1,
+						Span:         &span,
+						CheckpointTs: startTs - 1,
+						ResolvedTs:   startTs - 1,
+					},
+				},
+			},
+		},
+	})
+
+	nodes := coordinator.getCandidateNodes(nodeID1, &span, startTs)
+	require.Empty(t, nodes, "resolvedTs equal to checkpointTs should not be reused")
+}
+
 func TestUpdateChangefeedStates(t *testing.T) {
 	c := newLogCoordinatorForTest()
 

logservice/eventstore/event_store.go

@@ -479,6 +479,11 @@ func (e *eventStore) RegisterDispatcher(
 			if bytes.Compare(subStat.tableSpan.StartKey, dispatcherSpan.StartKey) <= 0 &&
 				bytes.Compare(subStat.tableSpan.EndKey, dispatcherSpan.EndKey) >= 0 {
 
+				// For onlyReuse register request, we only consider initialized subStats
+				if onlyReuse && !subStat.initialized.Load() {
+					continue
+				}
+
 				// Check whether the subStat ts range contains startTs
 				if subStat.checkpointTs.Load() > startTs || startTs > subStat.resolvedTs.Load() {
 					continue

logservice/eventstore/event_store_test.go

@@ -152,6 +152,14 @@ func TestEventStoreInteractionWithSubClient(t *testing.T) {
 	}
 }
 
+func markSubStatsInitializedForTest(store EventStore, tableID int64) {
+	es := store.(*eventStore)
+	subStats := es.dispatcherMeta.tableStats[tableID]
+	for _, subStat := range subStats {
+		subStat.initialized.Store(true)
+	}
+}
+
 func TestEventStoreOnlyReuseDispatcher(t *testing.T) {
 	_, store := newEventStoreForTest(fmt.Sprintf("/tmp/%s", t.Name()))
 
@@ -180,6 +188,18 @@ func TestEventStoreOnlyReuseDispatcher(t *testing.T) {
 		ok := store.RegisterDispatcher(cfID, dispatcherID2, span, 100, func(watermark uint64, latestCommitTs uint64) {}, true, false)
 		require.False(t, ok)
 	}
+	// when the existing subscription is not initialized, add a dispatcher(onlyReuse=true) should fail
+	{
+		span := &heartbeatpb.TableSpan{
+			TableID:  tableID,
+			StartKey: []byte("b"),
+			EndKey:   []byte("h"),
+		}
+		ok := store.RegisterDispatcher(cfID, dispatcherID3, span, 100, func(watermark uint64, latestCommitTs uint64) {}, true, false)
+		require.False(t, ok)
+	}
+	// mark existing subscription as initialized
+	markSubStatsInitializedForTest(store, tableID)
 	// add a dispatcher(onlyReuse=true) with a containing span which should success
 	{
 		span := &heartbeatpb.TableSpan{
@@ -229,6 +249,7 @@ func TestEventStoreOnlyReuseDispatcherSuccess(t *testing.T) {
 		ok := es.RegisterDispatcher(cfID, dispatcherID1, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false)
 		require.True(t, ok)
 	}
+	markSubStatsInitializedForTest(store, tableID)
 
 	// 2. Register a second dispatcher with onlyReuse=true, whose span is contained
 	//    by the first subscription. This registration should succeed.