dispatcher: limit resend message count

Signed-off-by: dongmen <414110582@qq.com>

Author: asddongmen

downstreamadapter/dispatcher/basic_dispatcher.go

@@ -35,6 +35,11 @@ import (
 	"go.uber.org/zap/zapcore"
 )
 
+const (
+	outdatedActionDoneReportInterval   = time.Second
+	outdatedActionDoneReportMaxEntries = 256
+)
+
 // DispatcherService defines the interface for providing dispatcher information and basic event handling.
 type DispatcherService interface {
 	GetId() common.DispatcherID
@@ -217,6 +222,11 @@ type BasicDispatcher struct {
 
 	BootstrapState bootstrapState
 
+	// outdatedActionDoneAt throttles DONE reports for the same outdated action to avoid
+	// flooding blockStatusesChan when maintainer resends stale actions repeatedly.
+	outdatedActionDoneMu sync.Mutex
+	outdatedActionDoneAt map[BlockEventIdentifier]time.Time
+
 	// tableModeCompatibilityChecked indicates whether we have already validated the newest
 	// table schema is compatible with the current replication mode configuration.
 	// Only when the initial case or a ddl event is received, we will reset tableModeCompatibilityChecked to check the compatibility.
@@ -256,12 +266,39 @@ func NewBasicDispatcher(
 		creationPDTs:           currentPDTs,
 		mode:                   mode,
 		BootstrapState:         BootstrapFinished,
+		outdatedActionDoneAt:   make(map[BlockEventIdentifier]time.Time),
 	}
 	dispatcher.resolvedTs.Store(startTs)
 
 	return dispatcher
 }
 
+func (d *BasicDispatcher) shouldReportOutdatedActionDone(identifier BlockEventIdentifier, now time.Time) bool {
+	d.outdatedActionDoneMu.Lock()
+	defer d.outdatedActionDoneMu.Unlock()
+
+	if last, ok := d.outdatedActionDoneAt[identifier]; ok && now.Sub(last) < outdatedActionDoneReportInterval {
+		return false
+	}
+
+	if len(d.outdatedActionDoneAt) >= outdatedActionDoneReportMaxEntries {
+		expireBefore := now.Add(-2 * outdatedActionDoneReportInterval)
+		for id, ts := range d.outdatedActionDoneAt {
+			if ts.Before(expireBefore) {
+				delete(d.outdatedActionDoneAt, id)
+			}
+		}
+		if len(d.outdatedActionDoneAt) >= outdatedActionDoneReportMaxEntries {
+			for id := range d.outdatedActionDoneAt {
+				delete(d.outdatedActionDoneAt, id)
+			}
+		}
+	}
+
+	d.outdatedActionDoneAt[identifier] = now
+	return true
+}
+
 // AddDMLEventsToSink filters events for special tables, registers batch wake
 // callbacks, and returns true when at least one event remains to be written to
 // the downstream sink.
@@ -836,6 +873,13 @@ func (d *BasicDispatcher) HandleDispatcherStatus(dispatcherStatus *heartbeatpb.D
 		}
 
 		// Step3: whether the outdate message or not, we need to return message show we have finished the event.
+		identifier := BlockEventIdentifier{
+			CommitTs:    action.CommitTs,
+			IsSyncPoint: action.IsSyncPoint,
+		}
+		if !d.shouldReportOutdatedActionDone(identifier, time.Now()) {
+			return false
+		}
 		d.sharedInfo.blockStatusesChan <- &heartbeatpb.TableSpanBlockStatus{
 			ID: d.id.ToPB(),
 			State: &heartbeatpb.State{

downstreamadapter/dispatcher/event_dispatcher_test.go

@@ -429,6 +429,58 @@ func TestDispatcherHandleEvents(t *testing.T) {
 	t.Run("cloud storage wake callback after batch enqueue", verifyDMLWakeCallbackStorageAfterBatchEnqueue)
 }
 
+func TestOutdatedActionDoneIsThrottled(t *testing.T) {
+	tableSpan, err := getCompleteTableSpan(getTestingKeyspaceID())
+	require.NoError(t, err)
+	testSink := newDispatcherTestSink(t, common.MysqlSinkType)
+	dispatcher := newDispatcherForTest(testSink.Sink(), tableSpan)
+	nodeID := node.NewID()
+
+	block := dispatcher.HandleEvents([]DispatcherEvent{
+		NewDispatcherEvent(&nodeID, commonEvent.ResolvedEvent{ResolvedTs: 10}),
+	}, func() {})
+	require.False(t, block)
+	require.Equal(t, uint64(10), dispatcher.GetResolvedTs())
+
+	staleStatus := &heartbeatpb.DispatcherStatus{
+		Action: &heartbeatpb.DispatcherAction{
+			Action:      heartbeatpb.Action_Pass,
+			CommitTs:    5,
+			IsSyncPoint: false,
+		},
+	}
+
+	await := dispatcher.HandleDispatcherStatus(staleStatus)
+	require.False(t, await)
+	select {
+	case msg := <-dispatcher.GetBlockStatusesChan():
+		require.Equal(t, heartbeatpb.BlockStage_DONE, msg.State.Stage)
+		require.Equal(t, uint64(5), msg.State.BlockTs)
+	case <-time.After(time.Second):
+		require.FailNow(t, "expected stale action DONE")
+	}
+
+	await = dispatcher.HandleDispatcherStatus(staleStatus)
+	require.False(t, await)
+	select {
+	case msg := <-dispatcher.GetBlockStatusesChan():
+		require.FailNow(t, "unexpected duplicate stale action DONE", "msg=%v", msg)
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	time.Sleep(outdatedActionDoneReportInterval + 50*time.Millisecond)
+
+	await = dispatcher.HandleDispatcherStatus(staleStatus)
+	require.False(t, await)
+	select {
+	case msg := <-dispatcher.GetBlockStatusesChan():
+		require.Equal(t, heartbeatpb.BlockStage_DONE, msg.State.Stage)
+		require.Equal(t, uint64(5), msg.State.BlockTs)
+	case <-time.After(time.Second):
+		require.FailNow(t, "expected stale action DONE after throttle interval")
+	}
+}
+
 func TestBlockingDDLFlushBeforeWaitingAndWriteDoesNotFlushAgain(t *testing.T) {
 	keyspaceID := getTestingKeyspaceID()
 	tableSpan := getUncompleteTableSpan()