From 4b98a2c9ba557ce4478a256229d39a817d1fd5db Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Thu, 13 Aug 2026 18:29:51 +0800 Subject: [PATCH 1/6] first commit --- downstreamadapter/sink/kafka/sink_test.go | 315 ++++++++++++++++++ .../sink/topicmanager/topic_manager_mock.go | 77 +++++ scripts/generate-mock.sh | 1 + 3 files changed, 393 insertions(+) create mode 100644 downstreamadapter/sink/topicmanager/topic_manager_mock.go diff --git a/downstreamadapter/sink/kafka/sink_test.go b/downstreamadapter/sink/kafka/sink_test.go index f4e7860d4c..b06d5767bd 100644 --- a/downstreamadapter/sink/kafka/sink_test.go +++ b/downstreamadapter/sink/kafka/sink_test.go @@ -27,13 +27,16 @@ import ( "github.com/pingcap/ticdc/downstreamadapter/sink/eventrouter" "github.com/pingcap/ticdc/downstreamadapter/sink/helper" "github.com/pingcap/ticdc/downstreamadapter/sink/topicmanager" + "github.com/pingcap/ticdc/heartbeatpb" "github.com/pingcap/ticdc/pkg/common" commonEvent "github.com/pingcap/ticdc/pkg/common/event" "github.com/pingcap/ticdc/pkg/config" "github.com/pingcap/ticdc/pkg/errors" + "github.com/pingcap/ticdc/pkg/metrics" "github.com/pingcap/ticdc/pkg/sink/codec" codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/pingcap/ticdc/pkg/sink/kafka" + "github.com/pingcap/tidb/pkg/meta/model" "github.com/stretchr/testify/require" "go.uber.org/atomic" ) @@ -356,3 +359,315 @@ func TestKafkaSinkBatchConfig(t *testing.T) { require.Equal(t, 4096, sink.BatchCount()) require.Zero(t, sink.BatchBytes()) } + +func TestKafkaSinkConstructionAndCleanup(t *testing.T) { + t.Run("async producer creation fails", func(t *testing.T) { + ctrl := gomock.NewController(t) + factory := kafka.NewMockFactory(ctrl) + adminClient := kafka.NewMockAdminClient(ctrl) + topicManager := topicmanager.NewMockTopicManager(ctrl) + cause := errors.ErrKafkaSendMessage.GenWithStackByArgs() + + factory.EXPECT().AsyncProducer(gomock.Any()).Return(nil, cause) + adminClient.EXPECT().Close() + topicManager.EXPECT().Close() + + kafkaSink, err := newWithComponents( + t.Context(), + common.NewChangefeedID4Test("test", "async-creation-fails"), + common.DefaultKeyspaceID, + config.ProtocolOpen, + components{factory: factory, adminClient: adminClient, topicManager: topicManager}, + ) + + require.Nil(t, kafkaSink) + require.Equal(t, cause, err) + }) + + t.Run("sync producer creation fails", func(t *testing.T) { + ctrl := gomock.NewController(t) + factory := kafka.NewMockFactory(ctrl) + adminClient := kafka.NewMockAdminClient(ctrl) + topicManager := topicmanager.NewMockTopicManager(ctrl) + asyncProducer := kafka.NewMockAsyncProducer(ctrl) + cause := errors.ErrKafkaSendMessage.GenWithStackByArgs() + + factory.EXPECT().AsyncProducer(gomock.Any()).Return(asyncProducer, nil) + factory.EXPECT().SyncProducer(gomock.Any()).Return(nil, cause) + asyncProducer.EXPECT().Close() + adminClient.EXPECT().Close() + topicManager.EXPECT().Close() + + kafkaSink, err := newWithComponents( + t.Context(), + common.NewChangefeedID4Test("test", "sync-creation-fails"), + common.DefaultKeyspaceID, + config.ProtocolOpen, + components{factory: factory, adminClient: adminClient, topicManager: topicManager}, + ) + + require.Nil(t, kafkaSink) + require.Equal(t, cause, err) + }) + + t.Run("successful construction owns resources until close", func(t *testing.T) { + ctrl := gomock.NewController(t) + factory := kafka.NewMockFactory(ctrl) + adminClient := kafka.NewMockAdminClient(ctrl) + topicManager := topicmanager.NewMockTopicManager(ctrl) + asyncProducer := kafka.NewMockAsyncProducer(ctrl) + syncProducer := kafka.NewMockSyncProducer(ctrl) + var closeCount atomic.Int64 + + factory.EXPECT().AsyncProducer(gomock.Any()).Return(asyncProducer, nil) + factory.EXPECT().SyncProducer(gomock.Any()).Return(syncProducer, nil) + factory.EXPECT().MetricsCollector(adminClient).Return(noopMetricsCollector{}) + asyncProducer.EXPECT().Close().Do(func() { closeCount.Add(1) }) + syncProducer.EXPECT().Close().Do(func() { closeCount.Add(1) }) + adminClient.EXPECT().Close().Do(func() { closeCount.Add(1) }) + topicManager.EXPECT().Close().Do(func() { closeCount.Add(1) }) + + kafkaSink, err := newWithComponents( + t.Context(), + common.NewChangefeedID4Test("test", "successful-construction"), + common.DefaultKeyspaceID, + config.ProtocolOpen, + components{factory: factory, adminClient: adminClient, topicManager: topicManager}, + ) + + require.NoError(t, err) + require.Zero(t, closeCount.Load()) + kafkaSink.Close() + require.Equal(t, int64(4), closeCount.Load()) + }) +} + +func TestKafkaSinkDML(t *testing.T) { + eventHelper := commonEvent.NewEventTestHelper(t) + defer eventHelper.Close() + eventHelper.Tk().MustExec("use test") + require.NotNil(t, eventHelper.DDL2Job("create table t (id int primary key, name varchar(32))")) + + t.Run("forwards routed message and waits for producer callback", func(t *testing.T) { + var callbackCount atomic.Int64 + dmlEvent := eventHelper.DML2Event("test", "t", "insert into t values (1, 'one')") + dmlEvent.PostTxnFlushed = []func(){func() { callbackCount.Add(1) }} + + ctrl := gomock.NewController(t) + asyncProducer := kafka.NewMockAsyncProducer(ctrl) + syncProducer := kafka.NewMockSyncProducer(ctrl) + sent := make(chan *codecCommon.Message, 1) + asyncProducer.EXPECT().AsyncSend(gomock.Any(), kafkaSinkTestTopic, int32(0), gomock.Any()). + DoAndReturn(func(_ context.Context, _ string, _ int32, message *codecCommon.Message) error { + sent <- message + return nil + }) + asyncProducer.EXPECT().Close() + syncProducer.EXPECT().Close() + + ctx, cancel := context.WithCancelCause(t.Context()) + kafkaSink, err := newKafkaSinkForTestWithProducers(ctx, t, ctrl, asyncProducer, syncProducer) + require.NoError(t, err) + defer kafkaSink.Close() + + runDone := make(chan error, 1) + go func() { runDone <- kafkaSink.sendDMLEvent(ctx) }() + kafkaSink.AddDMLEvent(dmlEvent) + + select { + case message := <-sent: + require.NotEmpty(t, message.Key) + require.NotEmpty(t, message.Value) + require.Equal(t, 1, message.GetRowsCount()) + require.NotNil(t, message.Callback) + require.Zero(t, callbackCount.Load()) + case <-time.After(10 * time.Second): + t.Fatal("timed out waiting for Kafka Sink to send the DML message") + } + + cause := errors.ErrKafkaSinkClosed.GenWithStackByArgs() + cancel(cause) + require.Equal(t, cause, <-runDone) + require.Zero(t, callbackCount.Load()) + }) + + t.Run("returns AsyncSend error unchanged", func(t *testing.T) { + dmlEvent := eventHelper.DML2Event("test", "t", "insert into t values (2, 'two')") + ctrl := gomock.NewController(t) + asyncProducer := kafka.NewMockAsyncProducer(ctrl) + syncProducer := kafka.NewMockSyncProducer(ctrl) + cause := errors.ErrKafkaSendMessage.GenWithStackByArgs() + asyncProducer.EXPECT().AsyncSend(gomock.Any(), kafkaSinkTestTopic, int32(0), gomock.Any()).Return(cause) + asyncProducer.EXPECT().Close() + syncProducer.EXPECT().Close() + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + kafkaSink, err := newKafkaSinkForTestWithProducers(ctx, t, ctrl, asyncProducer, syncProducer) + require.NoError(t, err) + defer kafkaSink.Close() + + kafkaSink.AddDMLEvent(dmlEvent) + err = kafkaSink.sendDMLEvent(ctx) + + require.Equal(t, cause, err) + }) +} + +func TestKafkaSinkDDL(t *testing.T) { + ddlEvent := &commonEvent.DDLEvent{ + Type: byte(model.ActionCreateTable), + SchemaName: "test", + TableName: "t", + Query: "create table test.t (id int primary key)", + FinishedTs: 1, + } + + t.Run("all partitions", func(t *testing.T) { + kafkaSink, topicManager, producer := newKafkaSinkForTest(t, config.ProtocolOpen, &config.SinkConfig{}) + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(4), nil) + producer.EXPECT().SendMessages(kafkaSinkTestTopic, int32(4), gomock.Any()). + DoAndReturn(func(_ string, _ int32, message *codecCommon.Message) error { + require.NotEmpty(t, message.Key) + require.NotEmpty(t, message.Value) + return nil + }) + + require.NoError(t, kafkaSink.sendDDLEvent(ddlEvent)) + }) + + t.Run("partition zero", func(t *testing.T) { + kafkaSink, topicManager, producer := newKafkaSinkForTest(t, config.ProtocolCanalJSON, &config.SinkConfig{}) + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(4), nil) + producer.EXPECT().SendMessage(kafkaSinkTestTopic, int32(0), gomock.Any()). + DoAndReturn(func(_ string, _ int32, message *codecCommon.Message) error { + require.NotEmpty(t, message.Value) + return nil + }) + + require.NoError(t, kafkaSink.sendDDLEvent(ddlEvent)) + }) + + t.Run("topic manager error", func(t *testing.T) { + kafkaSink, topicManager, _ := newKafkaSinkForTest(t, config.ProtocolOpen, &config.SinkConfig{}) + cause := context.DeadlineExceeded + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(0), cause) + + require.Equal(t, cause, kafkaSink.sendDDLEvent(ddlEvent)) + }) + + t.Run("producer error", func(t *testing.T) { + kafkaSink, topicManager, producer := newKafkaSinkForTest(t, config.ProtocolOpen, &config.SinkConfig{}) + cause := errors.ErrKafkaSendMessage.GenWithStackByArgs() + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(2), nil) + producer.EXPECT().SendMessages(kafkaSinkTestTopic, int32(2), gomock.Any()).Return(cause) + + require.Equal(t, cause, kafkaSink.sendDDLEvent(ddlEvent)) + }) + + t.Run("nil encoded message", func(t *testing.T) { + kafkaSink, _, _ := newKafkaSinkForTest(t, config.ProtocolDebezium, &config.SinkConfig{}) + unsupportedDDL := &commonEvent.DDLEvent{Type: byte(model.ActionNone), Query: "unsupported"} + + require.NoError(t, kafkaSink.sendDDLEvent(unsupportedDDL)) + }) +} + +func TestKafkaSinkCheckpoint(t *testing.T) { + t.Run("default topic without tables", func(t *testing.T) { + kafkaSink, topicManager, producer := newKafkaSinkForTest(t, config.ProtocolOpen, &config.SinkConfig{}) + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(3), nil) + producer.EXPECT().SendMessages(kafkaSinkTestTopic, int32(3), gomock.Any()). + DoAndReturn(func(_ string, _ int32, message *codecCommon.Message) error { + require.NotEmpty(t, message.Key) + return nil + }) + kafkaSink.checkpointChan <- 100 + close(kafkaSink.checkpointChan) + + require.NoError(t, kafkaSink.sendCheckpoint(t.Context())) + }) + + t.Run("all active topics", func(t *testing.T) { + sinkConfig := &config.SinkConfig{DispatchRules: []*config.DispatchRule{ + {Matcher: []string{"db1.t1"}, PartitionRule: "table", TopicRule: "topic-a"}, + {Matcher: []string{"db2.t2"}, PartitionRule: "table", TopicRule: "topic-b"}, + }} + kafkaSink, topicManager, producer := newKafkaSinkForTest(t, config.ProtocolOpen, sinkConfig) + kafkaSink.SetTableSchemaStore(commonEvent.NewTableSchemaStore([]*heartbeatpb.SchemaInfo{ + {SchemaName: "db1", Tables: []*heartbeatpb.TableInfo{{TableName: "t1"}}}, + {SchemaName: "db2", Tables: []*heartbeatpb.TableInfo{{TableName: "t2"}}}, + }, common.KafkaSinkType, false)) + partitionCounts := map[string]int32{"topic-a": 2, "topic-b": 3, kafkaSinkTestTopic: 4} + for topic, partitionCount := range partitionCounts { + topicManager.EXPECT().GetPartitionNum(gomock.Any(), topic).Return(partitionCount, nil) + producer.EXPECT().SendMessages(topic, partitionCount, gomock.Any()).Return(nil) + } + kafkaSink.checkpointChan <- 100 + close(kafkaSink.checkpointChan) + + require.NoError(t, kafkaSink.sendCheckpoint(t.Context())) + }) + + t.Run("topic manager error", func(t *testing.T) { + kafkaSink, topicManager, _ := newKafkaSinkForTest(t, config.ProtocolOpen, &config.SinkConfig{}) + cause := context.DeadlineExceeded + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(0), cause) + kafkaSink.checkpointChan <- 100 + + require.Equal(t, cause, kafkaSink.sendCheckpoint(t.Context())) + }) + + t.Run("producer error", func(t *testing.T) { + kafkaSink, topicManager, producer := newKafkaSinkForTest(t, config.ProtocolOpen, &config.SinkConfig{}) + cause := errors.ErrKafkaSendMessage.GenWithStackByArgs() + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(2), nil) + producer.EXPECT().SendMessages(kafkaSinkTestTopic, int32(2), gomock.Any()).Return(cause) + kafkaSink.checkpointChan <- 100 + + require.Equal(t, cause, kafkaSink.sendCheckpoint(t.Context())) + }) + + t.Run("context cancellation", func(t *testing.T) { + kafkaSink, _, _ := newKafkaSinkForTest(t, config.ProtocolOpen, &config.SinkConfig{}) + ctx, cancel := context.WithCancelCause(t.Context()) + cause := errors.ErrKafkaSinkClosed.GenWithStackByArgs() + cancel(cause) + + require.Equal(t, cause, kafkaSink.sendCheckpoint(ctx)) + }) +} + +func newKafkaSinkForTest( + t *testing.T, + protocol config.Protocol, + sinkConfig *config.SinkConfig, +) (*sink, *topicmanager.MockTopicManager, *kafka.MockSyncProducer) { + t.Helper() + + ctrl := gomock.NewController(t) + changefeedID := common.NewChangefeedID4Test("test", t.Name()) + router, err := eventrouter.NewEventRouter(sinkConfig, kafkaSinkTestTopic, false, false) + require.NoError(t, err) + encoder, err := codec.NewEventEncoder(t.Context(), codecCommon.NewConfig(protocol).WithChangefeedID(changefeedID), nil) + require.NoError(t, err) + statistics := metrics.NewStatistics(changefeedID, common.DefaultKeyspaceID, "sink") + t.Cleanup(statistics.Close) + topicManager := topicmanager.NewMockTopicManager(ctrl) + producer := kafka.NewMockSyncProducer(ctrl) + + return &sink{ + changefeedID: changefeedID, + ddlProducer: producer, + partitionRule: helper.GetDDLDispatchRule(protocol), + protocol: protocol, + comp: components{ + encoder: encoder, + eventRouter: router, + topicManager: topicManager, + }, + statistics: statistics, + checkpointChan: make(chan uint64, 1), + ctx: t.Context(), + }, topicManager, producer +} diff --git a/downstreamadapter/sink/topicmanager/topic_manager_mock.go b/downstreamadapter/sink/topicmanager/topic_manager_mock.go new file mode 100644 index 0000000000..5e4aa0cd58 --- /dev/null +++ b/downstreamadapter/sink/topicmanager/topic_manager_mock.go @@ -0,0 +1,77 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: downstreamadapter/sink/topicmanager/topic_manager.go + +// Package topicmanager is a generated GoMock package. +package topicmanager + +import ( + context "context" + reflect "reflect" + + gomock "github.com/golang/mock/gomock" +) + +// MockTopicManager is a mock of TopicManager interface. +type MockTopicManager struct { + ctrl *gomock.Controller + recorder *MockTopicManagerMockRecorder +} + +// MockTopicManagerMockRecorder is the mock recorder for MockTopicManager. +type MockTopicManagerMockRecorder struct { + mock *MockTopicManager +} + +// NewMockTopicManager creates a new mock instance. +func NewMockTopicManager(ctrl *gomock.Controller) *MockTopicManager { + mock := &MockTopicManager{ctrl: ctrl} + mock.recorder = &MockTopicManagerMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockTopicManager) EXPECT() *MockTopicManagerMockRecorder { + return m.recorder +} + +// Close mocks base method. +func (m *MockTopicManager) Close() { + m.ctrl.T.Helper() + m.ctrl.Call(m, "Close") +} + +// Close indicates an expected call of Close. +func (mr *MockTopicManagerMockRecorder) Close() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockTopicManager)(nil).Close)) +} + +// CreateTopicAndWaitUntilVisible mocks base method. +func (m *MockTopicManager) CreateTopicAndWaitUntilVisible(ctx context.Context, topicName string) (int32, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "CreateTopicAndWaitUntilVisible", ctx, topicName) + ret0, _ := ret[0].(int32) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// CreateTopicAndWaitUntilVisible indicates an expected call of CreateTopicAndWaitUntilVisible. +func (mr *MockTopicManagerMockRecorder) CreateTopicAndWaitUntilVisible(ctx, topicName interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateTopicAndWaitUntilVisible", reflect.TypeOf((*MockTopicManager)(nil).CreateTopicAndWaitUntilVisible), ctx, topicName) +} + +// GetPartitionNum mocks base method. +func (m *MockTopicManager) GetPartitionNum(ctx context.Context, topic string) (int32, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetPartitionNum", ctx, topic) + ret0, _ := ret[0].(int32) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// GetPartitionNum indicates an expected call of GetPartitionNum. +func (mr *MockTopicManagerMockRecorder) GetPartitionNum(ctx, topic interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetPartitionNum", reflect.TypeOf((*MockTopicManager)(nil).GetPartitionNum), ctx, topic) +} diff --git a/scripts/generate-mock.sh b/scripts/generate-mock.sh index e4f0a36b51..d7c654cf06 100755 --- a/scripts/generate-mock.sh +++ b/scripts/generate-mock.sh @@ -38,6 +38,7 @@ fi "$MOCKGEN" -source pkg/sink/kafka/factory.go -destination pkg/sink/kafka/factory_mock.go -package kafka "$MOCKGEN" -source pkg/sink/kafka/admin.go -destination pkg/sink/kafka/admin_mock.go -package kafka "$MOCKGEN" -source pkg/sink/kafka/sarama_sync_producer.go -destination pkg/sink/kafka/sarama_sync_producer_mock.go -package kafka +"$MOCKGEN" -source downstreamadapter/sink/topicmanager/topic_manager.go -destination downstreamadapter/sink/topicmanager/topic_manager_mock.go -package topicmanager "$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 pkg/txnutil/gc/gc_client.go -destination pkg/txnutil/gc/gc_client_mock.go -package gc From 5a21dba1cab69e3a8e97aa476d5713e92d1bc34e Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Thu, 13 Aug 2026 19:09:43 +0800 Subject: [PATCH 2/6] unify test code --- downstreamadapter/sink/kafka/sink_test.go | 305 ++++++++-------------- 1 file changed, 114 insertions(+), 191 deletions(-) diff --git a/downstreamadapter/sink/kafka/sink_test.go b/downstreamadapter/sink/kafka/sink_test.go index b06d5767bd..b46d807dcd 100644 --- a/downstreamadapter/sink/kafka/sink_test.go +++ b/downstreamadapter/sink/kafka/sink_test.go @@ -15,7 +15,6 @@ package kafka import ( "context" - "fmt" "net/http" "net/http/httptest" "net/url" @@ -25,14 +24,12 @@ import ( "github.com/golang/mock/gomock" "github.com/pingcap/ticdc/downstreamadapter/sink/columnselector" "github.com/pingcap/ticdc/downstreamadapter/sink/eventrouter" - "github.com/pingcap/ticdc/downstreamadapter/sink/helper" "github.com/pingcap/ticdc/downstreamadapter/sink/topicmanager" "github.com/pingcap/ticdc/heartbeatpb" "github.com/pingcap/ticdc/pkg/common" commonEvent "github.com/pingcap/ticdc/pkg/common/event" "github.com/pingcap/ticdc/pkg/config" "github.com/pingcap/ticdc/pkg/errors" - "github.com/pingcap/ticdc/pkg/metrics" "github.com/pingcap/ticdc/pkg/sink/codec" codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/pingcap/ticdc/pkg/sink/kafka" @@ -131,130 +128,15 @@ func TestVerifyInvalidConfig(t *testing.T) { require.ErrorContains(t, err, "ErrAvroSchemaAPIError") } -func newKafkaSinkForTestWithProducers(ctx context.Context, - t *testing.T, - ctrl *gomock.Controller, - asyncProducer kafka.AsyncProducer, - syncProducer kafka.SyncProducer, -) (*sink, error) { - t.Helper() - - changefeedID := common.NewChangefeedID4Test("test", "test") - openProtocol := config.ProtocolOpen.String() - sinkConfig := &config.SinkConfig{Protocol: &openProtocol} - uriTemplate := "kafka://%s/%s?kafka-version=0.9.0.0&max-batch-size=1" + - "&max-message-bytes=1048576&partition-num=1" + - "&kafka-client-id=unit-test&auto-create-topic=false&compression=gzip&protocol=open-protocol" - uri := fmt.Sprintf(uriTemplate, "127.0.0.1:9092", kafkaSinkTestTopic) - - sinkURI, err := url.Parse(uri) - if err != nil { - return nil, err - } - protocol, err := helper.GetProtocol(openProtocol) - if err != nil { - return nil, err - } - topic, err := helper.GetTopic(sinkURI) - if err != nil { - return nil, err - } - options := kafka.NewOptions() - if err = options.Apply(changefeedID, sinkURI, sinkConfig); err != nil { - return nil, err - } - options.Topic = topic - - adminClient := kafka.NewMockAdminClient(ctrl) - adminClient.EXPECT().GetTopicsMeta([]string{kafkaSinkTestTopic}, true).Return( - map[string]kafka.TopicDetail{ - kafkaSinkTestTopic: { - Name: kafkaSinkTestTopic, - NumPartitions: 1, - }, - }, nil) - adminClient.EXPECT().Close().AnyTimes() - - factory := kafka.NewMockFactory(ctrl) - factory.EXPECT().AsyncProducer(gomock.Any()).Return(asyncProducer, nil) - factory.EXPECT().SyncProducer(gomock.Any()).Return(syncProducer, nil) - factory.EXPECT().MetricsCollector(adminClient).Return(noopMetricsCollector{}) - - eventRouter, err := eventrouter.NewEventRouter(sinkConfig, topic, false, false) - if err != nil { - return nil, err - } - columnSelector, err := columnselector.New(sinkConfig) - if err != nil { - return nil, err - } - encoderConfig, err := helper.GetEncoderConfig( - changefeedID, sinkURI, protocol, sinkConfig, - options.MaxMessageBytes, options.MaxBatchedBytes, - ) - if err != nil { - return nil, err - } - encoderGroup, err := codec.NewEncoderGroup(ctx, sinkConfig, encoderConfig, nil, changefeedID) - if err != nil { - return nil, err - } - encoder, err := codec.NewEventEncoder(ctx, encoderConfig, nil) - if err != nil { - return nil, err - } - topicManager, err := topicmanager.GetTopicManagerAndTryCreateTopic( - ctx, - changefeedID, - topic, - options.DeriveTopicConfig(), - adminClient, - ) - if err != nil { - return nil, err - } - - comp := components{ - encoderGroup: encoderGroup, - encoder: encoder, - columnSelector: columnSelector, - eventRouter: eventRouter, - topicManager: topicManager, - adminClient: adminClient, - factory: factory, - } - - // We must close adminClient when this func return cause by an error - // otherwise the adminClient will never be closed and lead to a goroutine leak. - defer func() { - if err != nil && comp.adminClient != nil { - comp.close() - } - }() - - s, err := newWithComponents(ctx, changefeedID, common.DefaultKeyspaceID, protocol, comp) - if err != nil { - return nil, err - } - return s, nil -} - func TestKafkaSinkRunReturnsAsyncProducerError(t *testing.T) { ctx := t.Context() - ctrl := gomock.NewController(t) producerErr := errors.ErrKafkaSendMessage.GenWithStackByArgs() - asyncProducer := kafka.NewMockAsyncProducer(ctrl) - syncProducer := kafka.NewMockSyncProducer(ctrl) + kafkaSink, _, asyncProducer, _ := newKafkaSinkForTest(t, ctx, config.ProtocolOpen, &config.SinkConfig{}) asyncProducer.EXPECT().AsyncRunCallback(gomock.Any()).Return(producerErr) - asyncProducer.EXPECT().Close().AnyTimes() - syncProducer.EXPECT().Close().AnyTimes() - - kafkaSink, err := newKafkaSinkForTestWithProducers(ctx, t, ctrl, asyncProducer, syncProducer) - require.NoError(t, err) defer kafkaSink.Close() - err = kafkaSink.Run(ctx) + err := kafkaSink.Run(ctx) require.ErrorIs(t, err, errors.ErrKafkaSendMessage) require.False(t, kafkaSink.IsNormal()) @@ -311,9 +193,9 @@ func TestKafkaSinkBasicFunctionality(t *testing.T) { dmlEvent.CommitTs = 2 ctx, cancel := context.WithCancel(context.Background()) - ctrl := gomock.NewController(t) - asyncProducer := kafka.NewMockAsyncProducer(ctrl) - syncProducer := kafka.NewMockSyncProducer(ctrl) + kafkaSink, topicManager, asyncProducer, syncProducer := newKafkaSinkForTest( + t, ctx, config.ProtocolOpen, &config.SinkConfig{}) + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(1), nil).AnyTimes() asyncProducer.EXPECT().AsyncRunCallback(gomock.Any()).Return(nil).AnyTimes() asyncProducer.EXPECT().AsyncSend(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()). DoAndReturn(func( @@ -327,16 +209,11 @@ func TestKafkaSinkBasicFunctionality(t *testing.T) { } return nil }).Times(2) - asyncProducer.EXPECT().Close().AnyTimes() syncProducer.EXPECT().SendMessages(gomock.Any(), int32(1), gomock.Any()).Return(nil) - syncProducer.EXPECT().Close().AnyTimes() - - kafkaSink, err := newKafkaSinkForTestWithProducers(ctx, t, ctrl, asyncProducer, syncProducer) - require.NoError(t, err) defer cancel() go kafkaSink.Run(ctx) - err = kafkaSink.WriteBlockEvent(ddlEvent) + err := kafkaSink.WriteBlockEvent(ddlEvent) require.NoError(t, err) kafkaSink.AddDMLEvent(dmlEvent) @@ -453,21 +330,16 @@ func TestKafkaSinkDML(t *testing.T) { dmlEvent := eventHelper.DML2Event("test", "t", "insert into t values (1, 'one')") dmlEvent.PostTxnFlushed = []func(){func() { callbackCount.Add(1) }} - ctrl := gomock.NewController(t) - asyncProducer := kafka.NewMockAsyncProducer(ctrl) - syncProducer := kafka.NewMockSyncProducer(ctrl) sent := make(chan *codecCommon.Message, 1) + ctx, cancel := context.WithCancelCause(t.Context()) + kafkaSink, topicManager, asyncProducer, _ := newKafkaSinkForTest( + t, ctx, config.ProtocolOpen, &config.SinkConfig{}) + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(1), nil) asyncProducer.EXPECT().AsyncSend(gomock.Any(), kafkaSinkTestTopic, int32(0), gomock.Any()). DoAndReturn(func(_ context.Context, _ string, _ int32, message *codecCommon.Message) error { sent <- message return nil }) - asyncProducer.EXPECT().Close() - syncProducer.EXPECT().Close() - - ctx, cancel := context.WithCancelCause(t.Context()) - kafkaSink, err := newKafkaSinkForTestWithProducers(ctx, t, ctrl, asyncProducer, syncProducer) - require.NoError(t, err) defer kafkaSink.Close() runDone := make(chan error, 1) @@ -493,25 +365,31 @@ func TestKafkaSinkDML(t *testing.T) { t.Run("returns AsyncSend error unchanged", func(t *testing.T) { dmlEvent := eventHelper.DML2Event("test", "t", "insert into t values (2, 'two')") - ctrl := gomock.NewController(t) - asyncProducer := kafka.NewMockAsyncProducer(ctrl) - syncProducer := kafka.NewMockSyncProducer(ctrl) cause := errors.ErrKafkaSendMessage.GenWithStackByArgs() - asyncProducer.EXPECT().AsyncSend(gomock.Any(), kafkaSinkTestTopic, int32(0), gomock.Any()).Return(cause) - asyncProducer.EXPECT().Close() - syncProducer.EXPECT().Close() - ctx, cancel := context.WithCancel(t.Context()) defer cancel() - kafkaSink, err := newKafkaSinkForTestWithProducers(ctx, t, ctrl, asyncProducer, syncProducer) - require.NoError(t, err) + kafkaSink, topicManager, asyncProducer, _ := newKafkaSinkForTest( + t, ctx, config.ProtocolCanalJSON, &config.SinkConfig{}) + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(1), nil) + asyncProducer.EXPECT().AsyncSend(gomock.Any(), kafkaSinkTestTopic, int32(0), gomock.Any()).Return(cause) defer kafkaSink.Close() kafkaSink.AddDMLEvent(dmlEvent) - err = kafkaSink.sendDMLEvent(ctx) + err := kafkaSink.sendDMLEvent(ctx) require.Equal(t, cause, err) }) + + t.Run("returns topic manager error unchanged", func(t *testing.T) { + dmlEvent := eventHelper.DML2Event("test", "t", "insert into t values (3, 'three')") + kafkaSink, topicManager, _, _ := newKafkaSinkForTest( + t, t.Context(), config.ProtocolOpen, &config.SinkConfig{}) + cause := context.DeadlineExceeded + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(0), cause) + kafkaSink.AddDMLEvent(dmlEvent) + + require.Equal(t, cause, kafkaSink.calculateKeyPartitions(t.Context())) + }) } func TestKafkaSinkDDL(t *testing.T) { @@ -524,9 +402,10 @@ func TestKafkaSinkDDL(t *testing.T) { } t.Run("all partitions", func(t *testing.T) { - kafkaSink, topicManager, producer := newKafkaSinkForTest(t, config.ProtocolOpen, &config.SinkConfig{}) + kafkaSink, topicManager, _, syncProducer := newKafkaSinkForTest( + t, t.Context(), config.ProtocolOpen, &config.SinkConfig{}) topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(4), nil) - producer.EXPECT().SendMessages(kafkaSinkTestTopic, int32(4), gomock.Any()). + syncProducer.EXPECT().SendMessages(kafkaSinkTestTopic, int32(4), gomock.Any()). DoAndReturn(func(_ string, _ int32, message *codecCommon.Message) error { require.NotEmpty(t, message.Key) require.NotEmpty(t, message.Value) @@ -537,9 +416,10 @@ func TestKafkaSinkDDL(t *testing.T) { }) t.Run("partition zero", func(t *testing.T) { - kafkaSink, topicManager, producer := newKafkaSinkForTest(t, config.ProtocolCanalJSON, &config.SinkConfig{}) + kafkaSink, topicManager, _, syncProducer := newKafkaSinkForTest( + t, t.Context(), config.ProtocolCanalJSON, &config.SinkConfig{}) topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(4), nil) - producer.EXPECT().SendMessage(kafkaSinkTestTopic, int32(0), gomock.Any()). + syncProducer.EXPECT().SendMessage(kafkaSinkTestTopic, int32(0), gomock.Any()). DoAndReturn(func(_ string, _ int32, message *codecCommon.Message) error { require.NotEmpty(t, message.Value) return nil @@ -549,35 +429,48 @@ func TestKafkaSinkDDL(t *testing.T) { }) t.Run("topic manager error", func(t *testing.T) { - kafkaSink, topicManager, _ := newKafkaSinkForTest(t, config.ProtocolOpen, &config.SinkConfig{}) + kafkaSink, topicManager, _, _ := newKafkaSinkForTest( + t, t.Context(), config.ProtocolOpen, &config.SinkConfig{}) cause := context.DeadlineExceeded topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(0), cause) require.Equal(t, cause, kafkaSink.sendDDLEvent(ddlEvent)) }) - t.Run("producer error", func(t *testing.T) { - kafkaSink, topicManager, producer := newKafkaSinkForTest(t, config.ProtocolOpen, &config.SinkConfig{}) + t.Run("producer error marks sink abnormal", func(t *testing.T) { + kafkaSink, topicManager, _, syncProducer := newKafkaSinkForTest( + t, t.Context(), config.ProtocolOpen, &config.SinkConfig{}) cause := errors.ErrKafkaSendMessage.GenWithStackByArgs() topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(2), nil) - producer.EXPECT().SendMessages(kafkaSinkTestTopic, int32(2), gomock.Any()).Return(cause) + syncProducer.EXPECT().SendMessages(kafkaSinkTestTopic, int32(2), gomock.Any()).Return(cause) - require.Equal(t, cause, kafkaSink.sendDDLEvent(ddlEvent)) + require.Equal(t, cause, kafkaSink.WriteBlockEvent(ddlEvent)) + require.False(t, kafkaSink.IsNormal()) }) t.Run("nil encoded message", func(t *testing.T) { - kafkaSink, _, _ := newKafkaSinkForTest(t, config.ProtocolDebezium, &config.SinkConfig{}) + kafkaSink, _, _, _ := newKafkaSinkForTest( + t, t.Context(), config.ProtocolDebezium, &config.SinkConfig{}) unsupportedDDL := &commonEvent.DDLEvent{Type: byte(model.ActionNone), Query: "unsupported"} require.NoError(t, kafkaSink.sendDDLEvent(unsupportedDDL)) }) + + t.Run("unsupported block event", func(t *testing.T) { + kafkaSink, _, _, _ := newKafkaSinkForTest( + t, t.Context(), config.ProtocolOpen, &config.SinkConfig{}) + syncPoint := commonEvent.NewSyncPointEvent(common.NewDispatcherID(), 1, 1, 1) + + require.ErrorIs(t, kafkaSink.WriteBlockEvent(syncPoint), errors.ErrInvalidEventType) + }) } func TestKafkaSinkCheckpoint(t *testing.T) { t.Run("default topic without tables", func(t *testing.T) { - kafkaSink, topicManager, producer := newKafkaSinkForTest(t, config.ProtocolOpen, &config.SinkConfig{}) + kafkaSink, topicManager, _, syncProducer := newKafkaSinkForTest( + t, t.Context(), config.ProtocolOpen, &config.SinkConfig{}) topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(3), nil) - producer.EXPECT().SendMessages(kafkaSinkTestTopic, int32(3), gomock.Any()). + syncProducer.EXPECT().SendMessages(kafkaSinkTestTopic, int32(3), gomock.Any()). DoAndReturn(func(_ string, _ int32, message *codecCommon.Message) error { require.NotEmpty(t, message.Key) return nil @@ -593,7 +486,8 @@ func TestKafkaSinkCheckpoint(t *testing.T) { {Matcher: []string{"db1.t1"}, PartitionRule: "table", TopicRule: "topic-a"}, {Matcher: []string{"db2.t2"}, PartitionRule: "table", TopicRule: "topic-b"}, }} - kafkaSink, topicManager, producer := newKafkaSinkForTest(t, config.ProtocolOpen, sinkConfig) + kafkaSink, topicManager, _, syncProducer := newKafkaSinkForTest( + t, t.Context(), config.ProtocolOpen, sinkConfig) kafkaSink.SetTableSchemaStore(commonEvent.NewTableSchemaStore([]*heartbeatpb.SchemaInfo{ {SchemaName: "db1", Tables: []*heartbeatpb.TableInfo{{TableName: "t1"}}}, {SchemaName: "db2", Tables: []*heartbeatpb.TableInfo{{TableName: "t2"}}}, @@ -601,7 +495,7 @@ func TestKafkaSinkCheckpoint(t *testing.T) { partitionCounts := map[string]int32{"topic-a": 2, "topic-b": 3, kafkaSinkTestTopic: 4} for topic, partitionCount := range partitionCounts { topicManager.EXPECT().GetPartitionNum(gomock.Any(), topic).Return(partitionCount, nil) - producer.EXPECT().SendMessages(topic, partitionCount, gomock.Any()).Return(nil) + syncProducer.EXPECT().SendMessages(topic, partitionCount, gomock.Any()).Return(nil) } kafkaSink.checkpointChan <- 100 close(kafkaSink.checkpointChan) @@ -610,7 +504,8 @@ func TestKafkaSinkCheckpoint(t *testing.T) { }) t.Run("topic manager error", func(t *testing.T) { - kafkaSink, topicManager, _ := newKafkaSinkForTest(t, config.ProtocolOpen, &config.SinkConfig{}) + kafkaSink, topicManager, _, _ := newKafkaSinkForTest( + t, t.Context(), config.ProtocolOpen, &config.SinkConfig{}) cause := context.DeadlineExceeded topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(0), cause) kafkaSink.checkpointChan <- 100 @@ -618,19 +513,35 @@ func TestKafkaSinkCheckpoint(t *testing.T) { require.Equal(t, cause, kafkaSink.sendCheckpoint(t.Context())) }) - t.Run("producer error", func(t *testing.T) { - kafkaSink, topicManager, producer := newKafkaSinkForTest(t, config.ProtocolOpen, &config.SinkConfig{}) + t.Run("producer error stops active topic fan-out", func(t *testing.T) { + sinkConfig := &config.SinkConfig{DispatchRules: []*config.DispatchRule{ + {Matcher: []string{"db1.t1"}, PartitionRule: "table", TopicRule: "topic-a"}, + }} + kafkaSink, topicManager, _, syncProducer := newKafkaSinkForTest( + t, t.Context(), config.ProtocolOpen, sinkConfig) + kafkaSink.SetTableSchemaStore(commonEvent.NewTableSchemaStore([]*heartbeatpb.SchemaInfo{ + {SchemaName: "db1", Tables: []*heartbeatpb.TableInfo{{TableName: "t1"}}}, + }, common.KafkaSinkType, false)) cause := errors.ErrKafkaSendMessage.GenWithStackByArgs() - topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(2), nil) - producer.EXPECT().SendMessages(kafkaSinkTestTopic, int32(2), gomock.Any()).Return(cause) + topicManager.EXPECT().GetPartitionNum(gomock.Any(), "topic-a").Return(int32(2), nil) + syncProducer.EXPECT().SendMessages("topic-a", int32(2), gomock.Any()).Return(cause) kafkaSink.checkpointChan <- 100 require.Equal(t, cause, kafkaSink.sendCheckpoint(t.Context())) }) + t.Run("nil encoded message", func(t *testing.T) { + kafkaSink, _, _, _ := newKafkaSinkForTest( + t, t.Context(), config.ProtocolCanalJSON, &config.SinkConfig{}) + kafkaSink.checkpointChan <- 100 + close(kafkaSink.checkpointChan) + + require.NoError(t, kafkaSink.sendCheckpoint(t.Context())) + }) + t.Run("context cancellation", func(t *testing.T) { - kafkaSink, _, _ := newKafkaSinkForTest(t, config.ProtocolOpen, &config.SinkConfig{}) ctx, cancel := context.WithCancelCause(t.Context()) + kafkaSink, _, _, _ := newKafkaSinkForTest(t, ctx, config.ProtocolOpen, &config.SinkConfig{}) cause := errors.ErrKafkaSinkClosed.GenWithStackByArgs() cancel(cause) @@ -639,35 +550,47 @@ func TestKafkaSinkCheckpoint(t *testing.T) { } func newKafkaSinkForTest( - t *testing.T, - protocol config.Protocol, - sinkConfig *config.SinkConfig, -) (*sink, *topicmanager.MockTopicManager, *kafka.MockSyncProducer) { + t *testing.T, ctx context.Context, protocol config.Protocol, sinkConfig *config.SinkConfig, +) (*sink, *topicmanager.MockTopicManager, *kafka.MockAsyncProducer, *kafka.MockSyncProducer) { t.Helper() ctrl := gomock.NewController(t) changefeedID := common.NewChangefeedID4Test("test", t.Name()) + protocolName := protocol.String() + testSinkConfig := *sinkConfig + testSinkConfig.Protocol = &protocolName + sinkConfig = &testSinkConfig router, err := eventrouter.NewEventRouter(sinkConfig, kafkaSinkTestTopic, false, false) require.NoError(t, err) - encoder, err := codec.NewEventEncoder(t.Context(), codecCommon.NewConfig(protocol).WithChangefeedID(changefeedID), nil) + columnSelector, err := columnselector.New(sinkConfig) + require.NoError(t, err) + encoderConfig := codecCommon.NewConfig(protocol).WithChangefeedID(changefeedID) + encoderConfig.MaxBatchSize = 1 + encoderGroup, err := codec.NewEncoderGroup(ctx, sinkConfig, encoderConfig, nil, changefeedID) + require.NoError(t, err) + encoder, err := codec.NewEventEncoder(ctx, encoderConfig, nil) require.NoError(t, err) - statistics := metrics.NewStatistics(changefeedID, common.DefaultKeyspaceID, "sink") - t.Cleanup(statistics.Close) topicManager := topicmanager.NewMockTopicManager(ctrl) - producer := kafka.NewMockSyncProducer(ctrl) - - return &sink{ - changefeedID: changefeedID, - ddlProducer: producer, - partitionRule: helper.GetDDLDispatchRule(protocol), - protocol: protocol, - comp: components{ - encoder: encoder, - eventRouter: router, - topicManager: topicManager, - }, - statistics: statistics, - checkpointChan: make(chan uint64, 1), - ctx: t.Context(), - }, topicManager, producer + asyncProducer := kafka.NewMockAsyncProducer(ctrl) + syncProducer := kafka.NewMockSyncProducer(ctrl) + topicManager.EXPECT().Close().AnyTimes() + asyncProducer.EXPECT().Close().AnyTimes() + syncProducer.EXPECT().Close().AnyTimes() + factory := kafka.NewMockFactory(ctrl) + factory.EXPECT().AsyncProducer(gomock.Any()).Return(asyncProducer, nil) + factory.EXPECT().SyncProducer(gomock.Any()).Return(syncProducer, nil) + factory.EXPECT().MetricsCollector(nil).Return(noopMetricsCollector{}) + + kafkaSink, err := newWithComponents(ctx, changefeedID, common.DefaultKeyspaceID, protocol, components{ + encoderGroup: encoderGroup, + encoder: encoder, + columnSelector: columnSelector, + eventRouter: router, + topicManager: topicManager, + factory: factory, + }) + require.NoError(t, err) + t.Cleanup(kafkaSink.statistics.Close) + + return kafkaSink, topicManager, asyncProducer, syncProducer } From 9f0599cb05871502a73d422dfd657c9a49dc0e8f Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Fri, 14 Aug 2026 01:54:02 +0800 Subject: [PATCH 3/6] fix unit test --- downstreamadapter/sink/kafka/sink_test.go | 25 +++++++++++++++-------- 1 file changed, 16 insertions(+), 9 deletions(-) diff --git a/downstreamadapter/sink/kafka/sink_test.go b/downstreamadapter/sink/kafka/sink_test.go index b46d807dcd..5e098b67a6 100644 --- a/downstreamadapter/sink/kafka/sink_test.go +++ b/downstreamadapter/sink/kafka/sink_test.go @@ -134,7 +134,6 @@ func TestKafkaSinkRunReturnsAsyncProducerError(t *testing.T) { producerErr := errors.ErrKafkaSendMessage.GenWithStackByArgs() kafkaSink, _, asyncProducer, _ := newKafkaSinkForTest(t, ctx, config.ProtocolOpen, &config.SinkConfig{}) asyncProducer.EXPECT().AsyncRunCallback(gomock.Any()).Return(producerErr) - defer kafkaSink.Close() err := kafkaSink.Run(ctx) @@ -325,7 +324,7 @@ func TestKafkaSinkDML(t *testing.T) { eventHelper.Tk().MustExec("use test") require.NotNil(t, eventHelper.DDL2Job("create table t (id int primary key, name varchar(32))")) - t.Run("forwards routed message and waits for producer callback", func(t *testing.T) { + t.Run("routes DML event without invoking callback synchronously", func(t *testing.T) { var callbackCount atomic.Int64 dmlEvent := eventHelper.DML2Event("test", "t", "insert into t values (1, 'one')") dmlEvent.PostTxnFlushed = []func(){func() { callbackCount.Add(1) }} @@ -340,7 +339,6 @@ func TestKafkaSinkDML(t *testing.T) { sent <- message return nil }) - defer kafkaSink.Close() runDone := make(chan error, 1) go func() { runDone <- kafkaSink.sendDMLEvent(ctx) }() @@ -353,13 +351,18 @@ func TestKafkaSinkDML(t *testing.T) { require.Equal(t, 1, message.GetRowsCount()) require.NotNil(t, message.Callback) require.Zero(t, callbackCount.Load()) - case <-time.After(10 * time.Second): + case <-time.After(3 * time.Second): t.Fatal("timed out waiting for Kafka Sink to send the DML message") } cause := errors.ErrKafkaSinkClosed.GenWithStackByArgs() cancel(cause) - require.Equal(t, cause, <-runDone) + select { + case err := <-runDone: + require.Equal(t, cause, err) + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for Kafka Sink workers to exit") + } require.Zero(t, callbackCount.Load()) }) @@ -372,7 +375,6 @@ func TestKafkaSinkDML(t *testing.T) { t, ctx, config.ProtocolCanalJSON, &config.SinkConfig{}) topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(1), nil) asyncProducer.EXPECT().AsyncSend(gomock.Any(), kafkaSinkTestTopic, int32(0), gomock.Any()).Return(cause) - defer kafkaSink.Close() kafkaSink.AddDMLEvent(dmlEvent) err := kafkaSink.sendDMLEvent(ctx) @@ -492,6 +494,8 @@ func TestKafkaSinkCheckpoint(t *testing.T) { {SchemaName: "db1", Tables: []*heartbeatpb.TableInfo{{TableName: "t1"}}}, {SchemaName: "db2", Tables: []*heartbeatpb.TableInfo{{TableName: "t2"}}}, }, common.KafkaSinkType, false)) + // The checkpoint must be fanned out to every active topic: the two + // rule topics and the default topic. partitionCounts := map[string]int32{"topic-a": 2, "topic-b": 3, kafkaSinkTestTopic: 4} for topic, partitionCount := range partitionCounts { topicManager.EXPECT().GetPartitionNum(gomock.Any(), topic).Return(partitionCount, nil) @@ -523,8 +527,11 @@ func TestKafkaSinkCheckpoint(t *testing.T) { {SchemaName: "db1", Tables: []*heartbeatpb.TableInfo{{TableName: "t1"}}}, }, common.KafkaSinkType, false)) cause := errors.ErrKafkaSendMessage.GenWithStackByArgs() - topicManager.EXPECT().GetPartitionNum(gomock.Any(), "topic-a").Return(int32(2), nil) - syncProducer.EXPECT().SendMessages("topic-a", int32(2), gomock.Any()).Return(cause) + // Fail whichever topic the fan-out reaches first: sendCheckpoint must + // return the error and stop, so exactly one GetPartitionNum and one + // SendMessages call are expected regardless of the topic order. + topicManager.EXPECT().GetPartitionNum(gomock.Any(), gomock.Any()).Return(int32(2), nil) + syncProducer.EXPECT().SendMessages(gomock.Any(), int32(2), gomock.Any()).Return(cause) kafkaSink.checkpointChan <- 100 require.Equal(t, cause, kafkaSink.sendCheckpoint(t.Context())) @@ -590,7 +597,7 @@ func newKafkaSinkForTest( factory: factory, }) require.NoError(t, err) - t.Cleanup(kafkaSink.statistics.Close) + t.Cleanup(kafkaSink.Close) return kafkaSink, topicManager, asyncProducer, syncProducer } From d3a4ddfc869e084652427372be0d4a7aa805cde2 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Fri, 14 Aug 2026 11:31:07 +0800 Subject: [PATCH 4/6] fix unit test --- downstreamadapter/sink/kafka/sink_test.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/downstreamadapter/sink/kafka/sink_test.go b/downstreamadapter/sink/kafka/sink_test.go index 5e098b67a6..280bdf3075 100644 --- a/downstreamadapter/sink/kafka/sink_test.go +++ b/downstreamadapter/sink/kafka/sink_test.go @@ -324,7 +324,7 @@ func TestKafkaSinkDML(t *testing.T) { eventHelper.Tk().MustExec("use test") require.NotNil(t, eventHelper.DDL2Job("create table t (id int primary key, name varchar(32))")) - t.Run("routes DML event without invoking callback synchronously", func(t *testing.T) { + t.Run("routes DML event and forwards producer callback", func(t *testing.T) { var callbackCount atomic.Int64 dmlEvent := eventHelper.DML2Event("test", "t", "insert into t values (1, 'one')") dmlEvent.PostTxnFlushed = []func(){func() { callbackCount.Add(1) }} @@ -351,6 +351,8 @@ func TestKafkaSinkDML(t *testing.T) { require.Equal(t, 1, message.GetRowsCount()) require.NotNil(t, message.Callback) require.Zero(t, callbackCount.Load()) + message.Callback() + require.Equal(t, int64(1), callbackCount.Load()) case <-time.After(3 * time.Second): t.Fatal("timed out waiting for Kafka Sink to send the DML message") } @@ -363,7 +365,7 @@ func TestKafkaSinkDML(t *testing.T) { case <-time.After(3 * time.Second): t.Fatal("timed out waiting for Kafka Sink workers to exit") } - require.Zero(t, callbackCount.Load()) + require.Equal(t, int64(1), callbackCount.Load()) }) t.Run("returns AsyncSend error unchanged", func(t *testing.T) { From fd29ed0a0c0839a350d926a55aedf7a83dee30d8 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Fri, 14 Aug 2026 12:18:48 +0800 Subject: [PATCH 5/6] fix all --- .../{admin_mock.go => sarama_admin_mock.go} | 0 .../{admin_test.go => sarama_admin_test.go} | 0 pkg/sink/kafka/scram_client_test.go | 20 ++++--------------- scripts/generate-mock.sh | 2 +- 4 files changed, 5 insertions(+), 17 deletions(-) rename pkg/sink/kafka/{admin_mock.go => sarama_admin_mock.go} (100%) rename pkg/sink/kafka/{admin_test.go => sarama_admin_test.go} (100%) diff --git a/pkg/sink/kafka/admin_mock.go b/pkg/sink/kafka/sarama_admin_mock.go similarity index 100% rename from pkg/sink/kafka/admin_mock.go rename to pkg/sink/kafka/sarama_admin_mock.go diff --git a/pkg/sink/kafka/admin_test.go b/pkg/sink/kafka/sarama_admin_test.go similarity index 100% rename from pkg/sink/kafka/admin_test.go rename to pkg/sink/kafka/sarama_admin_test.go diff --git a/pkg/sink/kafka/scram_client_test.go b/pkg/sink/kafka/scram_client_test.go index ae0829f2a7..73168be5e6 100644 --- a/pkg/sink/kafka/scram_client_test.go +++ b/pkg/sink/kafka/scram_client_test.go @@ -16,21 +16,19 @@ package kafka import ( "testing" - "github.com/IBM/sarama" "github.com/stretchr/testify/require" "github.com/xdg/scram" ) -func TestSCRAMClientGeneratorHandshake(t *testing.T) { +func TestSCRAMClientHandshake(t *testing.T) { t.Parallel() tests := []struct { name string - mechanism saslMechanism hashGenerator scram.HashGeneratorFcn }{ - {name: "SHA-256", mechanism: scram256Mechanism, hashGenerator: sha256HashGenerator}, - {name: "SHA-512", mechanism: scram512Mechanism, hashGenerator: sha512HashGenerator}, + {name: "SHA-256", hashGenerator: sha256HashGenerator}, + {name: "SHA-512", hashGenerator: sha512HashGenerator}, } for _, test := range tests { @@ -41,17 +39,7 @@ func TestSCRAMClientGeneratorHandshake(t *testing.T) { username = "user" password = "password" ) - options := NewOptions() - options.sasl = &saslConfig{ - user: username, - password: password, - mechanism: test.mechanism, - } - config := sarama.NewConfig() - require.NoError(t, completeSaramaSASLConfig(t.Context(), config, options)) - require.NotNil(t, config.Net.SASL.SCRAMClientGeneratorFunc) - - client := config.Net.SASL.SCRAMClientGeneratorFunc() + client := &xdgSCRAMClient{HashGeneratorFcn: test.hashGenerator} require.NoError(t, client.Begin(username, password, "")) credentialClient, err := test.hashGenerator.NewClient(username, password, "") diff --git a/scripts/generate-mock.sh b/scripts/generate-mock.sh index d7c654cf06..f1db9a6a6a 100755 --- a/scripts/generate-mock.sh +++ b/scripts/generate-mock.sh @@ -36,7 +36,7 @@ fi "$MOCKGEN" -source pkg/sink/codec/simple/marshaller.go -destination pkg/sink/codec/simple/mock/marshaller.go "$MOCKGEN" -source pkg/sink/kafka/admin_client.go -destination pkg/sink/kafka/admin_client_mock.go -package kafka "$MOCKGEN" -source pkg/sink/kafka/factory.go -destination pkg/sink/kafka/factory_mock.go -package kafka -"$MOCKGEN" -source pkg/sink/kafka/admin.go -destination pkg/sink/kafka/admin_mock.go -package kafka +"$MOCKGEN" -source pkg/sink/kafka/admin.go -destination pkg/sink/kafka/sarama_admin_mock.go -package kafka "$MOCKGEN" -source pkg/sink/kafka/sarama_sync_producer.go -destination pkg/sink/kafka/sarama_sync_producer_mock.go -package kafka "$MOCKGEN" -source downstreamadapter/sink/topicmanager/topic_manager.go -destination downstreamadapter/sink/topicmanager/topic_manager_mock.go -package topicmanager "$MOCKGEN" -source pkg/keyspace/keyspace_manager.go -destination pkg/keyspace/keyspace_manager_mock.go -package keyspace From abe91bab7a37154dfa186f9c22fd2613966dbeed Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Fri, 14 Aug 2026 12:27:18 +0800 Subject: [PATCH 6/6] mv oath2 to sarama dedicated file name --- .../{oauth2_token_provider.go => sarama_oauth2_token_provider.go} | 0 ...oken_provider_test.go => sarama_oauth2_token_provider_test.go} | 0 2 files changed, 0 insertions(+), 0 deletions(-) rename pkg/sink/kafka/{oauth2_token_provider.go => sarama_oauth2_token_provider.go} (100%) rename pkg/sink/kafka/{oauth2_token_provider_test.go => sarama_oauth2_token_provider_test.go} (100%) diff --git a/pkg/sink/kafka/oauth2_token_provider.go b/pkg/sink/kafka/sarama_oauth2_token_provider.go similarity index 100% rename from pkg/sink/kafka/oauth2_token_provider.go rename to pkg/sink/kafka/sarama_oauth2_token_provider.go diff --git a/pkg/sink/kafka/oauth2_token_provider_test.go b/pkg/sink/kafka/sarama_oauth2_token_provider_test.go similarity index 100% rename from pkg/sink/kafka/oauth2_token_provider_test.go rename to pkg/sink/kafka/sarama_oauth2_token_provider_test.go