cdc: region worker should handle stale events correctly (#9078)

cdc/kv/client.go

@@ -436,7 +436,9 @@ func currentRequestID() uint64 {
 }
 
 type eventFeedSession struct {
-	client *CDCClient
+	client     *CDCClient
+	startTs    model.Ts
+	changefeed model.ChangeFeedID
 
 	lockResolver txnutil.LockResolver
 	isPullerInit PullerInitialization
@@ -491,7 +493,10 @@ func newEventFeedSession(
 		totalSpan.Start, totalSpan.End, startTs,
 		client.changefeed.Namespace+"-"+client.changefeed.ID)
 	return &eventFeedSession{
-		client:            client,
+		client:     client,
+		startTs:    startTs,
+		changefeed: client.changefeed,
+
 		totalSpan:         totalSpan,
 		eventCh:           eventCh,
 		regionRouter:      NewSizedRegionRouter(ctx, client.config.RegionScanLimit),

cdc/kv/client_test.go

@@ -2173,56 +2173,6 @@ func testEventCommitTsFallback(t *testing.T, events []*cdcpb.ChangeDataEvent) {
 	cancel()
 }
 
-// TestCommittedFallback tests kv client should panic when receiving a fallback committed event
-func TestCommittedFallback(t *testing.T) {
-	events := []*cdcpb.ChangeDataEvent{
-		{Events: []*cdcpb.Event{
-			{
-				RegionId:  3,
-				RequestId: currentRequestID(),
-				Event: &cdcpb.Event_Entries_{
-					Entries: &cdcpb.Event_Entries{
-						Entries: []*cdcpb.Event_Row{{
-							Type:     cdcpb.Event_COMMITTED,
-							OpType:   cdcpb.Event_Row_PUT,
-							Key:      []byte("a"),
-							Value:    []byte("committed with commit ts before resolved ts"),
-							StartTs:  92,
-							CommitTs: 98,
-						}},
-					},
-				},
-			},
-		}},
-	}
-	testEventCommitTsFallback(t, events)
-}
-
-// TestCommitFallback tests kv client should panic when receiving a fallback commit event
-func TestCommitFallback(t *testing.T) {
-	events := []*cdcpb.ChangeDataEvent{
-		mockInitializedEvent(3, currentRequestID()),
-		{Events: []*cdcpb.Event{
-			{
-				RegionId:  3,
-				RequestId: currentRequestID(),
-				Event: &cdcpb.Event_Entries_{
-					Entries: &cdcpb.Event_Entries{
-						Entries: []*cdcpb.Event_Row{{
-							Type:     cdcpb.Event_COMMIT,
-							OpType:   cdcpb.Event_Row_PUT,
-							Key:      []byte("a-commit-event-ts-fallback"),
-							StartTs:  92,
-							CommitTs: 98,
-						}},
-					},
-				},
-			},
-		}},
-	}
-	testEventCommitTsFallback(t, events)
-}
-
 // TestDeuplicateRequest tests kv client should panic when meeting a duplicate error
 func TestDuplicateRequest(t *testing.T) {
 	events := []*cdcpb.ChangeDataEvent{

cdc/kv/region_worker.go

@@ -731,16 +731,20 @@ func (w *regionWorker) handleEventEntry(
 			state.matcher.putPrewriteRow(entry)
 		case cdcpb.Event_COMMIT:
 			w.metrics.metricPullEventCommitCounter.Inc()
-			if entry.CommitTs <= state.lastResolvedTs {
+			// NOTE: state.getLastResolvedTs() will never less than session.startTs.
+			resolvedTs := state.getLastResolvedTs()
+			// TiKV can send events with StartTs/CommitTs less than startTs.
+			isStaleEvent := entry.CommitTs <= w.session.startTs
+			if entry.CommitTs <= resolvedTs && !isStaleEvent {
 				logPanic("The CommitTs must be greater than the resolvedTs",
 					zap.String("EventType", "COMMIT"),
 					zap.Uint64("CommitTs", entry.CommitTs),
 					zap.Uint64("resolvedTs", state.lastResolvedTs),
 					zap.Uint64("regionID", regionID))
 				return errUnreachable
 			}
-			ok := state.matcher.matchRow(entry, state.initialized)
-			if !ok {
+
+			if !state.matcher.matchRow(entry, state.initialized) {
 				if !state.initialized {
 					state.matcher.cacheCommitRow(entry)
 					continue
@@ -751,16 +755,17 @@ func (w *regionWorker) handleEventEntry(
 					entry.GetType(), entry.GetOpType())
 			}
 
-			revent, err := assembleRowEvent(regionID, entry)
-			if err != nil {
-				return errors.Trace(err)
-			}
-
-			select {
-			case w.outputCh <- revent:
-				w.metrics.metricSendEventCommitCounter.Inc()
-			case <-ctx.Done():
-				return errors.Trace(ctx.Err())
+			if !isStaleEvent {
+				revent, err := assembleRowEvent(regionID, entry)
+				if err != nil {
+					return errors.Trace(err)
+				}
+				select {
+				case w.outputCh <- revent:
+					w.metrics.metricSendEventCommitCounter.Inc()
+				case <-ctx.Done():
+					return errors.Trace(ctx.Err())
+				}
 			}
 		case cdcpb.Event_ROLLBACK:
 			w.metrics.metricPullEventRollbackCounter.Inc()

cdc/kv/region_worker_test.go

@@ -19,6 +19,7 @@ import (
 	"runtime"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/pingcap/kvproto/pkg/cdcpb"
 	"github.com/pingcap/tiflow/cdc/model"
@@ -306,3 +307,53 @@ func TestRegionWorkerHandleResolvedTs(t *testing.T) {
 	require.Equal(t, 1, len(event.Resolved.Spans))
 	require.Equal(t, uint64(1), event.Resolved.Spans[0].Region)
 }
+
+func TestRegionWorkerHandleEventsBeforeStartTs(t *testing.T) {
+	ctx := context.Background()
+	s := createFakeEventFeedSession(ctx)
+	s.eventCh = make(chan model.RegionFeedEvent, 2)
+	s1 := newRegionFeedState(singleRegionInfo{
+		verID:  tikv.RegionVerID{},
+		ts:     9,
+		rpcCtx: &tikv.RPCContext{},
+	}, 0)
+	s1.start()
+	w := newRegionWorker(model.ChangeFeedID{}, s, "")
+
+	err := w.handleResolvedTs(ctx, &resolvedTsEvent{
+		resolvedTs: 5,
+		regions:    []*regionFeedState{s1},
+	})
+	require.Nil(t, err)
+	require.Equal(t, uint64(9), s1.lastResolvedTs)
+
+	timer := time.NewTimer(time.Second)
+	select {
+	case <-w.rtsUpdateCh:
+		if !timer.Stop() {
+			<-timer.C
+		}
+		require.False(t, true, "should never get a ResolvedTs")
+	case <-timer.C:
+	}
+
+	events := &cdcpb.Event_Entries_{
+		Entries: &cdcpb.Event_Entries{
+			Entries: []*cdcpb.Event_Row{
+				{Type: cdcpb.Event_PREWRITE, StartTs: 7},
+			},
+		},
+	}
+	err = w.handleEventEntry(ctx, events, s1)
+	require.Nil(t, err)
+
+	events = &cdcpb.Event_Entries_{
+		Entries: &cdcpb.Event_Entries{
+			Entries: []*cdcpb.Event_Row{
+				{Type: cdcpb.Event_COMMIT, CommitTs: 8},
+			},
+		},
+	}
+	err = w.handleEventEntry(ctx, events, s1)
+	require.Nil(t, err)
+}