dynamicstream: remove batch configuration from the stream level, and support it at the area level

downstreamadapter/dispatchermanager/helper_test.go

@@ -20,7 +20,7 @@ import (
 
 	"github.com/golang/mock/gomock"
 	"github.com/pingcap/ticdc/downstreamadapter/dispatcher"
-	"github.com/pingcap/ticdc/downstreamadapter/sink/mock"
+	sinkmock "github.com/pingcap/ticdc/downstreamadapter/sink/mock"
 	"github.com/pingcap/ticdc/downstreamadapter/sink/redo"
 	"github.com/pingcap/ticdc/heartbeatpb"
 	"github.com/pingcap/ticdc/pkg/common"
@@ -34,112 +34,97 @@ func TestCheckpointTsMessageHandlerDeadlock(t *testing.T) {
 		Keyspace: "test-namespace",
 		Name:     "test-changefeed",
 	}
-
 	checkpointTsMessage := NewCheckpointTsMessage(&heartbeatpb.CheckpointTsMessage{
 		ChangefeedID: changefeedID,
 		CheckpointTs: 12345,
 	})
-
 	handler := &CheckpointTsMessageHandler{}
 
-	t.Run("normal_operation", func(t *testing.T) {
+	t.Run("normalOperation", func(t *testing.T) {
 		ctrl := gomock.NewController(t)
-		defer ctrl.Finish()
-
-		mockSink := mock.NewMockSink(ctrl)
-		mockSink.EXPECT().AddCheckpointTs(checkpointTsMessage.CheckpointTs).Times(1)
-
+		mockSink := sinkmock.NewMockSink(ctrl)
+		mockSink.EXPECT().AddCheckpointTs(uint64(12345)).Times(1)
 		dispatcherManager := &DispatcherManager{
 			sink:                        mockSink,
 			tableTriggerEventDispatcher: &dispatcher.EventDispatcher{},
 		}
 
 		done := make(chan bool, 1)
 		go func() {
-			blocking := handler.Handle(dispatcherManager, checkpointTsMessage)
-			require.False(t, blocking, "Handler should not return blocking=true")
-			done <- true
+			done <- handler.Handle(dispatcherManager, checkpointTsMessage)
 		}()
 
 		select {
-		case <-done:
-		case <-time.After(1 * time.Second):
-			t.Fatal("Normal operation took too long, unexpected")
+		case blocking := <-done:
+			require.False(t, blocking)
+		case <-time.After(time.Second):
+			t.Fatal("normal operation timed out")
 		}
 	})
 
-	t.Run("deadlock_scenario", func(t *testing.T) {
+	t.Run("deadlockScenario", func(t *testing.T) {
 		ctrl := gomock.NewController(t)
-		defer ctrl.Finish()
-
+		mockSink := sinkmock.NewMockSink(ctrl)
 		blockCh := make(chan struct{})
-		deadlockMockSink := mock.NewMockSink(ctrl)
-		deadlockMockSink.EXPECT().AddCheckpointTs(checkpointTsMessage.CheckpointTs).Do(
-			func(uint64) {
-				<-blockCh
-			},
-		).Times(1)
-
-		deadlockDispatcherManager := &DispatcherManager{
-			sink:                        deadlockMockSink,
+		mockSink.EXPECT().AddCheckpointTs(uint64(12345)).DoAndReturn(func(uint64) {
+			<-blockCh
+		}).Times(1)
+
+		dispatcherManager := &DispatcherManager{
+			sink:                        mockSink,
 			tableTriggerEventDispatcher: &dispatcher.EventDispatcher{},
 		}
 
 		done := make(chan bool, 1)
 		go func() {
-			handler.Handle(deadlockDispatcherManager, checkpointTsMessage)
-			done <- true
+			done <- handler.Handle(dispatcherManager, checkpointTsMessage)
 		}()
 
 		select {
 		case <-done:
-			t.Fatal("Handler completed unexpectedly - deadlock was not reproduced")
-		case <-time.After(1 * time.Second):
-			t.Log("Successfully reproduced the deadlock: handler is blocked in AddCheckpointTs")
+			t.Fatal("handler completed unexpectedly, deadlock not reproduced")
+		case <-time.After(time.Second):
 		}
 
 		close(blockCh)
+
 		select {
-		case <-done:
-		case <-time.After(1 * time.Second):
-			t.Fatal("handler should resume once AddCheckpointTs unblocks")
+		case blocking := <-done:
+			require.False(t, blocking)
+		case <-time.After(time.Second):
+			t.Fatal("handler did not return after unblocking AddCheckpointTs")
 		}
 	})
 
-	t.Run("deadlock_resolve_scenario", func(t *testing.T) {
+	t.Run("deadlockResolveScenario", func(t *testing.T) {
 		ctrl := gomock.NewController(t)
-		defer ctrl.Finish()
-
-		ctx, cancel := context.WithCancel(context.Background())
-		cancel()
+		mockSink := sinkmock.NewMockSink(ctrl)
+		sinkCtx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		mockSink.EXPECT().AddCheckpointTs(uint64(12345)).DoAndReturn(func(uint64) {
+			<-sinkCtx.Done()
+		}).Times(1)
 
-		deadlockMockSink := mock.NewMockSink(ctrl)
-		deadlockMockSink.EXPECT().AddCheckpointTs(checkpointTsMessage.CheckpointTs).Do(
-			func(uint64) {
-				select {
-				case <-ctx.Done():
-					return
-				case <-time.After(1 * time.Second):
-					t.Fatal("context cancellation should unblock AddCheckpointTs path")
-				}
-			},
-		).Times(1)
-
-		deadlockDispatcherManager := &DispatcherManager{
-			sink:                        deadlockMockSink,
+		dispatcherManager := &DispatcherManager{
+			sink:                        mockSink,
 			tableTriggerEventDispatcher: &dispatcher.EventDispatcher{},
 		}
-
 		done := make(chan bool, 1)
 		go func() {
-			handler.Handle(deadlockDispatcherManager, checkpointTsMessage)
-			done <- true
+			done <- handler.Handle(dispatcherManager, checkpointTsMessage)
 		}()
 
 		select {
 		case <-done:
-			t.Log("Handler completed normally")
-		case <-time.After(1 * time.Second):
+			t.Fatal("handler completed unexpectedly before cancel")
+		case <-time.After(200 * time.Millisecond):
+		}
+
+		cancel()
+		select {
+		case blocking := <-done:
+			require.False(t, blocking)
+		case <-time.After(time.Second):
 			t.Fatal("deadlock: handler is blocked in AddCheckpointTs")
 		}
 	})

downstreamadapter/eventcollector/event_collector.go

@@ -285,6 +285,8 @@ func (c *EventCollector) PrepareAddDispatcher(
 	err := ds.AddPath(target.GetId(), stat, areaSetting)
 	if err != nil {
 		log.Warn("add dispatcher to dynamic stream failed", zap.Error(err))
+		stat.run()
+		return
 	}
 	stat.run()
 }

downstreamadapter/sink/mysql/sink.go

@@ -61,6 +61,7 @@ type Sink struct {
 	cfg        *mysql.Config
 	maxTxnRows int
 	bdrMode    bool
+
 	// enableActiveActive enables active-active replication behaviors in the MySQL-class sink.
 	enableActiveActive bool
 
@@ -440,7 +441,7 @@ func (s *Sink) CleanupRemovedChangefeed() error {
 }
 
 func (s *Sink) BatchCount() int {
-	return s.maxTxnRows * len(s.dmlWriter)
+	return s.cfg.MaxTxnRow * len(s.dmlWriter)
 }
 
 func (s *Sink) BatchBytes() int {

downstreamadapter/sink/mysql/sink_test.go

@@ -68,6 +68,7 @@ func MysqlSinkForTest() (*Sink, sqlmock.Sqlmock) {
 
 func MysqlSinkForTestWithMaxTxnRows(maxTxnRows int) (*Sink, sqlmock.Sqlmock) {
 	ctx, sink, mock := getMysqlSink()
+	sink.cfg.MaxTxnRow = maxTxnRows
 	sink.maxTxnRows = maxTxnRows
 	go sink.Run(ctx)
 	return sink, mock
@@ -334,12 +335,12 @@ func TestMysqlSinkFlushLargeBatchEvent(t *testing.T) {
 	require.Equal(t, int32(10), dmlEvent1.Length, "First event should contain 10 rows")
 	require.Equal(t, int32(10), dmlEvent2.Length, "Second event should contain 10 rows")
 
-	// Set up mock expectations for DDL
-	mock.ExpectExec("BEGIN;INSERT INTO `test`.`t` (`id`,`name`) VALUES (?,?),(?,?),(?,?),(?,?),(?,?),(?,?),(?,?),(?,?),(?,?),(?,?);COMMIT;").
+	// Each 10-row event is split into 4+4+2 rows when maxTxnRows is 4.
+	mock.ExpectExec("BEGIN;INSERT INTO `test`.`t` (`id`,`name`) VALUES (?,?),(?,?),(?,?),(?,?);INSERT INTO `test`.`t` (`id`,`name`) VALUES (?,?),(?,?),(?,?),(?,?);INSERT INTO `test`.`t` (`id`,`name`) VALUES (?,?),(?,?);COMMIT;").
 		WithArgs(1, "test1", 2, "test2", 3, "test3", 4, "test4", 5, "test5", 6, "test6", 7, "test7", 8, "test8", 9, "test9", 10, "test10").
 		WillReturnResult(sqlmock.NewResult(1, 1))
 
-	mock.ExpectExec("BEGIN;INSERT INTO `test`.`t` (`id`,`name`) VALUES (?,?),(?,?),(?,?),(?,?),(?,?),(?,?),(?,?),(?,?),(?,?),(?,?);COMMIT;").
+	mock.ExpectExec("BEGIN;INSERT INTO `test`.`t` (`id`,`name`) VALUES (?,?),(?,?),(?,?),(?,?);INSERT INTO `test`.`t` (`id`,`name`) VALUES (?,?),(?,?),(?,?),(?,?);INSERT INTO `test`.`t` (`id`,`name`) VALUES (?,?),(?,?);COMMIT;").
 		WithArgs(11, "test11", 12, "test12", 13, "test13", 14, "test14", 15, "test15", 16, "test16", 17, "test17", 18, "test18", 19, "test19", 20, "test20").
 		WillReturnResult(sqlmock.NewResult(1, 1))
 

logservice/logpuller/region_event_handler_test.go

@@ -75,7 +75,7 @@ func TestHandleEventEntryEventOutOfOrder(t *testing.T) {
 		advanceResolvedTs: advanceResolvedTs,
 		advanceInterval:   0,
 	}
-	ds.AddPath(subID, subSpan, dynstream.AreaSettings{})
+	ds.AddPath(subID, subSpan)
 
 	worker := &regionRequestWorker{
 		requestCache: &requestCache{},
@@ -235,7 +235,7 @@ func TestHandleResolvedTs(t *testing.T) {
 			advanceResolvedTs: advanceResolvedTs,
 			advanceInterval:   0,
 		}
-		ds.AddPath(subID1, subSpan, dynstream.AreaSettings{})
+		ds.AddPath(subID1, subSpan)
 		state1.region.subscribedSpan = subSpan
 		state1.region.lockedRangeState = &regionlock.LockedRangeState{}
 		state1.setInitialized()
@@ -259,7 +259,7 @@ func TestHandleResolvedTs(t *testing.T) {
 			advanceResolvedTs: advanceResolvedTs,
 			advanceInterval:   0,
 		}
-		ds.AddPath(subID2, subSpan, dynstream.AreaSettings{})
+		ds.AddPath(subID2, subSpan)
 		state2.region.subscribedSpan = subSpan
 		state2.region.lockedRangeState = &regionlock.LockedRangeState{}
 		state2.setInitialized()
@@ -283,7 +283,7 @@ func TestHandleResolvedTs(t *testing.T) {
 			advanceResolvedTs: advanceResolvedTs,
 			advanceInterval:   0,
 		}
-		ds.AddPath(subID3, subSpan, dynstream.AreaSettings{})
+		ds.AddPath(subID3, subSpan)
 		state3.region.subscribedSpan = subSpan
 		state3.region.lockedRangeState = &regionlock.LockedRangeState{}
 		state3.updateResolvedTs(8)

pkg/config/consistent.go

@@ -110,6 +110,11 @@ func (c *ConsistentConfig) validateAndAdjust(enableIOCheck bool) error {
 			fmt.Sprintf("The consistent.meta-flush-interval:%d must be equal or greater than %d",
 				util.GetOrZero(c.MetaFlushIntervalInMs), redo.MinFlushIntervalInMs))
 	}
+	if c.EventCollectorBatchCount != nil && *c.EventCollectorBatchCount <= 0 {
+		return errors.ErrInvalidReplicaConfig.FastGenByArgs(
+			fmt.Sprintf("The consistent.event-collector-batch-count:%d must be larger than 0",
+				*c.EventCollectorBatchCount))
+	}
 
 	if c.EventCollectorBatchCount != nil {
 		if *c.EventCollectorBatchCount < 0 {

pkg/metrics/dynamic_stream.go

@@ -77,11 +77,11 @@ var (
 
 func initDynamicStreamMetrics(registry *prometheus.Registry) {
 	registry.MustRegister(DynamicStreamMemoryUsage)
+	registry.MustRegister(DynamicStreamBatchCount)
+	registry.MustRegister(DynamicStreamBatchBytes)
+	registry.MustRegister(DynamicStreamBatchDuration)
 	registry.MustRegister(DynamicStreamEventChanSize)
 	registry.MustRegister(DynamicStreamPendingQueueLen)
 	registry.MustRegister(DynamicStreamAddPathNum)
 	registry.MustRegister(DynamicStreamRemovePathNum)
-	registry.MustRegister(DynamicStreamBatchCount)
-	registry.MustRegister(DynamicStreamBatchBytes)
-	registry.MustRegister(DynamicStreamBatchDuration)
 }

scripts/generate-mock.sh

@@ -36,6 +36,7 @@ fi
 "$MOCKGEN" -source pkg/sink/codec/simple/marshaller.go -destination pkg/sink/codec/simple/mock/marshaller.go
 "$MOCKGEN" -source pkg/keyspace/keyspace_manager.go -destination pkg/keyspace/keyspace_manager_mock.go -package keyspace
 "$MOCKGEN" -source pkg/txnutil/gc/gc_manager.go -destination pkg/txnutil/gc/gc_manager_mock.go -package gc
+"$MOCKGEN" -source downstreamadapter/sink/sink.go -destination downstreamadapter/sink/mock/sink_mock.go -package mock
 "$MOCKGEN" -source pkg/txnutil/gc/gc_client.go -destination pkg/txnutil/gc/gc_client_mock.go -package gc
 "$MOCKGEN" -source pkg/redo/writer/writer.go -destination pkg/redo/writer/writer_mock.go -package writer
 "$MOCKGEN" -source downstreamadapter/sink/sink.go -destination downstreamadapter/sink/mock/sink_mock.go -package mock

utils/dynstream/stream.go

@@ -251,6 +251,9 @@ func (s *stream[A, P, T, D, H]) handleLoop() {
 		s.option.handleWait.Wait()
 	}
 
+	// Variables below will be used in the Loop below.
+	// Declared here to avoid repeated allocation.
+	// todo: shall we preallocate the event buff and path here ?
 	// 1. Drain the eventChan to pendingQueue.
 	// 2. Pop events from the eventQueue and handle them.
 Loop: