sinkv2(ticdc): address comments

cdc/sinkv2/eventsink/event.go

@@ -32,7 +32,7 @@ type CallbackFunc func()
 type CallbackableEvent[E TableEvent] struct {
 	Event       E
 	Callback    CallbackFunc
-	TableStatus pipeline.TableState
+	TableStatus *pipeline.TableState
 }
 
 // RowChangeCallbackableEvent is the row change event which can be callbacked.

cdc/sinkv2/tablesink/event_table_sink.go

@@ -87,7 +87,7 @@ func (e *eventTableSink[E]) UpdateResolvedTs(resolvedTs model.ResolvedTs) {
 			Callback: func() {
 				e.progressTracker.remove(eventID)
 			},
-			TableStatus: e.state,
+			TableStatus: &e.state,
 		}
 		resolvedCallbackableEvents = append(resolvedCallbackableEvents, ce)
 		e.progressTracker.addEvent(eventID)

cdc/sinkv2/tablesink/event_table_sink_test.go

@@ -48,18 +48,23 @@ func (m *mockEventSink) acknowledge(commitTs uint64) {
 		return
 	}
 	ackedEvents := m.events[:i]
-	m.events = append(
-		make([]*eventsink.TxnCallbackableEvent,
-			0,
-			len(m.events[i:])),
-		m.events[i:]...,
-	)
 
 	for _, event := range ackedEvents {
 		if event.TableStatus.Load() != pipeline.TableStateStopped {
 			event.Callback()
+		} else {
+			// If the table is stopped, the event should be ignored.
+			return
 		}
 	}
+
+	// Remove the acked events from the event buffer.
+	m.events = append(
+		make([]*eventsink.TxnCallbackableEvent,
+			0,
+			len(m.events[i:])),
+		m.events[i:]...,
+	)
 }
 
 func getTestRows() []*model.RowChangedEvent {
@@ -246,7 +251,10 @@ func TestClose(t *testing.T) {
 	tb := New[*model.SingleTableTxn](sink, &eventsink.TxnEventAppender{})
 
 	tb.AppendRowChangedEvents(getTestRows()...)
-	tb.UpdateResolvedTs(model.NewResolvedTs(100))
+	tb.UpdateResolvedTs(model.NewResolvedTs(105))
+	require.Len(t, sink.events, 7, "all events should be flushed")
 	tb.Close()
 	require.Equal(t, pipeline.TableStateStopped, tb.state, "tableState should be closed")
+	sink.acknowledge(105)
+	require.Len(t, sink.events, 7, "no event should be acked")
 }