diff --git a/downstreamadapter/sink/kafka/sink_test.go b/downstreamadapter/sink/kafka/sink_test.go index f4e7860d4c..280bdf3075 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,8 +24,8 @@ 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" @@ -34,6 +33,7 @@ import ( "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" ) @@ -128,130 +128,14 @@ 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()) @@ -308,9 +192,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( @@ -324,16 +208,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) @@ -356,3 +235,371 @@ 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("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) }} + + 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 + }) + + 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()) + 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") + } + + cause := errors.ErrKafkaSinkClosed.GenWithStackByArgs() + cancel(cause) + 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.Equal(t, int64(1), callbackCount.Load()) + }) + + t.Run("returns AsyncSend error unchanged", func(t *testing.T) { + dmlEvent := eventHelper.DML2Event("test", "t", "insert into t values (2, 'two')") + cause := errors.ErrKafkaSendMessage.GenWithStackByArgs() + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + 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) + + kafkaSink.AddDMLEvent(dmlEvent) + 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) { + 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, _, syncProducer := newKafkaSinkForTest( + t, t.Context(), config.ProtocolOpen, &config.SinkConfig{}) + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(4), nil) + 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) + return nil + }) + + require.NoError(t, kafkaSink.sendDDLEvent(ddlEvent)) + }) + + t.Run("partition zero", func(t *testing.T) { + kafkaSink, topicManager, _, syncProducer := newKafkaSinkForTest( + t, t.Context(), config.ProtocolCanalJSON, &config.SinkConfig{}) + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(4), nil) + syncProducer.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, 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 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) + syncProducer.EXPECT().SendMessages(kafkaSinkTestTopic, int32(2), gomock.Any()).Return(cause) + + require.Equal(t, cause, kafkaSink.WriteBlockEvent(ddlEvent)) + require.False(t, kafkaSink.IsNormal()) + }) + + t.Run("nil encoded message", func(t *testing.T) { + 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, _, syncProducer := newKafkaSinkForTest( + t, t.Context(), config.ProtocolOpen, &config.SinkConfig{}) + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(3), nil) + syncProducer.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, _, 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"}}}, + }, 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) + syncProducer.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, t.Context(), 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 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() + // 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())) + }) + + 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) { + ctx, cancel := context.WithCancelCause(t.Context()) + kafkaSink, _, _, _ := newKafkaSinkForTest(t, ctx, config.ProtocolOpen, &config.SinkConfig{}) + cause := errors.ErrKafkaSinkClosed.GenWithStackByArgs() + cancel(cause) + + require.Equal(t, cause, kafkaSink.sendCheckpoint(ctx)) + }) +} + +func newKafkaSinkForTest( + 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) + 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) + topicManager := topicmanager.NewMockTopicManager(ctrl) + 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.Close) + + return kafkaSink, topicManager, asyncProducer, syncProducer +} 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/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/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 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 e4f0a36b51..f1db9a6a6a 100755 --- a/scripts/generate-mock.sh +++ b/scripts/generate-mock.sh @@ -36,8 +36,9 @@ 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 "$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