diff --git a/downstreamadapter/sink/eventrouter/topic/expression.go b/downstreamadapter/sink/eventrouter/topic/expression.go index bc60085974..453ba423a3 100644 --- a/downstreamadapter/sink/eventrouter/topic/expression.go +++ b/downstreamadapter/sink/eventrouter/topic/expression.go @@ -68,15 +68,14 @@ func (e Expression) validate() error { return nil } - return errors.ErrKafkaInvalidTopicExpression.GenWithStackByArgs(e) + return errors.ErrKafkaInvalidConfig.GenWithStack("invalid topic expression: %s", e) } // ValidateForAvro checks whether topic pattern is {schema}_{table}, the only allowed func (e Expression) validateForAvro() error { if ok := avroTopicNameRE.MatchString(string(e)); !ok { - return errors.ErrKafkaInvalidTopicExpression.GenWithStackByArgs(e, - "topic rule for Avro must contain {schema} and {table}", - ) + return errors.ErrKafkaInvalidConfig.GenWithStack( + "invalid topic expression %s: topic rule for Avro must contain {schema} and {table}", e) } return nil diff --git a/downstreamadapter/sink/eventrouter/topic/expression_test.go b/downstreamadapter/sink/eventrouter/topic/expression_test.go index 82c75f265a..78151ad23d 100644 --- a/downstreamadapter/sink/eventrouter/topic/expression_test.go +++ b/downstreamadapter/sink/eventrouter/topic/expression_test.go @@ -265,11 +265,11 @@ func TestInvalidExpression(t *testing.T) { topicExpr := Expression(invalidExpr) err := topicExpr.validate() - require.ErrorIs(t, err, errors.ErrKafkaInvalidTopicExpression) + require.ErrorIs(t, err, errors.ErrKafkaInvalidConfig) require.ErrorContains(t, err, invalidExpr) err = topicExpr.validateForAvro() - require.ErrorIs(t, err, errors.ErrKafkaInvalidTopicExpression) + require.ErrorIs(t, err, errors.ErrKafkaInvalidConfig) require.ErrorContains(t, err, "Avro") require.ErrorContains(t, err, invalidExpr) } diff --git a/downstreamadapter/sink/kafka/helper.go b/downstreamadapter/sink/kafka/helper.go index c45b7eb10f..c66b62f96e 100644 --- a/downstreamadapter/sink/kafka/helper.go +++ b/downstreamadapter/sink/kafka/helper.go @@ -21,18 +21,17 @@ import ( "github.com/pingcap/ticdc/downstreamadapter/sink/eventrouter" "github.com/pingcap/ticdc/downstreamadapter/sink/helper" "github.com/pingcap/ticdc/downstreamadapter/sink/topicmanager" - commonType "github.com/pingcap/ticdc/pkg/common" + "github.com/pingcap/ticdc/pkg/common" "github.com/pingcap/ticdc/pkg/config" - "github.com/pingcap/ticdc/pkg/errors" "github.com/pingcap/ticdc/pkg/sink/codec" - "github.com/pingcap/ticdc/pkg/sink/codec/common" + codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/pingcap/ticdc/pkg/sink/kafka" "github.com/pingcap/tidb/br/pkg/utils" ) type components struct { encoderGroup codec.EncoderGroup - encoder common.EventEncoder + encoder codecCommon.EventEncoder columnSelector *columnselector.ColumnSelectors eventRouter *eventrouter.EventRouter topicManager topicmanager.TopicManager @@ -131,10 +130,11 @@ func newKafkaSinkComponentWithFactory(ctx context.Context, func newKafkaSinkComponent( ctx context.Context, - changefeedID commonType.ChangeFeedID, + changefeedID common.ChangeFeedID, sinkURI *url.URL, sinkConfig *config.SinkConfig, ) (components, config.Protocol, error) { +<<<<<<< HEAD return newKafkaSinkComponentWithFactory(ctx, changefeedID, sinkURI, sinkConfig, kafka.NewSaramaFactory) } @@ -145,4 +145,88 @@ func newKafkaSinkComponentForTest( sinkConfig *config.SinkConfig, ) (components, config.Protocol, error) { return newKafkaSinkComponentWithFactory(ctx, changefeedID, sinkURI, sinkConfig, kafka.NewMockFactory) +======= + var ( + comp components + err error + ) + // must release resources when error occurs. + defer func() { + if err != nil { + comp.close() + } + }() + protocol, err := helper.GetProtocol(utils.GetOrZero(sinkConfig.Protocol)) + if err != nil { + return comp, config.ProtocolUnknown, err + } + + topic, err := helper.GetTopic(sinkURI) + if err != nil { + return comp, protocol, err + } + + options := kafka.NewOptions() + if err = options.Apply(changefeedID, sinkURI, sinkConfig); err != nil { + return comp, protocol, err + } + options.Topic = topic + + comp.factory, err = kafka.NewSaramaFactory(ctx, options, changefeedID) + if err != nil { + return comp, protocol, err + } + + isAvroLike := protocol == config.ProtocolAvro || protocol == config.ProtocolDebeziumAvro + comp.eventRouter, err = eventrouter.NewEventRouter( + sinkConfig, topic, false, isAvroLike) + if err != nil { + return comp, protocol, err + } + + comp.columnSelector, err = columnselector.New(sinkConfig) + if err != nil { + return comp, protocol, err + } + + encoderConfig, err := helper.GetEncoderConfig( + changefeedID, sinkURI, protocol, sinkConfig, + options.MaxMessageBytes, options.MaxBatchedBytes, + ) + if err != nil { + return comp, protocol, err + } + + comp.claimCheck, err = claimcheck.New(ctx, encoderConfig.LargeMessageHandle, changefeedID) + if err != nil { + return comp, protocol, err + } + + comp.encoderGroup, err = codec.NewEncoderGroup(ctx, sinkConfig, encoderConfig, comp.claimCheck, changefeedID) + if err != nil { + return comp, protocol, err + } + + comp.encoder, err = codec.NewEventEncoder(ctx, encoderConfig, comp.claimCheck) + if err != nil { + return comp, protocol, err + } + + comp.adminClient, err = comp.factory.AdminClient(ctx) + if err != nil { + return comp, protocol, err + } + + comp.topicManager, err = topicmanager.GetTopicManagerAndTryCreateTopic( + ctx, + changefeedID, + topic, + options.DeriveTopicConfig(), + comp.adminClient, + ) + if err != nil { + return comp, protocol, err + } + return comp, protocol, nil +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index e6d1038b26..e7ee8cd08a 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -20,12 +20,17 @@ import ( "github.com/pingcap/log" "github.com/pingcap/ticdc/downstreamadapter/sink/helper" - commonType "github.com/pingcap/ticdc/pkg/common" + "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" +<<<<<<< HEAD "github.com/pingcap/ticdc/pkg/sink/codec/common" +======= + "github.com/pingcap/ticdc/pkg/sink/codec" + codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) "github.com/pingcap/ticdc/pkg/sink/kafka" "github.com/pingcap/ticdc/pkg/util" "github.com/pingcap/ticdc/utils/chann" @@ -40,7 +45,7 @@ const ( ) type sink struct { - changefeedID commonType.ChangeFeedID + changefeedID common.ChangeFeedID dmlProducer kafka.AsyncProducer ddlProducer kafka.SyncProducer @@ -63,10 +68,11 @@ type sink struct { ctx context.Context } -func (s *sink) SinkType() commonType.SinkType { - return commonType.KafkaSinkType +func (s *sink) SinkType() common.SinkType { + return common.KafkaSinkType } +<<<<<<< HEAD func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url.URL, sinkConfig *config.SinkConfig) error { comp, _, err := newKafkaSinkComponent(ctx, changefeedID, uri, sinkConfig) defer comp.close() @@ -75,17 +81,108 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url. func New( ctx context.Context, changefeedID commonType.ChangeFeedID, sinkURI *url.URL, sinkConfig *config.SinkConfig, +======= +func Verify(ctx context.Context, changefeedID common.ChangeFeedID, uri *url.URL, sinkConfig *config.SinkConfig) error { + protocol, err := helper.GetProtocol(util.GetOrZero(sinkConfig.Protocol)) + if err != nil { + return err + } + + topic, err := helper.GetTopic(uri) + if err != nil { + return err + } + + options := kafka.NewOptions() + if err = options.Apply(changefeedID, uri, sinkConfig); err != nil { + return err + } + options.Topic = topic + + encoderConfig, err := helper.GetEncoderConfig( + changefeedID, uri, protocol, sinkConfig, + options.MaxMessageBytes, options.MaxBatchedBytes, + ) + if err != nil { + return err + } + + claimCheck, err := claimcheck.New(ctx, encoderConfig.LargeMessageHandle, changefeedID) + if err != nil { + return err + } + defer claimCheck.Close() + + isAvroLike := protocol == config.ProtocolAvro || protocol == config.ProtocolDebeziumAvro + if _, err = eventrouter.NewEventRouter(sinkConfig, topic, false, isAvroLike); err != nil { + return err + } + + if _, err = columnselector.New(sinkConfig); err != nil { + return err + } + + factory, err := kafka.NewSaramaFactory(ctx, options, changefeedID) + if err != nil { + return err + } + + adminClient, err := factory.AdminClient(ctx) + if err != nil { + return err + } + defer adminClient.Close() + + topics, err := adminClient.GetTopicsMeta([]string{topic}, false) + if err != nil { + return err + } + if _, exists := topics[topic]; !exists { + topicConfig := options.DeriveTopicConfig() + if !topicConfig.AutoCreate { + return errors.ErrKafkaInvalidConfig.GenWithStack("`auto-create-topic` is false, and %s not found", topic) + } + if err = topicConfig.ValidateReplicationFactor(adminClient); err != nil { + return err + } + + // the topic is not created, only validate. + err = adminClient.CreateTopic(&kafka.TopicDetail{ + Name: topic, + NumPartitions: topicConfig.PartitionNum, + ReplicationFactor: topicConfig.ReplicationFactor, + }, true) + if err != nil { + return err + } + } + + _, err = codec.NewEventEncoder(ctx, encoderConfig, claimCheck) + if err != nil { + return err + } + return nil +} + +func New( + ctx context.Context, changefeedID common.ChangeFeedID, sinkURI *url.URL, sinkConfig *config.SinkConfig, keyspaceID uint32, +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) ) (*sink, error) { comp, protocol, err := newKafkaSinkComponent(ctx, changefeedID, sinkURI, sinkConfig) if err != nil { - return nil, errors.Trace(err) + return nil, err } return newWithComponents(ctx, changefeedID, protocol, comp) } func newWithComponents( ctx context.Context, +<<<<<<< HEAD changefeedID commonType.ChangeFeedID, +======= + changefeedID common.ChangeFeedID, + keyspaceID uint32, +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) protocol config.Protocol, comp components, ) (*sink, error) { @@ -153,7 +250,7 @@ func (s *sink) Run(ctx context.Context) error { }) err := g.Wait() s.isNormal.Store(false) - return errors.Trace(err) + return err } func (s *sink) IsNormal() bool { @@ -224,7 +321,7 @@ func (s *sink) calculateKeyPartitions(ctx context.Context) error { for { select { case <-ctx.Done(): - return errors.Trace(ctx.Err()) + return context.Cause(ctx) default: event, ok := s.eventChan.Get() if !ok { @@ -242,6 +339,7 @@ func (s *sink) calculateKeyPartitions(ctx context.Context) error { } partitionGenerator := s.comp.eventRouter.GetPartitionGenerator(schema, table) +<<<<<<< HEAD selector := s.comp.columnSelector.Get(schema, table) rowsCount := uint64(event.Len()) events := make([]*commonEvent.MQRowEvent, 0, rowsCount) @@ -277,6 +375,12 @@ func (s *sink) calculateKeyPartitions(ctx context.Context) error { Checksum: row.Checksum, }, }) +======= + selector := s.comp.columnSelector.GetForTableInfo(event.TableInfo) + events, err := helper.NewMQRowEvents(event, topic, partitionNum, partitionGenerator, selector) + if err != nil { + return err +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } s.rowChan.Push(events...) } @@ -287,7 +391,7 @@ func (s *sink) nonBatchEncodeRun(ctx context.Context) error { for { select { case <-ctx.Done(): - return errors.Trace(ctx.Err()) + return context.Cause(ctx) default: event, ok := s.rowChan.Get() if !ok { @@ -380,7 +484,7 @@ func (s *sink) sendMessages(ctx context.Context) error { for { select { case <-ctx.Done(): - return errors.Trace(ctx.Err()) + return context.Cause(ctx) case future, ok := <-outCh: if !ok { log.Info("kafka sink encoder's output channel closed", @@ -429,7 +533,7 @@ func (s *sink) sendDDLEvent(event *commonEvent.DDLEvent) error { zap.Stringer("changefeed", s.changefeedID)) continue } - common.SetDDLMessageLogInfo(message, e) + codecCommon.SetDDLMessageLogInfo(message, e) topic := s.comp.eventRouter.GetTopicForDDL(e) // Notice: We must call GetPartitionNum here, // which will be responsible for automatically creating topics when they don't exist. @@ -479,14 +583,14 @@ func (s *sink) sendCheckpoint(ctx context.Context) error { }() var ( - msg *common.Message + msg *codecCommon.Message partitionNum int32 err error ) for { select { case <-ctx.Done(): - return errors.Trace(ctx.Err()) + return context.Cause(ctx) case ts, ok := <-s.checkpointChan: if !ok { log.Warn("kafka sink checkpoint channel closed", @@ -503,7 +607,7 @@ func (s *sink) sendCheckpoint(ctx context.Context) error { if msg == nil { continue } - common.SetCheckpointMessageLogInfo(msg, ts) + codecCommon.SetCheckpointMessageLogInfo(msg, ts) tableNames := s.getAllTableNames(ts) // NOTICE: When there are no tables to replicate, diff --git a/downstreamadapter/sink/kafka/sink_test.go b/downstreamadapter/sink/kafka/sink_test.go index 0bb4708f58..319008d618 100644 --- a/downstreamadapter/sink/kafka/sink_test.go +++ b/downstreamadapter/sink/kafka/sink_test.go @@ -25,13 +25,107 @@ import ( "github.com/pingcap/ticdc/pkg/common" commonEvent "github.com/pingcap/ticdc/pkg/common/event" "github.com/pingcap/ticdc/pkg/config" +<<<<<<< HEAD "github.com/pingcap/ticdc/pkg/metrics" +======= + "github.com/pingcap/ticdc/pkg/errors" + "github.com/pingcap/ticdc/pkg/sink/codec" + codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) "github.com/pingcap/ticdc/pkg/sink/kafka" "github.com/pingcap/ticdc/utils/chann" "github.com/stretchr/testify/require" "go.uber.org/atomic" ) +<<<<<<< HEAD +======= +const kafkaSinkTestTopic = "mock_topic" + +func TestSinkWorkersReturnContextError(t *testing.T) { + contexts := []struct { + name string + newContext func() (context.Context, context.CancelFunc) + cause error + }{ + { + name: "canceled", + newContext: func() (context.Context, context.CancelFunc) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + return ctx, cancel + }, + cause: context.Canceled, + }, + { + name: "deadline exceeded", + newContext: func() (context.Context, context.CancelFunc) { + return context.WithTimeout(context.Background(), 0) + }, + cause: context.DeadlineExceeded, + }, + } + workers := []struct { + name string + run func(*sink, context.Context) error + }{ + {name: "calculate key partitions", run: (*sink).calculateKeyPartitions}, + {name: "non batch encode", run: (*sink).nonBatchEncodeRun}, + {name: "checkpoint", run: (*sink).sendCheckpoint}, + } + + for _, worker := range workers { + for _, contextCase := range contexts { + t.Run(worker.name+"/"+contextCase.name, func(t *testing.T) { + ctx, cancel := contextCase.newContext() + defer cancel() + + err := worker.run(&sink{}, ctx) + + require.ErrorIs(t, err, contextCase.cause) + }) + } + } +} + +func TestVerifyInvalidConfig(t *testing.T) { + broker := sarama.NewMockBroker(t, 1) + defer broker.Close() + broker.SetHandlerByMap(map[string]sarama.MockResponse{ + "ApiVersionsRequest": sarama.NewMockApiVersionsResponse(t).SetApiKeys( + []sarama.ApiVersionsResponseKey{ + {ApiKey: 0}, + {ApiKey: 1}, + {ApiKey: 2}, + {ApiKey: 3, MaxVersion: 9}, + }), + "MetadataRequest": sarama.NewMockMetadataResponse(t). + SetController(broker.BrokerID()). + SetBroker(broker.Addr(), broker.BrokerID()). + SetLeader(kafkaSinkTestTopic, 0, broker.BrokerID()), + "DescribeConfigsRequest": sarama.NewMockDescribeConfigsResponse(t), + }) + + schemaRegistry := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + http.Error(w, "invalid response", http.StatusInternalServerError) + })) + defer schemaRegistry.Close() + + avroProtocol := config.ProtocolAvro.String() + sinkConfig := &config.SinkConfig{ + Protocol: &avroProtocol, + SchemaRegistry: &schemaRegistry.URL, + } + sinkURI, err := url.Parse("kafka://" + broker.Addr() + "/" + kafkaSinkTestTopic + + "?required-acks=1&kafka-version=2.4.0") + require.NoError(t, err) + + changefeedID := common.NewChangefeedID4Test("test", "verify-invalid-config") + err = Verify(context.Background(), changefeedID, sinkURI, sinkConfig) + require.ErrorContains(t, err, "ErrAvroSchemaAPIError") +} + +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) func newKafkaSinkForTestWithProducers(ctx context.Context, asyncProducer kafka.AsyncProducer, syncProducer kafka.SyncProducer, @@ -94,12 +188,33 @@ func newKafkaSinkForTestWithProducers(ctx context.Context, isNormal: atomic.NewBool(true), ctx: ctx, } - go s.Run(ctx) return s, nil } +<<<<<<< HEAD func newKafkaSinkForTest(ctx context.Context) (*sink, error) { return newKafkaSinkForTestWithProducers(ctx, nil, 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) + 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) + + require.ErrorIs(t, err, errors.ErrKafkaSendMessage) + require.False(t, kafkaSink.IsNormal()) +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } func TestKafkaSinkBasicFunctionality(t *testing.T) { @@ -153,9 +268,34 @@ func TestKafkaSinkBasicFunctionality(t *testing.T) { dmlEvent.CommitTs = 2 ctx, cancel := context.WithCancel(context.Background()) +<<<<<<< HEAD kafkaSink, err := newKafkaSinkForTest(ctx) +======= + ctrl := gomock.NewController(t) + asyncProducer := kafka.NewMockAsyncProducer(ctrl) + syncProducer := kafka.NewMockSyncProducer(ctrl) + asyncProducer.EXPECT().AsyncRunCallback(gomock.Any()).Return(nil).AnyTimes() + asyncProducer.EXPECT().AsyncSend(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()). + DoAndReturn(func( + _ context.Context, + _ string, + _ int32, + message *codecCommon.Message, + ) error { + if message.Callback != nil { + message.Callback() + } + 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) +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) require.NoError(t, err) defer cancel() + go kafkaSink.Run(ctx) kafkaSink.ddlProducer.(*kafka.MockSaramaSyncProducer).SyncProducer.ExpectSendMessageAndSucceed() err = kafkaSink.WriteBlockEvent(ddlEvent) diff --git a/downstreamadapter/sink/pulsar/helper.go b/downstreamadapter/sink/pulsar/helper.go index acc1cddd38..cb80f8b0ff 100644 --- a/downstreamadapter/sink/pulsar/helper.go +++ b/downstreamadapter/sink/pulsar/helper.go @@ -23,11 +23,11 @@ import ( "github.com/pingcap/ticdc/downstreamadapter/sink/eventrouter" "github.com/pingcap/ticdc/downstreamadapter/sink/helper" "github.com/pingcap/ticdc/downstreamadapter/sink/topicmanager" - commonType "github.com/pingcap/ticdc/pkg/common" + "github.com/pingcap/ticdc/pkg/common" "github.com/pingcap/ticdc/pkg/config" "github.com/pingcap/ticdc/pkg/errors" "github.com/pingcap/ticdc/pkg/sink/codec" - "github.com/pingcap/ticdc/pkg/sink/codec/common" + codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/pingcap/ticdc/pkg/sink/pulsar" putil "github.com/pingcap/ticdc/pkg/util" "go.uber.org/zap" @@ -36,7 +36,7 @@ import ( type component struct { config *config.PulsarConfig encoderGroup codec.EncoderGroup - encoder common.EventEncoder + encoder codecCommon.EventEncoder columnSelector *columnselector.ColumnSelectors eventRouter *eventrouter.EventRouter topicManager topicmanager.TopicManager @@ -54,7 +54,7 @@ func (c component) close() { func newPulsarSinkComponent( ctx context.Context, - changefeedID commonType.ChangeFeedID, + changefeedID common.ChangeFeedID, sinkURI *url.URL, sinkConfig *config.SinkConfig, ) (component, config.Protocol, error) { @@ -63,7 +63,7 @@ func newPulsarSinkComponent( func newPulsarSinkComponentForTest( ctx context.Context, - changefeedID commonType.ChangeFeedID, + changefeedID common.ChangeFeedID, sinkURI *url.URL, sinkConfig *config.SinkConfig, ) (component, config.Protocol, error) { @@ -71,7 +71,7 @@ func newPulsarSinkComponentForTest( } func newPulsarSinkComponentWithFactory(ctx context.Context, - changefeedID commonType.ChangeFeedID, + changefeedID common.ChangeFeedID, sinkURI *url.URL, sinkConfig *config.SinkConfig, factoryCreator pulsar.FactoryCreator, @@ -98,7 +98,7 @@ func newPulsarSinkComponentWithFactory(ctx context.Context, pulsarComponent.client, err = factoryCreator(pulsarComponent.config, changefeedID, sinkConfig) if err != nil { - return pulsarComponent, protocol, errors.WrapError(errors.ErrKafkaNewProducer, err) + return pulsarComponent, protocol, errors.WrapError(errors.ErrPulsarNewProducer, err) } topic, err := helper.GetTopic(sinkURI) diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go index 8e92167327..c44b5c853e 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go @@ -63,7 +63,11 @@ func GetTopicManagerAndTryCreateTopic( ) if _, err := topicManager.CreateTopicAndWaitUntilVisible(ctx, topic); err != nil { +<<<<<<< HEAD return nil, cerror.WrapError(cerror.ErrKafkaCreateTopic, err) +======= + return nil, err +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } return topicManager, nil @@ -104,7 +108,7 @@ func (m *kafkaTopicManager) GetPartitionNum( // If the topic is not in the metadata, we try to create the topic. partitionNum, err := m.CreateTopicAndWaitUntilVisible(ctx, topic) if err != nil { - return 0, errors.Trace(err) + return 0, err } return partitionNum, nil @@ -264,7 +268,11 @@ func (m *kafkaTopicManager) createTopic( zap.Error(err), zap.Duration("duration", time.Since(start)), ) +<<<<<<< HEAD return 0, cerror.WrapError(cerror.ErrKafkaCreateTopic, err) +======= + return 0, err +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } log.Info( @@ -290,7 +298,14 @@ func (m *kafkaTopicManager) CreateTopicAndWaitUntilVisible( // which means we should create the topic later. topicDetails, err := m.admin.GetTopicsMeta([]string{topicName}, true) if err != nil { +<<<<<<< HEAD return 0, errors.Trace(err) +======= + if kafka.IsAdminAuthorizationFailed(err) { + return m.useConfiguredPartitionNum(topicName, err), nil + } + return 0, err +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } if detail, ok := topicDetails[topicName]; ok { numPartition := detail.NumPartitions @@ -303,12 +318,19 @@ func (m *kafkaTopicManager) CreateTopicAndWaitUntilVisible( partitionNum, err := m.createTopic(ctx, topicName) if err != nil { +<<<<<<< HEAD return 0, errors.Trace(err) +======= + if kafka.IsAdminAuthorizationFailed(err) { + return m.useConfiguredPartitionNum(topicName, err), nil + } + return 0, err +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } err = m.waitUntilTopicVisible(ctx, topicName) if err != nil { - return 0, errors.Trace(err) + return 0, err } return partitionNum, nil diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go index bf02658b24..1a2bd7eba1 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go @@ -18,6 +18,7 @@ import ( "testing" "github.com/pingcap/ticdc/pkg/common" + "github.com/pingcap/ticdc/pkg/errors" "github.com/pingcap/ticdc/pkg/sink/kafka" "github.com/stretchr/testify/require" ) @@ -35,7 +36,56 @@ func TestCreateTopic(t *testing.T) { changefeedID := common.NewChangefeedID4Test("test", "test") ctx := context.Background() +<<<<<<< HEAD manager := newKafkaTopicManager(ctx, kafka.DefaultMockTopicName, changefeedID, adminClient, cfg) +======= + var gotNewTopicDetail *kafka.TopicDetail + var gotNewTopicValidateOnly bool + var gotFailedTopicDetail *kafka.TopicDetail + var gotFailedTopicValidateOnly bool + gomock.InOrder( + adminClient.EXPECT().GetTopicsMeta([]string{kafkaTopicManagerTestTopic}, true).Return( + map[string]kafka.TopicDetail{ + kafkaTopicManagerTestTopic: { + Name: kafkaTopicManagerTestTopic, + NumPartitions: 2, + }, + }, nil), + adminClient.EXPECT().GetTopicsMeta([]string{"new-topic"}, true).Return( + map[string]kafka.TopicDetail{}, nil), + adminClient.EXPECT().GetTopicsMeta([]string{"new-topic"}, false).Return( + map[string]kafka.TopicDetail{}, nil), + adminClient.EXPECT().CreateTopic(gomock.Any(), false).DoAndReturn( + func(detail *kafka.TopicDetail, validateOnly bool) error { + gotNewTopicDetail = detail + gotNewTopicValidateOnly = validateOnly + return nil + }), + adminClient.EXPECT().GetTopicsMeta([]string{"new-topic"}, false).Return( + map[string]kafka.TopicDetail{ + "new-topic": { + Name: "new-topic", + NumPartitions: 2, + }, + }, nil), + adminClient.EXPECT().GetTopicsMeta([]string{"new-topic2"}, true).Return( + map[string]kafka.TopicDetail{}, nil), + adminClient.EXPECT().GetTopicsMeta([]string{"new-topic2"}, false).Return( + map[string]kafka.TopicDetail{}, nil), + adminClient.EXPECT().GetTopicsMeta([]string{"new-topic-failed"}, true).Return( + map[string]kafka.TopicDetail{}, nil), + adminClient.EXPECT().GetTopicsMeta([]string{"new-topic-failed"}, false).Return( + map[string]kafka.TopicDetail{}, nil), + adminClient.EXPECT().CreateTopic(gomock.Any(), false).DoAndReturn( + func(detail *kafka.TopicDetail, validateOnly bool) error { + gotFailedTopicDetail = detail + gotFailedTopicValidateOnly = validateOnly + return errors.WrapError(errors.ErrKafkaAdminAPI, sarama.ErrInvalidReplicationFactor, "create-topic", detail.Name) + }), + ) + + manager := newKafkaTopicManager(ctx, kafkaTopicManagerTestTopic, changefeedID, adminClient, cfg) +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) defer manager.Close() partitionNum, err := manager.CreateTopicAndWaitUntilVisible(ctx, kafka.DefaultMockTopicName) require.NoError(t, err) @@ -70,18 +120,64 @@ func TestCreateTopic(t *testing.T) { manager = newKafkaTopicManager(ctx, topic, changefeedID, adminClient, cfg) defer manager.Close() _, err = manager.CreateTopicAndWaitUntilVisible(ctx, topic) +<<<<<<< HEAD require.Regexp( t, "kafka create topic failed: kafka server: Replication-factor is invalid", err, ) +======= + require.ErrorIs(t, err, errors.ErrKafkaAdminAPI) + require.ErrorIs(t, err, sarama.ErrInvalidReplicationFactor) + require.NotNil(t, gotFailedTopicDetail) + require.Equal(t, "new-topic-failed", gotFailedTopicDetail.Name) + require.False(t, gotFailedTopicValidateOnly) +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } func TestCreateTopicWithDelay(t *testing.T) { t.Parallel() +<<<<<<< HEAD adminClient := kafka.NewClusterAdminClientMockImpl() defer adminClient.Close() +======= + ctrl := gomock.NewController(t) + adminClient := kafka.NewMockClusterAdminClient(ctrl) + topic := "new-topic" + gomock.InOrder( + adminClient.EXPECT().GetTopicsMeta([]string{topic}, true). + Return(map[string]kafka.TopicDetail{}, nil), + adminClient.EXPECT().GetTopicsMeta([]string{topic}, false). + Return(map[string]kafka.TopicDetail{}, nil), + adminClient.EXPECT().GetBrokerConfig(kafka.MinInsyncReplicasConfigName). + Return("2", true, nil), + ) + + manager := newKafkaTopicManager( + context.Background(), + topic, + common.NewChangefeedID4Test("test", "test"), + adminClient, + &kafka.AutoCreateTopicConfig{ + AutoCreate: true, + PartitionNum: 2, + ReplicationFactor: 1, + RequiredAcks: kafka.WaitForAll, + }, + ) + defer manager.Close() + + _, err := manager.CreateTopicAndWaitUntilVisible(context.Background(), topic) + require.ErrorContains(t, err, "`replication-factor` 1 is smaller than the `min.insync.replicas` 2 of broker") +} + +func TestCreateTopicWaitsUntilVisible(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + adminClient := kafka.NewMockClusterAdminClient(ctrl) +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) cfg := &kafka.AutoCreateTopicConfig{ AutoCreate: true, PartitionNum: 2, diff --git a/pkg/errors/error.go b/pkg/errors/error.go index e15d05e818..e75864ce9f 100644 --- a/pkg/errors/error.go +++ b/pkg/errors/error.go @@ -123,50 +123,21 @@ var ( "kafka send message failed", errors.RFCCodeText("CDC:ErrKafkaSendMessage"), ) - ErrKafkaProducerClosed = errors.Normalize( - "kafka producer closed", - errors.RFCCodeText("CDC:ErrKafkaProducerClosed"), + ErrKafkaSinkClosed = errors.Normalize( + "kafka sink closed", + errors.RFCCodeText("CDC:ErrKafkaSinkClosed"), ) - ErrKafkaAsyncSendMessage = errors.Normalize( - "kafka async send message failed", - errors.RFCCodeText("CDC:ErrKafkaAsyncSendMessage"), - ) - ErrKafkaInvalidPartitionNum = errors.Normalize( - "invalid partition num %d", - errors.RFCCodeText("CDC:ErrKafkaInvalidPartitionNum"), - ) - ErrKafkaInvalidRequiredAcks = errors.Normalize( - "invalid required acks %d, "+ - "only support these values: 0(NoResponse),1(WaitForLocal) and -1(WaitForAll)", - errors.RFCCodeText("CDC:ErrKafkaInvalidRequiredAcks"), - ) - ErrKafkaNewProducer = errors.Normalize( - "new kafka producer", - errors.RFCCodeText("CDC:ErrKafkaNewProducer"), - ) - ErrKafkaInvalidClientID = errors.Normalize( - "invalid kafka client ID '%s'", - errors.RFCCodeText("CDC:ErrKafkaInvalidClientID"), - ) - ErrKafkaInvalidVersion = errors.Normalize( - "invalid kafka version", - errors.RFCCodeText("CDC:ErrKafkaInvalidVersion"), + ErrNewKafkaSink = errors.Normalize( + "new kafka sink", + errors.RFCCodeText("CDC:ErrNewKafkaSink"), ) ErrKafkaInvalidConfig = errors.Normalize( "kafka config invalid", errors.RFCCodeText("CDC:ErrKafkaInvalidConfig"), ) - ErrKafkaCreateTopic = errors.Normalize( - "kafka create topic failed", - errors.RFCCodeText("CDC:ErrKafkaCreateTopic"), - ) - ErrKafkaInvalidTopicExpression = errors.Normalize( - "invalid topic expression: %s ", - errors.RFCCodeText("CDC:ErrKafkaTopicExprInvalid"), - ) - ErrKafkaConfigNotFound = errors.Normalize( - "kafka config item not found", - errors.RFCCodeText("CDC:ErrKafkaConfigNotFound"), + ErrKafkaAdminAPI = errors.Normalize( + "kafka admin API %s failed: %s", + errors.RFCCodeText("CDC:ErrKafkaAdminAPI"), ) ErrPulsarInvalidTopicExpression = errors.Normalize( "invalid topic expression", diff --git a/pkg/errors/error_test.go b/pkg/errors/error_test.go index e040eedc67..2c76ca8978 100644 --- a/pkg/errors/error_test.go +++ b/pkg/errors/error_test.go @@ -98,6 +98,26 @@ func TestShouldFailChangefeed(t *testing.T) { err: ErrKafkaInvalidConfig.GenWithStackByArgs("invalid config"), expected: true, }, + { + name: "ErrNewKafkaSink should return false", + err: ErrNewKafkaSink.GenWithStackByArgs(), + expected: false, + }, + { + name: "ErrKafkaAdminAPI should return false", + err: ErrKafkaAdminAPI.GenWithStackByArgs("describe-topic", "test-topic"), + expected: false, + }, + { + name: "ErrKafkaSendMessage should return false", + err: ErrKafkaSendMessage.GenWithStackByArgs(), + expected: false, + }, + { + name: "ErrKafkaSinkClosed should return false", + err: ErrKafkaSinkClosed.GenWithStackByArgs(), + expected: false, + }, { name: "ErrMySQLInvalidConfig should return true", err: ErrMySQLInvalidConfig.GenWithStackByArgs("invalid config"), diff --git a/pkg/sink/kafka/admin.go b/pkg/sink/kafka/admin.go index b8cfd1cfc3..b3d9029511 100644 --- a/pkg/sink/kafka/admin.go +++ b/pkg/sink/kafka/admin.go @@ -58,10 +58,10 @@ func (a *saramaAdminClient) GetAllBrokers() []Broker { return result } -func (a *saramaAdminClient) GetBrokerConfig(configName string) (string, error) { +func (a *saramaAdminClient) GetBrokerConfig(configName string) (string, bool, error) { _, controller, err := a.admin.DescribeCluster() if err != nil { - return "", errors.Trace(err) + return "", false, errors.WrapError(errors.ErrKafkaAdminAPI, err, "describe-cluster", "cluster") } configEntries, err := a.admin.DescribeConfig(sarama.ConfigResource{ @@ -70,7 +70,7 @@ func (a *saramaAdminClient) GetBrokerConfig(configName string) (string, error) { ConfigNames: []string{configName}, }) if err != nil { - return "", errors.Trace(err) + return "", false, errors.WrapError(errors.ErrKafkaAdminAPI, err, "describe-config", configName) } // For compatibility with KOP, we checked all return values. @@ -78,7 +78,7 @@ func (a *saramaAdminClient) GetBrokerConfig(configName string) (string, error) { // 2. Kop returns all configs. for _, entry := range configEntries { if entry.Name == configName { - return entry.Value, nil + return entry.Value, true, nil } } @@ -86,18 +86,17 @@ func (a *saramaAdminClient) GetBrokerConfig(configName string) (string, error) { zap.String("keyspace", a.changefeed.Keyspace()), zap.String("changefeed", a.changefeed.Name()), zap.String("configName", configName)) - return "", errors.ErrKafkaConfigNotFound.GenWithStack( - "cannot find the `%s` from the broker's configuration", configName) + return "", false, nil } -func (a *saramaAdminClient) GetTopicConfig(topicName string, configName string) (string, error) { +func (a *saramaAdminClient) GetTopicConfig(topicName string, configName string) (string, bool, error) { configEntries, err := a.admin.DescribeConfig(sarama.ConfigResource{ Type: sarama.TopicResource, Name: topicName, ConfigNames: []string{configName}, }) if err != nil { - return "", errors.Trace(err) + return "", false, errors.WrapError(errors.ErrKafkaAdminAPI, err, "describe-config", topicName) } // For compatibility with KOP, we checked all return values. @@ -110,7 +109,7 @@ func (a *saramaAdminClient) GetTopicConfig(topicName string, configName string) zap.String("changefeed", a.changefeed.Name()), zap.String("configName", configName), zap.String("configValue", entry.Value)) - return entry.Value, nil + return entry.Value, true, nil } } @@ -118,8 +117,7 @@ func (a *saramaAdminClient) GetTopicConfig(topicName string, configName string) zap.String("keyspace", a.changefeed.Keyspace()), zap.String("changefeed", a.changefeed.Name()), zap.String("configName", configName)) - return "", errors.ErrKafkaConfigNotFound.GenWithStack( - "cannot find the `%s` from the topic's configuration", configName) + return "", false, nil } func (a *saramaAdminClient) GetTopicsMeta(topics []string, ignoreTopicError bool) (map[string]TopicDetail, error) { @@ -127,7 +125,7 @@ func (a *saramaAdminClient) GetTopicsMeta(topics []string, ignoreTopicError bool metaList, err := a.admin.DescribeTopics(topics) if err != nil { - return nil, errors.Trace(err) + return nil, errors.WrapError(errors.ErrKafkaAdminAPI, err, "describe-topics", strings.Join(topics, ",")) } for _, meta := range metaList { @@ -136,7 +134,7 @@ func (a *saramaAdminClient) GetTopicsMeta(topics []string, ignoreTopicError bool continue } if !ignoreTopicError { - return nil, meta.Err + return nil, errors.WrapError(errors.ErrKafkaAdminAPI, meta.Err, "describe-topic", meta.Name) } log.Warn("fetch topic meta failed", zap.String("keyspace", a.changefeed.Keyspace()), @@ -158,7 +156,7 @@ func (a *saramaAdminClient) GetTopicsPartitionsNum(topics []string) (map[string] for _, topic := range topics { partition, err := a.client.Partitions(topic) if err != nil { - return nil, errors.Trace(err) + return nil, errors.WrapError(errors.ErrKafkaAdminAPI, err, "list-partitions", topic) } result[topic] = int32(len(partition)) } @@ -175,7 +173,7 @@ func (a *saramaAdminClient) CreateTopic(detail *TopicDetail, validateOnly bool) err := a.admin.CreateTopic(detail.Name, request, validateOnly) // Ignore the already exists error because it's not harmful. if err != nil && !strings.Contains(err.Error(), sarama.ErrTopicAlreadyExists.Error()) { - return err + return errors.WrapError(errors.ErrKafkaAdminAPI, err, "create-topic", detail.Name) } return nil } diff --git a/pkg/sink/kafka/admin_test.go b/pkg/sink/kafka/admin_test.go index ecab2e122d..4873f7ef45 100644 --- a/pkg/sink/kafka/admin_test.go +++ b/pkg/sink/kafka/admin_test.go @@ -14,13 +14,20 @@ package kafka import ( + "io" "testing" "github.com/IBM/sarama" +<<<<<<< HEAD +======= + "github.com/golang/mock/gomock" +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) "github.com/pingcap/ticdc/pkg/common" + "github.com/pingcap/ticdc/pkg/errors" "github.com/stretchr/testify/require" ) +<<<<<<< HEAD type testSaramaClient struct { closed bool } @@ -88,4 +95,83 @@ func TestSaramaAdminClientCloseFallsBackToClientWhenAdminIsNil(t *testing.T) { } require.NotPanics(t, func() { a.Close() }) require.True(t, client.closed) +======= +func TestGetBrokerConfig(t *testing.T) { + t.Parallel() + + t.Run("not found", func(t *testing.T) { + ctrl := gomock.NewController(t) + admin := NewMocksaramaClusterAdmin(ctrl) + admin.EXPECT().DescribeCluster().Return(nil, int32(1), nil) + admin.EXPECT().DescribeConfig(gomock.Any()).Return([]sarama.ConfigEntry{}, nil) + + client := &saramaAdminClient{ + changefeed: common.NewChangeFeedIDWithName("test", "default"), + admin: admin, + } + value, found, err := client.GetBrokerConfig("missing") + + require.NoError(t, err) + require.False(t, found) + require.Empty(t, value) + }) + + t.Run("admin error", func(t *testing.T) { + ctrl := gomock.NewController(t) + admin := NewMocksaramaClusterAdmin(ctrl) + cause := io.ErrUnexpectedEOF + admin.EXPECT().DescribeCluster().Return(nil, int32(0), cause) + + client := &saramaAdminClient{ + changefeed: common.NewChangeFeedIDWithName("test", "default"), + admin: admin, + } + _, _, err := client.GetBrokerConfig("missing") + + require.ErrorIs(t, err, errors.ErrKafkaAdminAPI) + require.ErrorIs(t, err, cause) + }) +} + +func TestAdminClientClose(t *testing.T) { + tests := []struct { + name string + setup func(*gomock.Controller) *saramaAdminClient + }{ + { + name: "uses admin close", + setup: func(ctrl *gomock.Controller) *saramaAdminClient { + client := NewMocksaramaClient(ctrl) + admin := NewMocksaramaClusterAdmin(ctrl) + admin.EXPECT().Close().Return(nil) + client.EXPECT().Close().Times(0) + return &saramaAdminClient{ + changefeed: common.NewChangeFeedIDWithName("test", "default"), + client: client, + admin: admin, + } + }, + }, + { + name: "falls back to client when admin is nil", + setup: func(ctrl *gomock.Controller) *saramaAdminClient { + client := NewMocksaramaClient(ctrl) + client.EXPECT().Close().Return(nil) + return &saramaAdminClient{ + changefeed: common.NewChangeFeedIDWithName("test", "default"), + client: client, + } + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + adminClient := test.setup(ctrl) + + require.NotPanics(t, func() { adminClient.Close() }) + }) + } +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } diff --git a/pkg/sink/kafka/claimcheck/claim_check.go b/pkg/sink/kafka/claimcheck/claim_check.go index 052785e2fa..50f1ec8356 100644 --- a/pkg/sink/kafka/claimcheck/claim_check.go +++ b/pkg/sink/kafka/claimcheck/claim_check.go @@ -22,9 +22,14 @@ import ( "github.com/google/uuid" "github.com/pingcap/errors" "github.com/pingcap/log" - commonType "github.com/pingcap/ticdc/pkg/common" + "github.com/pingcap/ticdc/pkg/common" "github.com/pingcap/ticdc/pkg/config" +<<<<<<< HEAD "github.com/pingcap/ticdc/pkg/sink/codec/common" +======= + "github.com/pingcap/ticdc/pkg/errors" + codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) "github.com/pingcap/ticdc/pkg/util" "github.com/pingcap/tidb/br/pkg/storage" "github.com/prometheus/client_golang/prometheus" @@ -40,7 +45,7 @@ type ClaimCheck struct { storage storage.ExternalStorage rawValue bool - changefeedID commonType.ChangeFeedID + changefeedID common.ChangeFeedID // metricSendMessageDuration tracks the time duration // cost on send messages to the claim check external storage. metricSendMessageDuration prometheus.Observer @@ -48,7 +53,7 @@ type ClaimCheck struct { } // New return a new ClaimCheck. -func New(ctx context.Context, config *config.LargeMessageHandleConfig, changefeedID commonType.ChangeFeedID) (*ClaimCheck, error) { +func New(ctx context.Context, config *config.LargeMessageHandleConfig, changefeedID common.ChangeFeedID) (*ClaimCheck, error) { if !config.EnableClaimCheck() { return nil, nil } @@ -67,7 +72,7 @@ func New(ctx context.Context, config *config.LargeMessageHandleConfig, changefee zap.String("storageURI", util.MaskSensitiveDataInURI(config.ClaimCheckStorageURI)), zap.Duration("duration", time.Since(start)), zap.Error(err)) - return nil, errors.Trace(err) + return nil, err } log.Info("claim-check create the external storage success", @@ -88,19 +93,19 @@ func New(ctx context.Context, config *config.LargeMessageHandleConfig, changefee // WriteMessage write message to the claim check external storage. func (c *ClaimCheck) WriteMessage(ctx context.Context, key, value []byte, fileName string) (err error) { if !c.rawValue { - m := common.ClaimCheckMessage{ + m := codecCommon.ClaimCheckMessage{ Key: key, Value: value, } value, err = json.Marshal(m) if err != nil { - return errors.Trace(err) + return errors.WrapError(errors.ErrMarshalFailed, err) } } start := time.Now() err = c.storage.WriteFile(ctx, fileName, value) if err != nil { - return errors.Trace(err) + return err } c.metricSendMessageDuration.Observe(time.Since(start).Seconds()) c.metricSendMessageCount.Inc() diff --git a/pkg/sink/kafka/claimcheck/claim_check_test.go b/pkg/sink/kafka/claimcheck/claim_check_test.go new file mode 100644 index 0000000000..080e62de0f --- /dev/null +++ b/pkg/sink/kafka/claimcheck/claim_check_test.go @@ -0,0 +1,110 @@ +// Copyright 2023 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package claimcheck + +import ( + "context" + "fmt" + "strings" + "testing" + + "github.com/pingcap/ticdc/pkg/common" + "github.com/pingcap/ticdc/pkg/config" + "github.com/pingcap/ticdc/pkg/errors" + "github.com/pingcap/tidb/pkg/objstore" + "github.com/pingcap/tidb/pkg/objstore/mockobjstore" + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" + "golang.org/x/sync/errgroup" +) + +func TestClaimCheck(t *testing.T) { + t.Parallel() + + ctx := context.Background() + + changefeedID := common.NewChangeFeedIDWithName("test", "") + largeHandleConfig := config.NewDefaultLargeMessageHandleConfig() + + claimCheck, err := New(ctx, largeHandleConfig, changefeedID) + require.NoError(t, err) + require.Nil(t, claimCheck) + + largeHandleConfig.LargeMessageHandleOption = config.LargeMessageHandleOptionClaimCheck + largeHandleConfig.ClaimCheckStorageURI = "file:///tmp/abc/" + claimCheck, err = New(ctx, largeHandleConfig, changefeedID) + require.NoError(t, err) + t.Cleanup(claimCheck.Close) + + fileName := claimCheck.FileNameWithPrefix("file.json") + require.Equal(t, "file:///tmp/abc/file.json", fileName) +} + +func TestClaimCheckStorageErrorWrappedOnce(t *testing.T) { + largeHandleConfig := config.NewDefaultLargeMessageHandleConfig() + largeHandleConfig.LargeMessageHandleOption = config.LargeMessageHandleOptionClaimCheck + largeHandleConfig.ClaimCheckStorageURI = "invalid://bucket" + + claimCheck, err := New(context.Background(), largeHandleConfig, common.NewChangeFeedIDWithName("test", "default")) + + require.Nil(t, claimCheck) + require.ErrorIs(t, err, errors.ErrExternalStorageAPI) + require.Equal(t, 1, strings.Count(err.Error(), string(errors.ErrExternalStorageAPI.RFCCode()))) +} + +func TestClaimCheckCloseClosesStorage(t *testing.T) { + var nilClaimCheck *ClaimCheck + require.NotPanics(t, nilClaimCheck.Close) + + ctrl := gomock.NewController(t) + storage := mockobjstore.NewMockStorage(ctrl) + storage.EXPECT().Close().Times(1) + claimCheck := &ClaimCheck{ + storage: storage, + changefeedID: common.NewChangeFeedIDWithName("test", "default"), + } + + claimCheck.Close() +} + +func TestClaimCheckConcurrentWrites(t *testing.T) { + ctx := context.Background() + storage := objstore.NewMemStorage() + changefeedID := common.NewChangeFeedIDWithName("test", "default") + claimCheck := &ClaimCheck{ + storage: storage, + rawValue: true, + changefeedID: changefeedID, + metricSendMessageDuration: claimCheckSendMessageDuration.WithLabelValues(changefeedID.Keyspace(), changefeedID.Name()), + metricSendMessageCount: claimCheckSendMessageCount.WithLabelValues(changefeedID.Keyspace(), changefeedID.Name()), + } + t.Cleanup(claimCheck.Close) + + const concurrency = 32 + group := new(errgroup.Group) + for i := range concurrency { + fileName := fmt.Sprintf("%d.json", i) + group.Go(func() error { + return claimCheck.WriteMessage(ctx, nil, []byte(fileName), fileName) + }) + } + require.NoError(t, group.Wait()) + + for i := range concurrency { + fileName := fmt.Sprintf("%d.json", i) + data, err := storage.ReadFile(ctx, fileName) + require.NoError(t, err) + require.Equal(t, fileName, string(data)) + } +} diff --git a/pkg/sink/kafka/cluster_admin_client.go b/pkg/sink/kafka/cluster_admin_client.go index 3c6c331c88..4f1ff36996 100644 --- a/pkg/sink/kafka/cluster_admin_client.go +++ b/pkg/sink/kafka/cluster_admin_client.go @@ -31,11 +31,11 @@ type ClusterAdminClient interface { // GetAllBrokers return all brokers among the cluster GetAllBrokers() []Broker - // GetBrokerConfig return the broker level configuration with the `configName` - GetBrokerConfig(configName string) (string, error) + // GetBrokerConfig returns the broker-level configuration and whether it exists. + GetBrokerConfig(configName string) (value string, found bool, err error) - // GetTopicConfig return the topic level configuration with the `configName` - GetTopicConfig(topicName string, configName string) (string, error) + // GetTopicConfig returns the topic-level configuration and whether it exists. + GetTopicConfig(topicName string, configName string) (value string, found bool, err error) // GetTopicsMeta return all target topics' metadata // if `ignoreTopicError` is true, ignore the topic error and return the metadata of valid topics diff --git a/pkg/sink/kafka/cluster_admin_client_mock.go b/pkg/sink/kafka/cluster_admin_client_mock.go new file mode 100644 index 0000000000..dfeebbd773 --- /dev/null +++ b/pkg/sink/kafka/cluster_admin_client_mock.go @@ -0,0 +1,136 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: pkg/sink/kafka/cluster_admin_client.go + +// Package kafka is a generated GoMock package. +package kafka + +import ( + reflect "reflect" + + gomock "github.com/golang/mock/gomock" +) + +// MockClusterAdminClient is a mock of ClusterAdminClient interface. +type MockClusterAdminClient struct { + ctrl *gomock.Controller + recorder *MockClusterAdminClientMockRecorder +} + +// MockClusterAdminClientMockRecorder is the mock recorder for MockClusterAdminClient. +type MockClusterAdminClientMockRecorder struct { + mock *MockClusterAdminClient +} + +// NewMockClusterAdminClient creates a new mock instance. +func NewMockClusterAdminClient(ctrl *gomock.Controller) *MockClusterAdminClient { + mock := &MockClusterAdminClient{ctrl: ctrl} + mock.recorder = &MockClusterAdminClientMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockClusterAdminClient) EXPECT() *MockClusterAdminClientMockRecorder { + return m.recorder +} + +// Close mocks base method. +func (m *MockClusterAdminClient) Close() { + m.ctrl.T.Helper() + m.ctrl.Call(m, "Close") +} + +// Close indicates an expected call of Close. +func (mr *MockClusterAdminClientMockRecorder) Close() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockClusterAdminClient)(nil).Close)) +} + +// CreateTopic mocks base method. +func (m *MockClusterAdminClient) CreateTopic(detail *TopicDetail, validateOnly bool) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "CreateTopic", detail, validateOnly) + ret0, _ := ret[0].(error) + return ret0 +} + +// CreateTopic indicates an expected call of CreateTopic. +func (mr *MockClusterAdminClientMockRecorder) CreateTopic(detail, validateOnly interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateTopic", reflect.TypeOf((*MockClusterAdminClient)(nil).CreateTopic), detail, validateOnly) +} + +// GetAllBrokers mocks base method. +func (m *MockClusterAdminClient) GetAllBrokers() []Broker { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetAllBrokers") + ret0, _ := ret[0].([]Broker) + return ret0 +} + +// GetAllBrokers indicates an expected call of GetAllBrokers. +func (mr *MockClusterAdminClientMockRecorder) GetAllBrokers() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetAllBrokers", reflect.TypeOf((*MockClusterAdminClient)(nil).GetAllBrokers)) +} + +// GetBrokerConfig mocks base method. +func (m *MockClusterAdminClient) GetBrokerConfig(configName string) (string, bool, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetBrokerConfig", configName) + ret0, _ := ret[0].(string) + ret1, _ := ret[1].(bool) + ret2, _ := ret[2].(error) + return ret0, ret1, ret2 +} + +// GetBrokerConfig indicates an expected call of GetBrokerConfig. +func (mr *MockClusterAdminClientMockRecorder) GetBrokerConfig(configName interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetBrokerConfig", reflect.TypeOf((*MockClusterAdminClient)(nil).GetBrokerConfig), configName) +} + +// GetTopicConfig mocks base method. +func (m *MockClusterAdminClient) GetTopicConfig(topicName, configName string) (string, bool, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetTopicConfig", topicName, configName) + ret0, _ := ret[0].(string) + ret1, _ := ret[1].(bool) + ret2, _ := ret[2].(error) + return ret0, ret1, ret2 +} + +// GetTopicConfig indicates an expected call of GetTopicConfig. +func (mr *MockClusterAdminClientMockRecorder) GetTopicConfig(topicName, configName interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTopicConfig", reflect.TypeOf((*MockClusterAdminClient)(nil).GetTopicConfig), topicName, configName) +} + +// GetTopicsMeta mocks base method. +func (m *MockClusterAdminClient) GetTopicsMeta(topics []string, ignoreTopicError bool) (map[string]TopicDetail, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetTopicsMeta", topics, ignoreTopicError) + ret0, _ := ret[0].(map[string]TopicDetail) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// GetTopicsMeta indicates an expected call of GetTopicsMeta. +func (mr *MockClusterAdminClientMockRecorder) GetTopicsMeta(topics, ignoreTopicError interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTopicsMeta", reflect.TypeOf((*MockClusterAdminClient)(nil).GetTopicsMeta), topics, ignoreTopicError) +} + +// GetTopicsPartitionsNum mocks base method. +func (m *MockClusterAdminClient) GetTopicsPartitionsNum(topics []string) (map[string]int32, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetTopicsPartitionsNum", topics) + ret0, _ := ret[0].(map[string]int32) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// GetTopicsPartitionsNum indicates an expected call of GetTopicsPartitionsNum. +func (mr *MockClusterAdminClientMockRecorder) GetTopicsPartitionsNum(topics interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTopicsPartitionsNum", reflect.TypeOf((*MockClusterAdminClient)(nil).GetTopicsPartitionsNum), topics) +} diff --git a/pkg/sink/kafka/logutil.go b/pkg/sink/kafka/logutil.go index 8a90e7e3e9..cb5ae54de2 100644 --- a/pkg/sink/kafka/logutil.go +++ b/pkg/sink/kafka/logutil.go @@ -18,12 +18,11 @@ import ( "strconv" "strings" - "github.com/pingcap/errors" - "github.com/pingcap/ticdc/pkg/sink/codec/common" + codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" ) // DetermineEventType infers the event type based on MessageLogInfo content. -func DetermineEventType(info *common.MessageLogInfo) string { +func DetermineEventType(info *codecCommon.MessageLogInfo) string { if info == nil { return "unknown" } @@ -40,7 +39,7 @@ func DetermineEventType(info *common.MessageLogInfo) string { } // BuildEventLogContext builds a textual representation of event info. -func BuildEventLogContext(keyspace, changefeed string, info *common.MessageLogInfo) string { +func BuildEventLogContext(keyspace, changefeed string, info *codecCommon.MessageLogInfo) string { var sb strings.Builder sb.WriteString("keyspace=") sb.WriteString(keyspace) @@ -83,22 +82,7 @@ func BuildEventLogContext(keyspace, changefeed string, info *common.MessageLogIn return sb.String() } -// AnnotateEventError logs the event context and annotates the error with that context. -func AnnotateEventError( - keyspace, changefeed string, - info *common.MessageLogInfo, - err error, -) error { - if err == nil { - return nil - } - if contextStr := BuildEventLogContext(keyspace, changefeed, info); contextStr != "" { - return errors.Annotate(err, contextStr+"; ErrorInfo:"+err.Error()) - } - return err -} - -func formatDMLInfo(rows []common.RowLogInfo) string { +func formatDMLInfo(rows []codecCommon.RowLogInfo) string { data, err := json.Marshal(rows) if err != nil { return "" diff --git a/pkg/sink/kafka/logutil_test.go b/pkg/sink/kafka/logutil_test.go index eddaa41f32..2af2a0d618 100644 --- a/pkg/sink/kafka/logutil_test.go +++ b/pkg/sink/kafka/logutil_test.go @@ -16,26 +16,26 @@ import ( "strings" "testing" - "github.com/pingcap/ticdc/pkg/sink/codec/common" + codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/stretchr/testify/require" ) func TestDetermineEventType(t *testing.T) { require.Equal(t, "unknown", DetermineEventType(nil)) - require.Equal(t, "dml", DetermineEventType(&common.MessageLogInfo{Rows: []common.RowLogInfo{{}}})) - require.Equal(t, "ddl", DetermineEventType(&common.MessageLogInfo{DDL: &common.DDLLogInfo{}})) - require.Equal(t, "checkpoint", DetermineEventType(&common.MessageLogInfo{Checkpoint: &common.CheckpointLogInfo{CommitTs: 1}})) - require.Equal(t, "unknown", DetermineEventType(&common.MessageLogInfo{})) + require.Equal(t, "dml", DetermineEventType(&codecCommon.MessageLogInfo{Rows: []codecCommon.RowLogInfo{{}}})) + require.Equal(t, "ddl", DetermineEventType(&codecCommon.MessageLogInfo{DDL: &codecCommon.DDLLogInfo{}})) + require.Equal(t, "checkpoint", DetermineEventType(&codecCommon.MessageLogInfo{Checkpoint: &codecCommon.CheckpointLogInfo{CommitTs: 1}})) + require.Equal(t, "unknown", DetermineEventType(&codecCommon.MessageLogInfo{})) } func TestBuildEventLogContextRowsIncluded(t *testing.T) { - rows := []common.RowLogInfo{ + rows := []codecCommon.RowLogInfo{ { Type: "insert", Database: "db1", Table: "t1", CommitTs: 1, - PrimaryKeys: []common.ColumnLogInfo{ + PrimaryKeys: []codecCommon.ColumnLogInfo{ {Name: "id", Value: 1}, }, }, @@ -46,7 +46,7 @@ func TestBuildEventLogContextRowsIncluded(t *testing.T) { CommitTs: 2, }, } - info := &common.MessageLogInfo{Rows: rows} + info := &codecCommon.MessageLogInfo{Rows: rows} ctx := BuildEventLogContext("ks", "cf", info) expected := formatDMLInfo(rows) require.Contains(t, ctx, "dmlInfo="+expected) @@ -56,8 +56,8 @@ func TestBuildEventLogContextRowsIncluded(t *testing.T) { func TestBuildEventLogContextLargeData(t *testing.T) { largeValue := strings.Repeat("a", 12*1024) - info := &common.MessageLogInfo{ - Rows: []common.RowLogInfo{ + info := &codecCommon.MessageLogInfo{ + Rows: []codecCommon.RowLogInfo{ {Type: "insert", Table: largeValue}, }, } @@ -65,3 +65,29 @@ func TestBuildEventLogContextLargeData(t *testing.T) { require.Contains(t, ctx, largeValue) require.NotContains(t, ctx, "...(truncated)") } + +func TestBuildEventLogContextBlockEvents(t *testing.T) { + t.Run("ddl", func(t *testing.T) { + ctx := BuildEventLogContext("ks", "cf", &codecCommon.MessageLogInfo{ + DDL: &codecCommon.DDLLogInfo{ + Query: "CREATE TABLE t(id INT PRIMARY KEY)", + StartTs: 1, + CommitTs: 2, + }, + }) + + require.Contains(t, ctx, "eventType=ddl") + require.Contains(t, ctx, "ddlQuery=\"CREATE TABLE t(id INT PRIMARY KEY)\"") + require.Contains(t, ctx, "ddlStartTs=1") + require.Contains(t, ctx, "ddlCommitTs=2") + }) + + t.Run("checkpoint", func(t *testing.T) { + ctx := BuildEventLogContext("ks", "cf", &codecCommon.MessageLogInfo{ + Checkpoint: &codecCommon.CheckpointLogInfo{CommitTs: 3}, + }) + + require.Contains(t, ctx, "eventType=checkpoint") + require.Contains(t, ctx, "checkpointTs=3") + }) +} diff --git a/pkg/sink/kafka/oauth2_token_provider.go b/pkg/sink/kafka/oauth2_token_provider.go index dd25b3ff31..b38e150d93 100644 --- a/pkg/sink/kafka/oauth2_token_provider.go +++ b/pkg/sink/kafka/oauth2_token_provider.go @@ -18,7 +18,7 @@ import ( "net/url" "github.com/IBM/sarama" - "github.com/pingcap/errors" + "github.com/pingcap/ticdc/pkg/errors" "golang.org/x/oauth2" "golang.org/x/oauth2/clientcredentials" ) @@ -68,7 +68,7 @@ func newTokenProvider(ctx context.Context, o *options) (sarama.AccessTokenProvid tokenURL, err := url.Parse(o.SASL.OAuth2.TokenURL) if err != nil { - return nil, errors.Trace(err) + return nil, errors.WrapError(errors.ErrKafkaInvalidConfig, err) } cfg := clientcredentials.Config{ diff --git a/pkg/sink/kafka/oauth2_token_provider_test.go b/pkg/sink/kafka/oauth2_token_provider_test.go index 4438377824..0ed4d7c044 100644 --- a/pkg/sink/kafka/oauth2_token_provider_test.go +++ b/pkg/sink/kafka/oauth2_token_provider_test.go @@ -15,8 +15,10 @@ package kafka import ( "context" + "net/url" "testing" + "github.com/pingcap/ticdc/pkg/errors" "github.com/pingcap/ticdc/pkg/security" "github.com/stretchr/testify/require" ) @@ -66,7 +68,9 @@ func TestNewTokenProvider(t *testing.T) { if ts.expectedErr == "" { require.NoError(t, err) } else { - require.Error(t, err) + require.ErrorIs(t, err, errors.ErrKafkaInvalidConfig) + var escapeErr url.EscapeError + require.ErrorAs(t, err, &escapeErr) require.Contains(t, err.Error(), ts.expectedErr) } }) diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go index c9b992814e..a19a6ddb44 100644 --- a/pkg/sink/kafka/options.go +++ b/pkg/sink/kafka/options.go @@ -108,7 +108,13 @@ func requireAcksFromString(acks int) (RequiredAcks, error) { case int(NoResponse): return NoResponse, nil default: +<<<<<<< HEAD return Unknown, cerror.ErrKafkaInvalidRequiredAcks.GenWithStackByArgs(acks) +======= + return Unknown, errors.ErrKafkaInvalidConfig.GenWithStack( + "invalid required acks %d, only support these values: "+ + "0(NoResponse), 1(WaitForLocal) and -1(WaitForAll)", acks) +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } } @@ -219,7 +225,11 @@ func (o *options) setPartitionNum(realPartitionCount int32) error { // the real partition count, since messages would be dispatched to different // partitions, this could prevent potential correctness problems. if o.PartitionNum > realPartitionCount { +<<<<<<< HEAD return cerror.ErrKafkaInvalidPartitionNum.GenWithStack( +======= + return errors.ErrKafkaInvalidConfig.GenWithStack( +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) "the number of partition (%d) specified in sink-uri is more than that of actual topic (%d)", o.PartitionNum, realPartitionCount) } @@ -236,15 +246,23 @@ func (o *options) Apply(changefeedID common.ChangeFeedID, req := &http.Request{URL: sinkURI} urlParameter := &urlConfig{} if err = binding.Query.Bind(req, urlParameter); err != nil { +<<<<<<< HEAD return cerror.WrapError(cerror.ErrMySQLInvalidConfig, err) +======= + return errors.WrapError(errors.ErrKafkaInvalidConfig, err) +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } if urlParameter, err = mergeConfig(sinkConfig, urlParameter); err != nil { - return err + return errors.WrapError(errors.ErrKafkaInvalidConfig, err) } if urlParameter.PartitionNum != nil { o.PartitionNum = *urlParameter.PartitionNum if o.PartitionNum <= 0 { +<<<<<<< HEAD return cerror.ErrKafkaInvalidPartitionNum.GenWithStackByArgs(o.PartitionNum) +======= + return errors.ErrKafkaInvalidConfig.GenWithStack("invalid partition num %d", o.PartitionNum) +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } } @@ -289,7 +307,7 @@ func (o *options) Apply(changefeedID common.ChangeFeedID, if urlParameter.DialTimeout != nil && *urlParameter.DialTimeout != "" { a, err := time.ParseDuration(*urlParameter.DialTimeout) if err != nil { - return err + return errors.WrapError(errors.ErrKafkaInvalidConfig, err) } o.DialTimeout = a } @@ -297,7 +315,7 @@ func (o *options) Apply(changefeedID common.ChangeFeedID, if urlParameter.WriteTimeout != nil && *urlParameter.WriteTimeout != "" { a, err := time.ParseDuration(*urlParameter.WriteTimeout) if err != nil { - return err + return errors.WrapError(errors.ErrKafkaInvalidConfig, err) } o.WriteTimeout = a } @@ -305,7 +323,7 @@ func (o *options) Apply(changefeedID common.ChangeFeedID, if urlParameter.ReadTimeout != nil && *urlParameter.ReadTimeout != "" { a, err := time.ParseDuration(*urlParameter.ReadTimeout) if err != nil { - return err + return errors.WrapError(errors.ErrKafkaInvalidConfig, err) } o.ReadTimeout = a } @@ -387,8 +405,12 @@ func (o *options) applyTLS(params *urlConfig) error { if o.Credential != nil && !o.Credential.IsEmpty() && !o.Credential.IsTLSEnabled() { +<<<<<<< HEAD return cerror.WrapError(cerror.ErrKafkaInvalidConfig, errors.New("ca, cert and key files should all be supplied")) +======= + return errors.ErrKafkaInvalidConfig.GenWithStack("ca, cert and key files should all be supplied") +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } // if enable-tls is not set, but credential files are set, @@ -401,8 +423,12 @@ func (o *options) applyTLS(params *urlConfig) error { enableTLS := *params.EnableTLS if o.Credential != nil && o.Credential.IsTLSEnabled() && !enableTLS { +<<<<<<< HEAD return cerror.WrapError(cerror.ErrKafkaInvalidConfig, errors.New("credential files are supplied, but 'enable-tls' is set to false")) +======= + return errors.ErrKafkaInvalidConfig.GenWithStack("credential files are supplied, but 'enable-tls' is set to false") +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } o.EnableTLS = enableTLS } else { @@ -493,8 +519,12 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf decodedClientSecret, err := base64.StdEncoding.DecodeString(clientSecret) if err != nil { log.Error("OAuth2 client secret is not base64 encoded", zap.Error(err)) +<<<<<<< HEAD return cerror.ErrKafkaInvalidConfig.GenWithStack( "OAuth2 client secret is not base64 encoded") +======= + return errors.ErrKafkaInvalidConfig.GenWithStack("OAuth2 client secret is not base64 encoded") +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } o.SASL.OAuth2.ClientSecret = string(decodedClientSecret) } @@ -516,7 +546,11 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf } if err := o.SASL.OAuth2.Validate(); err != nil { +<<<<<<< HEAD return cerror.ErrKafkaInvalidConfig.Wrap(err) +======= + return errors.WrapError(errors.ErrKafkaInvalidConfig, err) +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } o.SASL.OAuth2.SetDefault() } @@ -552,6 +586,47 @@ func (o *options) DeriveTopicConfig() *AutoCreateTopicConfig { } } +<<<<<<< HEAD +======= +// ValidateReplicationFactor checks whether a topic created with this config +// can satisfy the configured acknowledgment requirement. +func (c *AutoCreateTopicConfig) ValidateReplicationFactor(admin ClusterAdminClient) error { + if c.RequiredAcks != WaitForAll { + return nil + } + + raw, found, err := admin.GetBrokerConfig(MinInsyncReplicasConfigName) + if err != nil { + log.Warn("cannot get Kafka broker configuration, assume replication factor is valid", + zap.String("configName", MinInsyncReplicasConfigName), + zap.Int16("replicationFactor", c.ReplicationFactor), + zap.Error(err)) + return nil + } + if !found { + log.Warn("Kafka broker configuration not found, assume replication factor is valid", + zap.String("configName", MinInsyncReplicasConfigName), + zap.Int16("replicationFactor", c.ReplicationFactor)) + return nil + } + minInsyncReplicas, err := strconv.Atoi(raw) + if err != nil { + return errors.WrapError(errors.ErrKafkaAdminAPI, err, "parse-config", MinInsyncReplicasConfigName) + } + + if int(c.ReplicationFactor) < minInsyncReplicas { + return errors.ErrKafkaInvalidConfig.GenWithStack( + "TiCDC Kafka sink's `request.required.acks` defaults to -1, "+ + "TiCDC cannot deliver messages when the `replication-factor` %d "+ + "is smaller than the `min.insync.replicas` %d of broker", + c.ReplicationFactor, minInsyncReplicas, + ) + } + + return nil +} + +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) var ( validClientID = regexp.MustCompile(`\A[A-Za-z0-9._-]+\z`) commonInvalidChar = regexp.MustCompile(`[\?:,"]`) @@ -570,7 +645,11 @@ func NewKafkaClientID(captureAddr string, clientID = commonInvalidChar.ReplaceAllString(clientID, "_") } if !validClientID.MatchString(clientID) { +<<<<<<< HEAD return "", cerror.ErrKafkaInvalidClientID.GenWithStackByArgs(clientID) +======= + return "", errors.ErrKafkaInvalidConfig.GenWithStack("invalid kafka client ID %q", clientID) +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } return } @@ -584,7 +663,7 @@ func adjustOptions( ) error { topics, err := admin.GetTopicsMeta([]string{topic}, true) if err != nil { - return errors.Trace(err) + return err } // Only check replicationFactor >= minInsyncReplicas when producer's required acks is -1. @@ -650,11 +729,48 @@ func adjustOptions( } brokerMessageMaxBytes, err := strconv.Atoi(brokerMessageMaxBytesStr) if err != nil { +<<<<<<< HEAD return errors.Trace(err) +======= + return err + } + + options.MaxBatchedBytes = min(options.MaxBatchedBytes, options.MaxMessageBytes) + return nil +} + +func adjustExistingTopicOption( + changefeedID common.ChangeFeedID, + admin ClusterAdminClient, + options *options, + topic string, + info TopicDetail, +) error { + maxMessageBytes, found, err := getTopicMaxMessageBytes(admin, info.Name) + if err != nil || !found { + log.Warn("`max.message.bytes` not found from topic's configuration, use the option `MaxMessageBytes` as default", + zap.String("namespace", changefeedID.Keyspace()), zap.String("changefeed", changefeedID.Name()), + zap.Int("maxMessageBytes", options.MaxMessageBytes), zap.Error(err)) + maxMessageBytes = options.MaxMessageBytes + } + options.MaxMessageBytes = maxMessageBytes + + // no need to create the topic, + // but we would have to log user if they found enter wrong topic name later + if options.AutoCreate { + log.Warn("topic already exist, TiCDC will not create the topic", + zap.String("namespace", changefeedID.Keyspace()), zap.String("changefeed", changefeedID.Name()), + zap.String("topic", topic), zap.Any("detail", info)) + } + + if err = options.setPartitionNum(changefeedID, info.NumPartitions); err != nil { + return err +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } // when create the topic, `max.message.bytes` is decided by the broker, // it would use broker's `message.max.bytes` to set topic's `max.message.bytes`. +<<<<<<< HEAD // TiCDC need to make sure that the producer's `MaxMessageBytes` won't larger than // broker's `message.max.bytes`. maxMessageBytes := brokerMessageMaxBytes - maxMessageBytesOverhead @@ -669,6 +785,14 @@ func adjustOptions( if maxMessageBytes < options.MaxMessageBytes { options.MaxMessageBytes = maxMessageBytes } +======= + messageMaxBytes, found, err := getBrokerMaxMessageBytes(admin) + if err != nil || !found { + log.Warn("`message.max.bytes` not found from broker's configuration, use the option `MaxMessageBytes` as default", + zap.String("namespace", changefeedID.Keyspace()), zap.String("changefeed", changefeedID.Name()), + zap.Int("maxMessageBytes", options.MaxMessageBytes), zap.Error(err)) + messageMaxBytes = options.MaxMessageBytes +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } // topic not exists yet, and user does not specify the `partition-num` in the sink uri. @@ -685,6 +809,7 @@ func validateMinInsyncReplicas( admin ClusterAdminClient, topics map[string]TopicDetail, topic string, +<<<<<<< HEAD replicationFactor int, ) error { minInsyncReplicasConfigGetter := func() (string, bool, error) { @@ -706,10 +831,39 @@ func validateMinInsyncReplicas( } return minInsyncReplicasStr, false, nil +======= +) (int, bool, error) { + raw, found, err := getTopicConfig( + admin, topic, + TopicMaxMessageBytesConfigName, + BrokerMessageMaxBytesConfigName, + ) + if err != nil { + return 0, false, err + } + if !found { + return 0, false, nil + } + maxMessageBytes, err := strconv.Atoi(raw) + if err != nil { + return 0, false, errors.WrapError(errors.ErrKafkaAdminAPI, err, "parse-config", TopicMaxMessageBytesConfigName) + } + return maxMessageBytes, true, nil +} + +func getBrokerMaxMessageBytes(admin ClusterAdminClient) (int, bool, error) { + raw, found, err := admin.GetBrokerConfig(BrokerMessageMaxBytesConfigName) + if err != nil { + return 0, false, err + } + if !found { + return 0, false, nil +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } minInsyncReplicasStr, exists, err := minInsyncReplicasConfigGetter() if err != nil { +<<<<<<< HEAD // 'min.insync.replica' is invisible to us in Confluent Cloud Kafka. if cerror.ErrKafkaConfigNotFound.Equal(err) { log.Warn("TiCDC cannot find `min.insync.replicas` from broker's configuration, " + @@ -745,6 +899,11 @@ func validateMinInsyncReplicas( } return nil +======= + return 0, false, errors.WrapError(errors.ErrKafkaAdminAPI, err, "parse-config", BrokerMessageMaxBytesConfigName) + } + return messageMaxBytes, true, nil +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } // getTopicConfig gets topic config by name. @@ -757,12 +916,13 @@ func getTopicConfig( topicName string, topicConfigName string, brokerConfigName string, -) (string, error) { - if c, err := admin.GetTopicConfig(topicName, topicConfigName); err == nil { - return c, nil +) (string, bool, error) { + c, found, err := admin.GetTopicConfig(topicName, topicConfigName) + if err == nil && found { + return c, true, nil } - log.Info("kafka sink cannot find the configuration from topic, try to get it from broker", - zap.String("topic", topicName), zap.String("config", topicConfigName)) + log.Info("kafka sink cannot get the configuration from topic, try to get it from broker", + zap.String("topic", topicName), zap.String("config", topicConfigName), zap.Error(err)) return admin.GetBrokerConfig(brokerConfigName) } diff --git a/pkg/sink/kafka/options_test.go b/pkg/sink/kafka/options_test.go index 8f64d49762..e7a20d51f1 100644 --- a/pkg/sink/kafka/options_test.go +++ b/pkg/sink/kafka/options_test.go @@ -23,14 +23,155 @@ import ( "time" "github.com/IBM/sarama" +<<<<<<< HEAD "github.com/aws/aws-sdk-go/aws" commonType "github.com/pingcap/ticdc/pkg/common" +======= + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/golang/mock/gomock" + "github.com/pingcap/ticdc/pkg/common" +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) "github.com/pingcap/ticdc/pkg/config" "github.com/pingcap/ticdc/pkg/errors" "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/stretchr/testify/require" ) +<<<<<<< HEAD +======= +const ( + defaultMockTopicName = "mock_topic" + + // These values model Kafka admin responses, not TiCDC option defaults. + mockClusterReplicationFactor int16 = 3 + mockBrokerMessageMaxBytes = "1048588" + mockTopicMessageMaxBytes = "1048588" + mockMinInsyncReplicas = "1" +) + +type kafkaAdminFixture struct { + admin *MockClusterAdminClient + topics map[string]TopicDetail + brokerConfig map[string]string + topicConfig map[string]map[string]string +} + +func newKafkaAdminFixture(t *testing.T) *kafkaAdminFixture { + t.Helper() + + ctrl := gomock.NewController(t) + fixture := &kafkaAdminFixture{ + admin: NewMockClusterAdminClient(ctrl), + topics: make(map[string]TopicDetail), + brokerConfig: map[string]string{ + BrokerMessageMaxBytesConfigName: mockBrokerMessageMaxBytes, + MinInsyncReplicasConfigName: mockMinInsyncReplicas, + }, + topicConfig: make(map[string]map[string]string), + } + fixture.addTopic(defaultMockTopicName, defaultPartitionNum) + fixture.topicConfig[defaultMockTopicName] = map[string]string{ + TopicMaxMessageBytesConfigName: mockTopicMessageMaxBytes, + MinInsyncReplicasConfigName: mockMinInsyncReplicas, + } + + fixture.admin.EXPECT().Close().AnyTimes() + fixture.admin.EXPECT().GetTopicsMeta(gomock.Any(), gomock.Any()). + DoAndReturn(fixture.getTopicsMeta).AnyTimes() + fixture.admin.EXPECT().GetTopicsPartitionsNum(gomock.Any()). + DoAndReturn(fixture.getTopicsPartitionsNum).AnyTimes() + fixture.admin.EXPECT().GetBrokerConfig(gomock.Any()). + DoAndReturn(fixture.getBrokerConfig).AnyTimes() + fixture.admin.EXPECT().GetTopicConfig(gomock.Any(), gomock.Any()). + DoAndReturn(fixture.getTopicConfig).AnyTimes() + fixture.admin.EXPECT().CreateTopic(gomock.Any(), gomock.Any()). + DoAndReturn(fixture.createTopic).AnyTimes() + + return fixture +} + +func (f *kafkaAdminFixture) addTopic(name string, partitionNum int32) { + f.topics[name] = TopicDetail{Name: name, NumPartitions: partitionNum} +} + +func (f *kafkaAdminFixture) getTopicsMeta( + topics []string, _ bool, +) (map[string]TopicDetail, error) { + result := make(map[string]TopicDetail, len(topics)) + for _, topic := range topics { + if detail, ok := f.topics[topic]; ok { + result[topic] = detail + } + } + return result, nil +} + +func (f *kafkaAdminFixture) getTopicsPartitionsNum( + topics []string, +) (map[string]int32, error) { + result := make(map[string]int32, len(topics)) + for _, topic := range topics { + if detail, ok := f.topics[topic]; ok { + result[topic] = detail.NumPartitions + } + } + return result, nil +} + +func (f *kafkaAdminFixture) getBrokerConfig(configName string) (string, bool, error) { + if value, ok := f.brokerConfig[configName]; ok { + return value, true, nil + } + return "", false, nil +} + +func (f *kafkaAdminFixture) getTopicConfig(topicName string, configName string) (string, bool, error) { + if _, ok := f.topics[topicName]; !ok { + return "", false, nil + } + if value, ok := f.topicConfig[topicName][configName]; ok { + return value, true, nil + } + return "", false, nil +} + +func (f *kafkaAdminFixture) createTopic(detail *TopicDetail, _ bool) error { + if detail.ReplicationFactor > mockClusterReplicationFactor { + return sarama.ErrInvalidReplicationFactor + } + if _, ok := f.brokerConfig[MinInsyncReplicasConfigName]; !ok && + detail.ReplicationFactor != mockClusterReplicationFactor { + return sarama.ErrPolicyViolation + } + f.topics[detail.Name] = *detail + return nil +} + +func (f *kafkaAdminFixture) brokerMessageMaxBytes() int { + value, _ := strconv.Atoi(f.brokerConfig[BrokerMessageMaxBytesConfigName]) + return value +} + +func (f *kafkaAdminFixture) topicMaxMessageBytes(topicName string) int { + value, _ := strconv.Atoi(f.topicConfig[topicName][TopicMaxMessageBytesConfigName]) + return value +} + +func (f *kafkaAdminFixture) setMessageMaxBytes(brokerValue, topicValue string) { + f.brokerConfig[BrokerMessageMaxBytesConfigName] = brokerValue + f.topicConfig[defaultMockTopicName][TopicMaxMessageBytesConfigName] = topicValue +} + +func (f *kafkaAdminFixture) setMinInsyncReplicas(minInsyncReplicas string) { + f.topicConfig[defaultMockTopicName][MinInsyncReplicasConfigName] = minInsyncReplicas + f.brokerConfig[MinInsyncReplicasConfigName] = minInsyncReplicas +} + +func (f *kafkaAdminFixture) dropBrokerConfig(configName string) { + delete(f.brokerConfig, configName) +} + +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) func TestCompleteOptions(t *testing.T) { options := NewOptions() @@ -43,7 +184,7 @@ func TestCompleteOptions(t *testing.T) { sinkURI, err := url.Parse(uri) require.NoError(t, err) - err = options.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) + err = options.Apply(common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) require.NoError(t, err) require.Equal(t, int32(1), options.PartitionNum) require.Equal(t, int16(3), options.ReplicationFactor) @@ -57,7 +198,7 @@ func TestCompleteOptions(t *testing.T) { sinkURI, err = url.Parse(uri) require.NoError(t, err) options = NewOptions() - err = options.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), + err = options.Apply(common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) require.NoError(t, err) require.Len(t, options.BrokerEndpoints, 3) @@ -67,15 +208,30 @@ func TestCompleteOptions(t *testing.T) { sinkURI, err = url.Parse(uri) require.NoError(t, err) options = NewOptions() - err = options.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) + err = options.Apply(common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) require.Regexp(t, ".*invalid syntax.*", errors.Cause(err)) +<<<<<<< HEAD +======= + for _, replicationFactor := range []string{"0", "-1"} { + uri = "kafka://127.0.0.1:9092/abc?replication-factor=" + replicationFactor + sinkURI, err = url.Parse(uri) + require.NoError(t, err) + options = NewOptions() + err = options.Apply( + common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), + sinkURI, + config.GetDefaultReplicaConfig().Sink, + ) + require.ErrorContains(t, err, "invalid replication-factor "+replicationFactor) + } +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) // Illegal max-message-bytes. uri = "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&max-message-bytes=a" sinkURI, err = url.Parse(uri) require.NoError(t, err) options = NewOptions() - err = options.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) + err = options.Apply(common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) require.Regexp(t, ".*invalid syntax.*", errors.Cause(err)) // Illegal max-retry. @@ -83,7 +239,7 @@ func TestCompleteOptions(t *testing.T) { sinkURI, err = url.Parse(uri) require.NoError(t, err) options = NewOptions() - err = options.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) + err = options.Apply(common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) require.Regexp(t, ".*invalid syntax.*", errors.Cause(err)) // Illegal partition-num. @@ -91,7 +247,7 @@ func TestCompleteOptions(t *testing.T) { sinkURI, err = url.Parse(uri) require.NoError(t, err) options = NewOptions() - err = options.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) + err = options.Apply(common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) require.Regexp(t, ".*invalid syntax.*", errors.Cause(err)) // Out of range partition-num. @@ -99,7 +255,7 @@ func TestCompleteOptions(t *testing.T) { sinkURI, err = url.Parse(uri) require.NoError(t, err) options = NewOptions() - err = options.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) + err = options.Apply(common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) require.Regexp(t, ".*invalid partition num.*", errors.Cause(err)) // Unknown required-acks. @@ -107,7 +263,7 @@ func TestCompleteOptions(t *testing.T) { sinkURI, err = url.Parse(uri) require.NoError(t, err) options = NewOptions() - err = options.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) + err = options.Apply(common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) require.Regexp(t, ".*invalid required acks 3.*", errors.Cause(err)) // invalid kafka client id @@ -115,15 +271,15 @@ func TestCompleteOptions(t *testing.T) { sinkURI, err = url.Parse(uri) require.NoError(t, err) options = NewOptions() - err = options.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) - require.True(t, errors.ErrKafkaInvalidClientID.Equal(err)) + err = options.Apply(common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) + require.True(t, errors.ErrKafkaInvalidConfig.Equal(err)) // max-retry accepts non-negative sink-uri values. uri = "kafka://127.0.0.1:9092/abc?max-retry=7" sinkURI, err = url.Parse(uri) require.NoError(t, err) options = NewOptions() - err = options.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) + err = options.Apply(common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) require.NoError(t, err) require.Equal(t, 7, options.MaxRetry) @@ -131,7 +287,7 @@ func TestCompleteOptions(t *testing.T) { sinkURI, err = url.Parse(uri) require.NoError(t, err) options = NewOptions() - err = options.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) + err = options.Apply(common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) require.NoError(t, err) require.Equal(t, 0, options.MaxRetry) @@ -140,14 +296,76 @@ func TestCompleteOptions(t *testing.T) { sinkURI, err = url.Parse(uri) require.NoError(t, err) options = NewOptions() - err = options.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) + err = options.Apply(common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) require.NoError(t, err) require.Equal(t, defaultMaxRetry, options.MaxRetry) } +<<<<<<< HEAD func TestSetPartitionNum(t *testing.T) { options := NewOptions() err := options.setPartitionNum(2) +======= +func TestApplyRejectsNonPositiveMaxMessageBytes(t *testing.T) { + tests := []struct { + name string + uri string + configValue *int + expected int + }{ + { + name: "zero from URI", + uri: "kafka://127.0.0.1:9092/test-topic?max-message-bytes=0", + expected: 0, + }, + { + name: "negative from URI", + uri: "kafka://127.0.0.1:9092/test-topic?max-message-bytes=-1", + expected: -1, + }, + { + name: "zero from sink config", + uri: "kafka://127.0.0.1:9092/test-topic", + configValue: aws.Int(0), + expected: 0, + }, + { + name: "negative from sink config", + uri: "kafka://127.0.0.1:9092/test-topic", + configValue: aws.Int(-1), + expected: -1, + }, + } + + changefeedID := common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test") + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + sinkURI, err := url.Parse(test.uri) + require.NoError(t, err) + + sinkConfig := config.GetDefaultReplicaConfig().Sink + if test.configValue != nil { + sinkConfig.KafkaConfig = &config.KafkaConfig{ + MaxMessageBytes: test.configValue, + } + } + + options := NewOptions() + err = options.Apply(changefeedID, sinkURI, sinkConfig) + require.ErrorContains( + t, err, fmt.Sprintf("invalid max-message-bytes %d", test.expected)) + errCode, ok := errors.RFCCode(err) + require.True(t, ok) + require.Equal(t, errors.ErrKafkaInvalidConfig.RFCCode(), errCode) + }) + } +} + +func TestSetPartitionNum(t *testing.T) { + options := NewOptions() + changefeedID := common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test") + err := options.setPartitionNum(changefeedID, 2) +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) require.NoError(t, err) require.Equal(t, int32(2), options.PartitionNum) @@ -157,8 +375,13 @@ func TestSetPartitionNum(t *testing.T) { require.Equal(t, int32(1), options.PartitionNum) options.PartitionNum = 3 +<<<<<<< HEAD err = options.setPartitionNum(2) require.True(t, errors.ErrKafkaInvalidPartitionNum.Equal(err)) +======= + err = options.setPartitionNum(changefeedID, 2) + require.True(t, errors.ErrKafkaInvalidConfig.Equal(err)) +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } func TestClientID(t *testing.T) { @@ -196,7 +419,7 @@ func TestClientID(t *testing.T) { } for _, tc := range testCases { id, err := NewKafkaClientID(tc.addr, - commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, tc.changefeedID), tc.configuredID) + common.NewChangefeedID4Test(common.DefaultKeyspaceName, tc.changefeedID), tc.configuredID) if tc.hasError { require.Error(t, err) } else { @@ -217,7 +440,7 @@ func TestTimeout(t *testing.T) { sinkURI, err := url.Parse(uri) require.NoError(t, err) - err = options.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) + err = options.Apply(common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) require.NoError(t, err) require.Equal(t, 5*time.Second, options.DialTimeout) @@ -230,8 +453,17 @@ func TestAdjustConfigTopicNotExist(t *testing.T) { adminClient := NewClusterAdminClientMockImpl() defer adminClient.Close() +<<<<<<< HEAD options := NewOptions() options.BrokerEndpoints = []string{"127.0.0.1:9092"} +======= + topicName := "test-topic" + changefeedID := common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test") + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + adminFixture := newKafkaAdminFixture(t) + adminClient := adminFixture.admin +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) // topic not exist, `max-message-bytes` = `message.max.bytes` options.MaxMessageBytes = adminClient.GetBrokerMessageMaxBytes() @@ -640,10 +872,17 @@ func TestConfigurationCombinations(t *testing.T) { sinkURI, err := url.Parse(uri) require.Nil(t, err) +<<<<<<< HEAD ctx := context.Background() options := NewOptions() err = options.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) require.Nil(t, err) +======= + options := NewOptions() + err = options.Apply(common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) + require.Nil(t, err) + configuredMaxMessageBytes := options.MaxMessageBytes +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) changefeed := commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "changefeed-test") factory, err := NewMockFactory(ctx, options, changefeed) @@ -652,11 +891,23 @@ func TestConfigurationCombinations(t *testing.T) { adminClient, err := factory.AdminClient(ctx) require.NoError(t, err) +<<<<<<< HEAD topic, ok := a.uriParams[0].(string) require.True(t, ok) require.NotEqual(t, "", topic) err = adjustOptions(ctx, adminClient, options, topic) require.Nil(t, err) +======= + changefeedID := common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test") + err = adjustOptions(changefeedID, adminClient, options, topic) + require.Nil(t, err) + require.Equal(t, sourceMaxMessageBytes, options.MaxMessageBytes) + require.Equal( + t, + min(configuredMaxMessageBytes, sourceMaxMessageBytes), + options.MaxBatchedBytes, + ) +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) encoderConfig := common.NewConfig(config.ProtocolOpen) err = encoderConfig.Apply(sinkURI, &config.SinkConfig{ @@ -709,7 +960,7 @@ func TestMerge(t *testing.T) { Key: aws.String("key.pem"), } c := NewOptions() - err = c.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), sinkURI, replicaConfig.Sink) + err = c.Apply(common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), sinkURI, replicaConfig.Sink) require.NoError(t, err) require.Equal(t, int32(12), c.PartitionNum) require.Equal(t, int16(5), c.ReplicationFactor) @@ -790,7 +1041,7 @@ func TestMerge(t *testing.T) { Key: aws.String("key2.pem"), } c = NewOptions() - err = c.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), sinkURI, replicaConfig.Sink) + err = c.Apply(common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), sinkURI, replicaConfig.Sink) require.NoError(t, err) require.Equal(t, int32(12), c.PartitionNum) require.Equal(t, int16(5), c.ReplicationFactor) diff --git a/pkg/sink/kafka/sarama_async_producer.go b/pkg/sink/kafka/sarama_async_producer.go index 9c0bd82104..f0c0f6b5d1 100644 --- a/pkg/sink/kafka/sarama_async_producer.go +++ b/pkg/sink/kafka/sarama_async_producer.go @@ -18,12 +18,10 @@ import ( "time" "github.com/IBM/sarama" - "github.com/pingcap/errors" - "github.com/pingcap/failpoint" "github.com/pingcap/log" - commonType "github.com/pingcap/ticdc/pkg/common" - cerror "github.com/pingcap/ticdc/pkg/errors" - "github.com/pingcap/ticdc/pkg/sink/codec/common" + "github.com/pingcap/ticdc/pkg/common" + "github.com/pingcap/ticdc/pkg/errors" + codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" "go.uber.org/atomic" "go.uber.org/zap" ) @@ -31,15 +29,14 @@ import ( type saramaAsyncProducer struct { client sarama.Client producer sarama.AsyncProducer - changefeedID commonType.ChangeFeedID + changefeedID common.ChangeFeedID - closed *atomic.Bool - failpointCh chan *sarama.ProducerError + closed *atomic.Bool } type messageMetadata struct { callback func() - logInfo *common.MessageLogInfo + logInfo *codecCommon.MessageLogInfo } func (p *saramaAsyncProducer) Close() { @@ -103,13 +100,7 @@ func (p *saramaAsyncProducer) AsyncRunCallback( log.Info("async producer exit since context is done", zap.String("keyspace", p.changefeedID.Keyspace()), zap.String("changefeed", p.changefeedID.Name())) - return errors.Trace(ctx.Err()) - case err := <-p.failpointCh: - log.Warn("Receive from failpoint chan in kafka DML producer", - zap.String("keyspace", p.changefeedID.Keyspace()), - zap.String("changefeed", p.changefeedID.Name()), - zap.Error(err)) - return p.handleProducerError(err) + return context.Cause(ctx) case ack := <-p.producer.Successes(): if ack != nil { switch meta := ack.Metadata.(type) { @@ -137,37 +128,23 @@ func (p *saramaAsyncProducer) AsyncRunCallback( } func (p *saramaAsyncProducer) handleProducerError(err *sarama.ProducerError) error { - errWithInfo := AnnotateEventError( - p.changefeedID.Keyspace(), - p.changefeedID.Name(), - extractLogInfo(err.Msg), - err.Err, - ) - return cerror.WrapError(cerror.ErrKafkaAsyncSendMessage, errWithInfo) + log.Error("send message to kafka failed", + zap.String("keyspace", p.changefeedID.Keyspace()), + zap.String("changefeed", p.changefeedID.Name()), + zap.String("eventContext", BuildEventLogContext( + p.changefeedID.Keyspace(), p.changefeedID.Name(), extractLogInfo(err.Msg))), + zap.Error(err.Err)) + return errors.WrapError(errors.ErrKafkaSendMessage, err.Err) } // AsyncSend is the input channel for the user to write messages to that they // wish to send. func (p *saramaAsyncProducer) AsyncSend( - ctx context.Context, topic string, partition int32, message *common.Message, + ctx context.Context, topic string, partition int32, message *codecCommon.Message, ) error { if p.closed.Load() { - return cerror.ErrKafkaProducerClosed.GenWithStackByArgs() + return errors.ErrKafkaSinkClosed.GenWithStackByArgs() } - failpoint.Inject("KafkaSinkAsyncSendError", func() { - // simulate sending message to input channel successfully but flushing - // message to Kafka meets error - log.Info("KafkaSinkAsyncSendError error injected", zap.String("keyspace", p.changefeedID.Keyspace()), - zap.String("changefeed", p.changefeedID.Name())) - p.failpointCh <- &sarama.ProducerError{ - Err: errors.New("kafka sink injected error"), - Msg: &sarama.ProducerMessage{Metadata: &messageMetadata{ - callback: message.Callback, - logInfo: message.LogInfo, - }}, - } - failpoint.Return(nil) - }) meta := &messageMetadata{ callback: message.Callback, logInfo: message.LogInfo, @@ -181,13 +158,13 @@ func (p *saramaAsyncProducer) AsyncSend( } select { case <-ctx.Done(): - return errors.Trace(ctx.Err()) + return context.Cause(ctx) case p.producer.Input() <- msg: } return nil } -func extractLogInfo(msg *sarama.ProducerMessage) *common.MessageLogInfo { +func extractLogInfo(msg *sarama.ProducerMessage) *codecCommon.MessageLogInfo { if msg == nil { return nil } diff --git a/pkg/sink/kafka/sarama_config.go b/pkg/sink/kafka/sarama_config.go index b53dc47b37..4988c79c52 100644 --- a/pkg/sink/kafka/sarama_config.go +++ b/pkg/sink/kafka/sarama_config.go @@ -108,7 +108,7 @@ func newSaramaConfig(ctx context.Context, o *options) (*sarama.Config, error) { if o.Credential != nil && o.Credential.IsTLSEnabled() { config.Net.TLS.Config, err = o.Credential.ToTLSConfig() if err != nil { - return nil, errors.Trace(err) + return nil, errors.WrapError(errors.ErrKafkaInvalidConfig, err) } } @@ -117,7 +117,7 @@ func newSaramaConfig(ctx context.Context, o *options) (*sarama.Config, error) { err = completeSaramaSASLConfig(ctx, config, o) if err != nil { - return nil, errors.WrapError(errors.ErrKafkaInvalidConfig, err) + return nil, err } kafkaVersion, err := getKafkaVersion(config, o) @@ -130,7 +130,7 @@ func newSaramaConfig(ctx context.Context, o *options) (*sarama.Config, error) { if o.IsAssignedVersion { version, err := sarama.ParseKafkaVersion(o.Version) if err != nil { - return nil, errors.WrapError(errors.ErrKafkaInvalidVersion, err) + return nil, errors.WrapError(errors.ErrKafkaInvalidConfig, err) } config.Version = version if !version.IsAtLeast(maxKafkaVersion) && version.String() != kafkaVersion.String() { @@ -177,7 +177,7 @@ func completeSaramaSASLConfig(ctx context.Context, config *sarama.Config, o *opt case SASLTypeOAuth: p, err := newTokenProvider(ctx, o) if err != nil { - return errors.Trace(err) + return err } config.Net.SASL.TokenProvider = p } @@ -216,7 +216,7 @@ func getKafkaVersion(config *sarama.Config, o *options) (sarama.KafkaVersion, er if o.IsAssignedVersion { assignedVersion, err := sarama.ParseKafkaVersion(o.Version) if err != nil { - return assignedVersion, errors.WrapError(errors.ErrKafkaInvalidVersion, err) + return assignedVersion, errors.WrapError(errors.ErrKafkaInvalidConfig, err) } if !assignedVersion.IsAtLeast(maxKafkaVersion) && assignedVersion.String() != targetVersion.String() { log.Warn("The Kafka version you assigned may not be correct. "+ diff --git a/pkg/sink/kafka/sarama_config_test.go b/pkg/sink/kafka/sarama_config_test.go index bfb0147a1c..ca3adbbdcb 100644 --- a/pkg/sink/kafka/sarama_config_test.go +++ b/pkg/sink/kafka/sarama_config_test.go @@ -21,9 +21,9 @@ import ( "github.com/IBM/sarama" "github.com/gin-gonic/gin/binding" - "github.com/pingcap/errors" - commonType "github.com/pingcap/ticdc/pkg/common" + "github.com/pingcap/ticdc/pkg/common" "github.com/pingcap/ticdc/pkg/config" + "github.com/pingcap/ticdc/pkg/errors" "github.com/pingcap/ticdc/pkg/security" "github.com/stretchr/testify/require" ) @@ -85,6 +85,21 @@ func TestNewSaramaConfig(t *testing.T) { require.Equal(t, sarama.SASLMechanism("SCRAM-SHA-256"), cfg.Net.SASL.Mechanism) } +func TestNewSaramaConfigInvalidOAuthTokenURL(t *testing.T) { + options := NewOptions() + options.SASL = &security.SASL{ + SASLMechanism: security.OAuthMechanism, + OAuth2: security.OAuth2{ + TokenURL: "http://test.com/Segment%%2815197306101420000%29", + }, + } + + _, err := newSaramaConfig(t.Context(), options) + require.ErrorIs(t, err, errors.ErrKafkaInvalidConfig) + var escapeErr url.EscapeError + require.ErrorAs(t, err, &escapeErr) +} + func TestNewSaramaConfigMaxRetryFromSinkURI(t *testing.T) { t.Parallel() @@ -126,7 +141,7 @@ func TestNewSaramaConfigMaxRetryFromSinkURI(t *testing.T) { sinkURI, err := url.Parse(test.sinkURI) require.NoError(t, err) err = options.Apply( - commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), + common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink, ) diff --git a/pkg/sink/kafka/sarama_factory.go b/pkg/sink/kafka/sarama_factory.go index 650f346b4c..04a195cffb 100644 --- a/pkg/sink/kafka/sarama_factory.go +++ b/pkg/sink/kafka/sarama_factory.go @@ -46,19 +46,24 @@ func NewSaramaFactory( zap.Stringer("changefeedID", changefeedID), zap.Any("duration", duration)) } if err != nil { - return nil, errors.Trace(err) + return nil, err } admin, err := newAdminClient(changefeedID, o.BrokerEndpoints, config) if err != nil { - return nil, errors.Trace(err) + return nil, err } defer func() { admin.Close() }() +<<<<<<< HEAD if err = adjustOptions(ctx, admin, o, o.Topic); err != nil { return nil, errors.Trace(err) +======= + if err = adjustOptions(changefeedID, admin, o, o.Topic); err != nil { + return nil, err +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) } return &saramaFactory{ @@ -77,7 +82,7 @@ func newAdminClient(changefeedID common.ChangeFeedID, endpoints []string, config zap.Any("duration", duration), zap.Stringer("changefeedID", changefeedID)) } if err != nil { - return nil, errors.Trace(err) + return nil, errors.WrapError(errors.ErrNewKafkaSink, err) } start = time.Now() @@ -91,7 +96,7 @@ func newAdminClient(changefeedID common.ChangeFeedID, endpoints []string, config // `sarama.NewClusterAdminFromClient` does not take ownership of the client, // so we need to close it on failures to avoid leaking background goroutines. _ = client.Close() - return nil, errors.Trace(err) + return nil, errors.WrapError(errors.ErrNewKafkaSink, err) } return &saramaAdminClient{ client: client, @@ -103,7 +108,7 @@ func newAdminClient(changefeedID common.ChangeFeedID, endpoints []string, config func (f *saramaFactory) AdminClient(ctx context.Context) (ClusterAdminClient, error) { config, err := newSaramaConfig(ctx, f.option) if err != nil { - return nil, errors.WrapError(errors.ErrKafkaNewProducer, err) + return nil, err } return newAdminClient(f.changefeedID, f.option.BrokerEndpoints, config) } @@ -113,18 +118,19 @@ func (f *saramaFactory) AdminClient(ctx context.Context) (ClusterAdminClient, er func (f *saramaFactory) SyncProducer(ctx context.Context) (SyncProducer, error) { config, err := newSaramaConfig(ctx, f.option) if err != nil { - return nil, errors.WrapError(errors.ErrKafkaNewProducer, err) + return nil, err } config.MetricRegistry = f.metricRegistry client, err := sarama.NewClient(f.option.BrokerEndpoints, config) if err != nil { - return nil, errors.WrapError(errors.ErrKafkaNewProducer, err) + return nil, errors.WrapError(errors.ErrNewKafkaSink, err) } p, err := sarama.NewSyncProducerFromClient(client) if err != nil { - return nil, errors.WrapError(errors.ErrKafkaNewProducer, err) + _ = client.Close() + return nil, errors.WrapError(errors.ErrNewKafkaSink, err) } return &saramaSyncProducer{ @@ -140,25 +146,25 @@ func (f *saramaFactory) SyncProducer(ctx context.Context) (SyncProducer, error) func (f *saramaFactory) AsyncProducer(ctx context.Context) (AsyncProducer, error) { config, err := newSaramaConfig(ctx, f.option) if err != nil { - return nil, errors.WrapError(errors.ErrKafkaNewProducer, err) + return nil, err } config.MetricRegistry = f.metricRegistry client, err := sarama.NewClient(f.option.BrokerEndpoints, config) if err != nil { - return nil, errors.WrapError(errors.ErrKafkaNewProducer, err) + return nil, errors.WrapError(errors.ErrNewKafkaSink, err) } p, err := sarama.NewAsyncProducerFromClient(client) if err != nil { - return nil, errors.WrapError(errors.ErrKafkaNewProducer, err) + _ = client.Close() + return nil, errors.WrapError(errors.ErrNewKafkaSink, err) } return &saramaAsyncProducer{ client: client, producer: p, changefeedID: f.changefeedID, closed: atomic.NewBool(false), - failpointCh: make(chan *sarama.ProducerError, 1), }, nil } diff --git a/pkg/sink/kafka/sarama_sync_producer.go b/pkg/sink/kafka/sarama_sync_producer.go index 9d5efdfb0b..754d1db9e1 100644 --- a/pkg/sink/kafka/sarama_sync_producer.go +++ b/pkg/sink/kafka/sarama_sync_producer.go @@ -17,11 +17,10 @@ import ( "time" "github.com/IBM/sarama" - "github.com/pingcap/failpoint" "github.com/pingcap/log" - commonType "github.com/pingcap/ticdc/pkg/common" + "github.com/pingcap/ticdc/pkg/common" "github.com/pingcap/ticdc/pkg/errors" - "github.com/pingcap/ticdc/pkg/sink/codec/common" + codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" "go.uber.org/atomic" "go.uber.org/zap" ) @@ -38,15 +37,15 @@ type saramaSyncProducerClient interface { } type saramaSyncProducer struct { - id commonType.ChangeFeedID + id common.ChangeFeedID client saramaSyncClient producer saramaSyncProducerClient closed *atomic.Bool } -func (p *saramaSyncProducer) SendMessage(topic string, partitionNum int32, message *common.Message) error { +func (p *saramaSyncProducer) SendMessage(topic string, partitionNum int32, message *codecCommon.Message) error { if p.closed.Load() { - return errors.ErrKafkaProducerClosed.GenWithStackByArgs() + return errors.ErrKafkaSinkClosed.GenWithStackByArgs() } msg := &sarama.ProducerMessage{ @@ -56,24 +55,20 @@ func (p *saramaSyncProducer) SendMessage(topic string, partitionNum int32, messa Partition: partitionNum, } _, _, err := p.producer.SendMessage(msg) - - failpoint.Inject("KafkaSinkSyncSendMessageError", func() { - err = errors.WrapError(errors.ErrKafkaSendMessage, errors.New("kafka sink sync send message injected error")) - }) - if err != nil { - err = AnnotateEventError( - p.id.Keyspace(), - p.id.Name(), - message.LogInfo, - err, - ) + if err == nil { + return nil } + log.Error("send message to kafka failed", + zap.String("keyspace", p.id.Keyspace()), + zap.String("changefeed", p.id.Name()), + zap.String("eventContext", BuildEventLogContext(p.id.Keyspace(), p.id.Name(), message.LogInfo)), + zap.Error(err)) return errors.WrapError(errors.ErrKafkaSendMessage, err) } -func (p *saramaSyncProducer) SendMessages(topic string, partitionNum int32, message *common.Message) error { +func (p *saramaSyncProducer) SendMessages(topic string, partitionNum int32, message *codecCommon.Message) error { if p.closed.Load() { - return errors.ErrKafkaProducerClosed.GenWithStackByArgs() + return errors.ErrKafkaSinkClosed.GenWithStackByArgs() } msgs := make([]*sarama.ProducerMessage, partitionNum) @@ -86,18 +81,14 @@ func (p *saramaSyncProducer) SendMessages(topic string, partitionNum int32, mess } } err := p.producer.SendMessages(msgs) - - failpoint.Inject("KafkaSinkSyncSendMessagesError", func() { - err = errors.WrapError(errors.ErrKafkaSendMessage, errors.New("kafka sink sync send messages injected error")) - }) - if err != nil { - err = AnnotateEventError( - p.id.Keyspace(), - p.id.Name(), - message.LogInfo, - err, - ) + if err == nil { + return nil } + log.Error("send message to kafka failed", + zap.String("keyspace", p.id.Keyspace()), + zap.String("changefeed", p.id.Name()), + zap.String("eventContext", BuildEventLogContext(p.id.Keyspace(), p.id.Name(), message.LogInfo)), + zap.Error(err)) return errors.WrapError(errors.ErrKafkaSendMessage, err) } diff --git a/pkg/sink/kafka/sarama_sync_producer_test.go b/pkg/sink/kafka/sarama_sync_producer_test.go index 281af9c634..cd1b0078fa 100644 --- a/pkg/sink/kafka/sarama_sync_producer_test.go +++ b/pkg/sink/kafka/sarama_sync_producer_test.go @@ -14,15 +14,25 @@ package kafka import ( - "errors" + "context" + "io" + "strings" "testing" "github.com/IBM/sarama" +<<<<<<< HEAD "github.com/pingcap/ticdc/pkg/common" +======= + "github.com/golang/mock/gomock" + "github.com/pingcap/ticdc/pkg/common" + "github.com/pingcap/ticdc/pkg/errors" + codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) "github.com/stretchr/testify/require" "go.uber.org/atomic" ) +<<<<<<< HEAD type testSyncProducerClient struct { closeCalls int closeErr error @@ -84,4 +94,121 @@ func TestSaramaSyncProducerCloseStillClosesProducerWhenClientCloseFails(t *testi require.Equal(t, 1, client.closeCalls) require.Equal(t, 1, producer.closeCalls) +======= +func TestProducerRejectsSendAfterClose(t *testing.T) { + t.Parallel() + + message := &codecCommon.Message{} + syncProducer := &saramaSyncProducer{closed: atomic.NewBool(true)} + require.ErrorIs(t, syncProducer.SendMessage("topic", 1, message), errors.ErrKafkaSinkClosed) + require.ErrorIs(t, syncProducer.SendMessages("topic", 1, message), errors.ErrKafkaSinkClosed) + + asyncProducer := &saramaAsyncProducer{closed: atomic.NewBool(true)} + require.ErrorIs(t, asyncProducer.AsyncSend(context.Background(), "topic", 0, message), errors.ErrKafkaSinkClosed) +} + +func TestSyncProducerClose(t *testing.T) { + tests := []struct { + name string + clientCloseErr error + }{ + { + name: "closes client and producer", + }, + { + name: "still closes producer when client close fails", + clientCloseErr: io.ErrClosedPipe, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + client := NewMocksaramaSyncClient(ctrl) + producer := NewMocksaramaSyncProducerClient(ctrl) + gomock.InOrder( + client.EXPECT().Close().Return(test.clientCloseErr), + producer.EXPECT().Close().Return(nil), + ) + + p := &saramaSyncProducer{ + id: common.NewChangeFeedIDWithName("test", "default"), + client: client, + producer: producer, + closed: atomic.NewBool(false), + } + + p.Close() + }) + } +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) +} + +func TestSyncProducerErrorWrappedOnce(t *testing.T) { + cause := io.ErrClosedPipe + tests := []struct { + name string + expectSend func(*MocksaramaSyncProducerClient) + send func(*saramaSyncProducer, *codecCommon.Message) error + }{ + { + name: "single message", + expectSend: func(producer *MocksaramaSyncProducerClient) { + producer.EXPECT().SendMessage(gomock.Any()).Return(int32(0), int64(0), cause) + }, + send: func(producer *saramaSyncProducer, message *codecCommon.Message) error { + return producer.SendMessage("topic", 0, message) + }, + }, + { + name: "message batch", + expectSend: func(producer *MocksaramaSyncProducerClient) { + producer.EXPECT().SendMessages(gomock.Any()).Return(cause) + }, + send: func(producer *saramaSyncProducer, message *codecCommon.Message) error { + return producer.SendMessages("topic", 1, message) + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + producer := NewMocksaramaSyncProducerClient(ctrl) + test.expectSend(producer) + p := &saramaSyncProducer{ + id: common.NewChangeFeedIDWithName("test", "default"), + producer: producer, + closed: atomic.NewBool(false), + } + message := &codecCommon.Message{LogInfo: &codecCommon.MessageLogInfo{}} + + err := test.send(p, message) + + requireKafkaSendError(t, err, cause) + }) + } +} + +func TestAsyncProducerErrorWrappedOnce(t *testing.T) { + cause := io.ErrClosedPipe + producer := &saramaAsyncProducer{ + changefeedID: common.NewChangeFeedIDWithName("test", "default"), + } + err := producer.handleProducerError(&sarama.ProducerError{ + Err: cause, + Msg: &sarama.ProducerMessage{Metadata: &messageMetadata{ + logInfo: &codecCommon.MessageLogInfo{}, + }}, + }) + + requireKafkaSendError(t, err, cause) +} + +func requireKafkaSendError(t *testing.T, err, cause error) { + t.Helper() + require.ErrorIs(t, err, errors.ErrKafkaSendMessage) + require.ErrorIs(t, err, cause) + require.Equal(t, 1, strings.Count(err.Error(), string(errors.ErrKafkaSendMessage.RFCCode()))) + require.NotContains(t, err.Error(), "keyspace=test") } diff --git a/pkg/util/external_storage.go b/pkg/util/external_storage.go index f276c69638..43e5fe3010 100644 --- a/pkg/util/external_storage.go +++ b/pkg/util/external_storage.go @@ -64,7 +64,7 @@ func getExternalStorage( ) (storage.ExternalStorage, error) { backEnd, err := storage.ParseBackend(uri, opts) if err != nil { - return nil, errors.Trace(err) + return nil, errors.WrapError(errors.ErrExternalStorageAPI, err) } ret, err := storage.New(ctx, backEnd, &storage.ExternalStorageOptions{ @@ -72,7 +72,7 @@ func getExternalStorage( S3Retryer: retryer, }) if err != nil { - return nil, errors.WrapError(errors.ErrFailToCreateExternalStorage, err) + return nil, errors.WrapError(errors.ErrExternalStorageAPI, err) } defer func() { if err != nil { @@ -83,7 +83,7 @@ func getExternalStorage( // Check the connection and ignore the returned bool value, since we don't care if the file exists. _, err = ret.FileExists(ctx, "test") if err != nil { - return nil, errors.WrapError(errors.ErrFailToCreateExternalStorage, err) + return nil, errors.WrapError(errors.ErrExternalStorageAPI, err) } return ret, nil } diff --git a/tests/integration_tests/http_api/util/test_case.py b/tests/integration_tests/http_api/util/test_case.py index bd1f27296c..b7120e6200 100644 --- a/tests/integration_tests/http_api/util/test_case.py +++ b/tests/integration_tests/http_api/util/test_case.py @@ -1,4 +1,5 @@ import sys +import os import requests as rq from requests.exceptions import RequestException import time @@ -175,7 +176,9 @@ def create_changefeed(sink_uri): }) headers = {"Content-Type": "application/json"} resp = rq.post(url, data=data, headers=headers) - assert "CDC:ErrKafkaNewProducer" in resp.text, f"{resp.text}" + expected_error = "CDC:ErrKafkaNewProducer" if os.getenv( + "TICDC_NEWARCH") == "false" else "CDC:ErrNewKafkaSink" + assert expected_error in resp.text, f"{resp.text}" assert "not found, ResolveEndpointV2" not in resp.text, f"{resp.text}" print("pass test: create changefeed") diff --git a/tests/integration_tests/kafka_big_messages/run.sh b/tests/integration_tests/kafka_big_messages/run.sh index 0628eaa92e..c36f7068e9 100755 --- a/tests/integration_tests/kafka_big_messages/run.sh +++ b/tests/integration_tests/kafka_big_messages/run.sh @@ -8,6 +8,178 @@ WORK_DIR=$OUT_DIR/$TEST_NAME CDC_BINARY=cdc.test SINK_TYPE=$1 +<<<<<<< HEAD +======= +STATE_WAIT_TIMEOUT_SECONDS=30 +STATE_CHECK_INTERVAL_SECONDS=1 +TABLE_CHECK_RETRIES=15 +BATCH_LIMIT=262144 +SMALL_TOPIC_LIMIT=524288 +LARGE_TOPIC_LIMIT=2097152 +ROW_BYTES=1048576 +SCHEMA_REGISTRY_URI=http://127.0.0.1:8088 +GENERATOR_DIR=$CUR/../../utils/gen_kafka_big_messages +consumer_pid="" + +function start_schema_registry() { + if curl -o /dev/null -s "$SCHEMA_REGISTRY_URI"; then + return + fi + + echo "Starting schema registry..." + ./bin/bin/schema-registry-start -daemon ./bin/etc/schema-registry/schema-registry.properties + local i=0 + while ! curl -o /dev/null -s "$SCHEMA_REGISTRY_URI"; do + i=$((i + 1)) + if [ "$i" -gt 30 ]; then + echo "Failed to start schema registry" + exit 1 + fi + sleep 2 + done + curl -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" --data '{"compatibility": "NONE"}' "$SCHEMA_REGISTRY_URI/config" +} + +function build_message_generator() { + if [ ! -f "$GENERATOR_DIR/gen_kafka_big_messages" ]; then + (cd "$GENERATOR_DIR" && GO111MODULE=on go build) + fi +} + +function kafka_sink_uri() { + local topic_name=$1 + local protocol=$2 + local extra_params=$3 + local sink_uri="kafka://127.0.0.1:9092/${topic_name}?protocol=${protocol}&partition-num=1&kafka-version=${KAFKA_VERSION}&max-message-bytes=${BATCH_LIMIT}" + if [ "$extra_params" != "" ]; then + sink_uri="${sink_uri}&${extra_params}" + fi + echo "$sink_uri" +} + +function start_kafka_consumer() { + local work_dir=$1 + local sink_uri=$2 + local schema_registry_uri=$3 + local protocol_case=$4 + local downstream_uri="mysql://root@${DOWN_TIDB_HOST}:${DOWN_TIDB_PORT}/?safe-mode=true&batch-dml-enable=false&enable-ddl-ts=false" + local args=( + --log-file "$work_dir/cdc_kafka_consumer.log" + --log-level debug + --upstream-uri "$sink_uri" + --downstream-uri "$downstream_uri" + ) + if [ "$schema_registry_uri" != "" ]; then + args+=(--schema-registry-uri "$schema_registry_uri") + fi + if [[ "$protocol_case" == simple_* ]]; then + args+=(--upstream-tidb-dsn "root@tcp(${UP_TIDB_HOST}:${UP_TIDB_PORT})/?") + fi + + cdc_kafka_consumer "${args[@]}" >>"$work_dir/cdc_kafka_consumer_stdout.log" 2>&1 & + consumer_pid=$! +} + +function stop_kafka_consumer() { + if [ "$consumer_pid" != "" ]; then + kill -9 "$consumer_pid" 2>/dev/null || true + wait "$consumer_pid" 2>/dev/null || true + consumer_pid="" + fi +} + +function wait_changefeed_state() { + local pd_addr=$1 + local changefeed_id=$2 + local expected_state=$3 + local expected_error=$4 + local deadline=$((SECONDS + STATE_WAIT_TIMEOUT_SECONDS)) + + while true; do + if check_changefeed_state "$pd_addr" "$changefeed_id" "$expected_state" "$expected_error" ""; then + return + fi + if [ "$SECONDS" -ge "$deadline" ]; then + echo "changefeed $changefeed_id did not reach state $expected_state within ${STATE_WAIT_TIMEOUT_SECONDS}s" + return 1 + fi + sleep "$STATE_CHECK_INTERVAL_SECONDS" + done +} + +function render_diff_config() { + local work_dir=$1 + local database_name=$2 + local diff_config=$3 + + sed -e "s/database_name/${database_name}/g" \ + -e "s|/tmp/tidb_cdc_test/kafka_big_messages/sync_diff/output|${work_dir}/sync_diff/output|g" \ + "$CUR/conf/diff_config.toml" >"$diff_config" +} + +function run_protocol_case() { + local protocol_case=$1 + local protocol=$2 + local schema_registry_uri=$3 + local extra_params=$4 + local topic_case=${protocol_case//_/-} + local topic_name="big-message-${topic_case}-${RANDOM}" + local changefeed_id="kafka-big-messages-${topic_case}" + local database_name="kafka_big_messages_${protocol_case}" + local work_dir="$WORK_DIR/$protocol_case" + local sql_file="$work_dir/test.sql" + local diff_config="$work_dir/diff_config.toml" + local pd_addr="http://${UP_PD_HOST_1}:${UP_PD_PORT_1}" + local sink_uri + local initial_topic_limit=$SMALL_TOPIC_LIMIT + local expected_error=ErrMessageTooLarge + if [ "$protocol_case" = "async_error" ]; then + initial_topic_limit=$LARGE_TOPIC_LIMIT + expected_error=ErrKafkaSendMessage + fi + + mkdir -p "$work_dir" + render_diff_config "$work_dir" "$database_name" "$diff_config" + kafka_topic --topic "$topic_name" --max-message-bytes "$initial_topic_limit" + local start_ts + start_ts=$(run_cdc_cli_tso_query "$UP_PD_HOST_1" "$UP_PD_PORT_1") + sink_uri=$(kafka_sink_uri "$topic_name" "$protocol" "$extra_params") + + if [ "$schema_registry_uri" != "" ]; then + cdc_cli_changefeed create --start-ts="$start_ts" --sink-uri="$sink_uri" --schema-registry="$schema_registry_uri" -c "$changefeed_id" + else + cdc_cli_changefeed create --start-ts="$start_ts" --sink-uri="$sink_uri" -c "$changefeed_id" + fi + start_kafka_consumer "$work_dir" "$sink_uri" "$schema_registry_uri" "$protocol_case" + wait_changefeed_state "$pd_addr" "$changefeed_id" "normal" "null" + + # Lower the topic limit after the producer has started. The encoder and + # producer still accept the message, then Kafka rejects it asynchronously. + if [ "$protocol_case" = "async_error" ]; then + local ready_database="${database_name}_ready" + run_sql "CREATE DATABASE ${ready_database}; CREATE TABLE ${ready_database}.ready(id INT PRIMARY KEY); INSERT INTO ${ready_database}.ready VALUES (1)" "$UP_TIDB_HOST" "$UP_TIDB_PORT" + ensure "$TABLE_CHECK_RETRIES" "run_sql 'SELECT id FROM ${ready_database}.ready' '$DOWN_TIDB_HOST' '$DOWN_TIDB_PORT' && check_contains 'id: 1'" + kafka_topic --topic "$topic_name" --max-message-bytes "$SMALL_TOPIC_LIMIT" --alter + fi + + "$GENERATOR_DIR/gen_kafka_big_messages" --row-bytes="$ROW_BYTES" --row-count=1 --database-name="$database_name" --table-name=test --sql-file-path="$sql_file" + run_sql_file "$sql_file" "$UP_TIDB_HOST" "$UP_TIDB_PORT" + run_sql "CREATE TABLE ${database_name}.finish_mark(id INT PRIMARY KEY)" "$UP_TIDB_HOST" "$UP_TIDB_PORT" + + wait_changefeed_state "$pd_addr" "$changefeed_id" "warning" "$expected_error" + + # Only increase Kafka's topic limit. TiCDC must recreate the sink, read the + # new limit, and resume without updating, pausing, or resuming the changefeed. + kafka_topic --topic "$topic_name" --max-message-bytes "$LARGE_TOPIC_LIMIT" --alter + wait_changefeed_state "$pd_addr" "$changefeed_id" "normal" "null" + check_table_exists "${database_name}.finish_mark" "$DOWN_TIDB_HOST" "$DOWN_TIDB_PORT" "$TABLE_CHECK_RETRIES" + check_sync_diff "$work_dir" "$diff_config" + + cdc_cli_changefeed remove -c "$changefeed_id" + stop_kafka_consumer +} + +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) function run() { # test kafka sink only in this case if [ "$SINK_TYPE" != "kafka" ]; then @@ -15,7 +187,18 @@ function run() { fi rm -rf $WORK_DIR && mkdir -p $WORK_DIR +<<<<<<< HEAD start_tidb_cluster --workdir $WORK_DIR +======= + local cases=( + "canal_json|canal-json||enable-tidb-extension=true" + "open_protocol|open-protocol||" + "async_error|open-protocol||max-retry=0" + "simple_json|simple||" + "simple_avro|simple||encoding-format=avro" + "avro|avro|$SCHEMA_REGISTRY_URI|enable-tidb-extension=true&avro-enable-watermark=true&avro-decimal-handling-mode=string&avro-bigint-unsigned-handling-mode=string" + ) +>>>>>>> fa340f118 (kafka: unify sink errors and replace failpoint tests (#5786)) TOPIC_NAME="big-message-test-$RANDOM" diff --git a/tests/integration_tests/kafka_log_info/run.sh b/tests/integration_tests/kafka_log_info/run.sh deleted file mode 100755 index de4ae60465..0000000000 --- a/tests/integration_tests/kafka_log_info/run.sh +++ /dev/null @@ -1,134 +0,0 @@ -#!/bin/bash - -set -eu - -CUR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) -source $CUR/../_utils/test_prepare -WORK_DIR=$OUT_DIR/$TEST_NAME -CDC_BINARY=cdc.test -SINK_TYPE=$1 - -MAX_RETRIES=20 -pd_addr="http://$UP_PD_HOST_1:$UP_PD_PORT_1" -protocols=("open-protocol" "canal-json" "simple") - -declare -r DB_NAME="kafka_log_info" - -function build_sink_uri() { - local protocol=$1 - local topic=$2 - echo "kafka://127.0.0.1:9092/$topic?protocol=$protocol&partition-num=1&kafka-version=${KAFKA_VERSION}&max-message-bytes=10485760" -} - -function cleanup_changefeed() { - local id=$1 - cdc_cli_changefeed remove --pd="${pd_addr}" --changefeed-id="$id" >/dev/null 2>&1 || true - # Wait for the changefeed removal to be fully persisted and visible to all components. - # Otherwise, the next TiCDC process may resume the leftover changefeed and consume failpoints unexpectedly. - sleep 5 -} - -function assert_no_changefeeds() { - local feed_count - feed_count=$(cdc_cli_changefeed list --pd="$pd_addr" | grep -v "Command to ticdc" | jq '.|length') - if [[ "$feed_count" != "0" ]]; then - echo "[$(date)] <<<<< existing changefeeds detected before create, count: ${feed_count} >>>>>" - cdc_cli_changefeed list --pd="$pd_addr" | grep -v "Command to ticdc" | jq . - exit 1 - fi -} - -function stop_cdc() { - cleanup_process $CDC_BINARY - export GO_FAILPOINTS="" -} - -function start_cdc_with_failpoint() { - local failpoints=$1 - export GO_FAILPOINTS="$failpoints" - run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY --addr "127.0.0.1:8300" --pd $pd_addr -} - -function test_dml_log_info() { - local protocol=$1 - local topic="kafka-log-info-dml-${protocol}-${RANDOM}" - local changefeed_id="kafka-log-info-${protocol}-dml" - local sink_uri=$(build_sink_uri $protocol $topic) - - run_sql "DROP TABLE IF EXISTS ${DB_NAME}.dml_table" ${UP_TIDB_HOST} ${UP_TIDB_PORT} - run_sql "CREATE TABLE ${DB_NAME}.dml_table(id INT PRIMARY KEY AUTO_INCREMENT, val INT);" ${UP_TIDB_HOST} ${UP_TIDB_PORT} - - start_cdc_with_failpoint 'github.com/pingcap/ticdc/pkg/sink/kafka/KafkaSinkAsyncSendError=1*return(true)' - assert_no_changefeeds - cdc_cli_changefeed create --pd=$pd_addr --sink-uri="$sink_uri" --changefeed-id="$changefeed_id" - - run_sql "INSERT INTO ${DB_NAME}.dml_table(val) VALUES (1);" ${UP_TIDB_HOST} ${UP_TIDB_PORT} - - local pattern='eventType=dml.*\\"Table\\":\\"dml_table\\".*\\"StartTs\\":.*\\"CommitTs\\":' - ensure $MAX_RETRIES "check_logs_contains $WORK_DIR '$pattern' ''" - - cleanup_changefeed $changefeed_id - stop_cdc -} - -function test_ddl_log_info() { - local protocol=$1 - local topic="kafka-log-info-ddl-${protocol}-${RANDOM}" - local changefeed_id="kafka-log-info-${protocol}-ddl" - local sink_uri=$(build_sink_uri $protocol $topic) - - run_sql "DROP TABLE IF EXISTS ${DB_NAME}.ddl_table;" ${UP_TIDB_HOST} ${UP_TIDB_PORT} - - start_cdc_with_failpoint 'github.com/pingcap/ticdc/pkg/sink/kafka/KafkaSinkSyncSendMessageError=1*return(true);github.com/pingcap/ticdc/pkg/sink/kafka/KafkaSinkSyncSendMessagesError=1*return(true)' - assert_no_changefeeds - cdc_cli_changefeed create --pd=$pd_addr --sink-uri="$sink_uri" --changefeed-id="$changefeed_id" - - run_sql "CREATE TABLE ${DB_NAME}.ddl_table(id INT PRIMARY KEY);" ${UP_TIDB_HOST} ${UP_TIDB_PORT} - - local ddl_pattern="eventType=ddl.*ddlQuery=.*CREATE TABLE*" - ensure $MAX_RETRIES "check_logs_contains $WORK_DIR '$ddl_pattern' ''" - - cleanup_changefeed $changefeed_id - stop_cdc -} - -function test_checkpoint_log_info() { - local protocol=$1 - local topic="kafka-log-info-checkpoint-${protocol}-${RANDOM}" - local changefeed_id="kafka-log-info-${protocol}-checkpoint" - local sink_uri=$(build_sink_uri $protocol $topic) - - start_cdc_with_failpoint 'github.com/pingcap/ticdc/pkg/sink/kafka/KafkaSinkSyncSendMessagesError=1*return(true)' - assert_no_changefeeds - cdc_cli_changefeed create --pd=$pd_addr --sink-uri="$sink_uri" --changefeed-id="$changefeed_id" - - ensure $MAX_RETRIES "check_logs_contains $WORK_DIR 'eventType=checkpoint.*checkpointTs=' ''" - - cleanup_changefeed $changefeed_id - stop_cdc -} - -function run() { - if [ "$SINK_TYPE" != "kafka" ]; then - echo "skip kafka_log_info for sink type $SINK_TYPE" - return - fi - - rm -rf $WORK_DIR && mkdir -p $WORK_DIR - start_tidb_cluster --workdir $WORK_DIR - run_sql "DROP DATABASE IF EXISTS ${DB_NAME};" ${UP_TIDB_HOST} ${UP_TIDB_PORT} - run_sql "CREATE DATABASE ${DB_NAME};" ${UP_TIDB_HOST} ${UP_TIDB_PORT} - - for protocol in "${protocols[@]}"; do - test_dml_log_info $protocol - test_ddl_log_info $protocol - test_checkpoint_log_info $protocol - done -} - -trap "stop_cdc; stop_tidb_cluster" EXIT - -run $* -check_logs $WORK_DIR - -echo "[$(date)] <<<<<< run test case $TEST_NAME success! >>>>>>" diff --git a/tests/integration_tests/mq_sink_error_resume/run.sh b/tests/integration_tests/mq_sink_error_resume/run.sh index 6fb047381f..d4f66e2f91 100755 --- a/tests/integration_tests/mq_sink_error_resume/run.sh +++ b/tests/integration_tests/mq_sink_error_resume/run.sh @@ -13,8 +13,7 @@ DB_COUNT=4 MAX_RETRIES=20 function run() { - # test MQ sink only in this case - if [ "$SINK_TYPE" != "kafka" ] && [ "$SINK_TYPE" != "pulsar" ]; then + if [ "$SINK_TYPE" != "pulsar" ]; then return fi @@ -24,22 +23,14 @@ function run() { pd_addr="http://$UP_PD_HOST_1:$UP_PD_PORT_1" TOPIC_NAME="ticdc-mq-sink-error-resume-test-$RANDOM" - case $SINK_TYPE in - kafka) SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=open-protocol&partition-num=4&kafka-version=${KAFKA_VERSION}&max-message-bytes=10485760" ;; - pulsar) - run_pulsar_cluster $WORK_DIR normal - SINK_URI="pulsar://127.0.0.1:6650/$TOPIC_NAME?protocol=canal-json&enable-tidb-extension=true" - ;; - esac - # Return an failpoint error to fail a kafka changefeed. + run_pulsar_cluster $WORK_DIR normal + SINK_URI="pulsar://127.0.0.1:6650/$TOPIC_NAME?protocol=canal-json&enable-tidb-extension=true" + # Return one failpoint error to fail the changefeed. # Note we return one error for the failpoint, if owner retry changefeed frequently, it may break the test. - export GO_FAILPOINTS='github.com/pingcap/ticdc/pkg/sink/kafka/KafkaSinkAsyncSendError=1*return(true);github.com/pingcap/ticdc/downstreamadapter/sink/pulsar/PulsarSinkAsyncSendError=1*return(true)' + export GO_FAILPOINTS='github.com/pingcap/ticdc/downstreamadapter/sink/pulsar/PulsarSinkAsyncSendError=1*return(true)' run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY --addr "127.0.0.1:8300" --pd $pd_addr changefeed_id=$(cdc_cli_changefeed create --pd=$pd_addr --sink-uri="$SINK_URI" | grep '^ID:' | head -n1 | awk '{print $2}') - case $SINK_TYPE in - kafka) run_kafka_consumer $WORK_DIR "kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=open-protocol&partition-num=4&version=${KAFKA_VERSION}&max-message-bytes=10485760" ;; - pulsar) run_pulsar_consumer --upstream-uri $SINK_URI ;; - esac + run_pulsar_consumer --upstream-uri $SINK_URI run_sql "CREATE DATABASE mq_sink_error_resume;" ${UP_TIDB_HOST} ${UP_TIDB_PORT} run_sql "CREATE table mq_sink_error_resume.t1(id int primary key auto_increment, val int);" ${UP_TIDB_HOST} ${UP_TIDB_PORT} diff --git a/tests/integration_tests/run_heavy_it_in_ci.sh b/tests/integration_tests/run_heavy_it_in_ci.sh index 1da9b6b814..c821353511 100755 --- a/tests/integration_tests/run_heavy_it_in_ci.sh +++ b/tests/integration_tests/run_heavy_it_in_ci.sh @@ -87,7 +87,7 @@ kafka_groups=( # 'kafka_simple_claim_check kafka_simple_claim_check_avro tidb_mysql_test' 'kafka_simple_claim_check kafka_simple_claim_check_avro' # G09 - 'kafka_simple_handle_key_only kafka_simple_handle_key_only_avro mq_sink_error_resume multi_source' + 'kafka_simple_handle_key_only kafka_simple_handle_key_only_avro multi_source' # G10 'kafka_column_selector kafka_column_selector_avro ddl_with_random_move_table' # G11 diff --git a/tests/integration_tests/run_light_it_in_ci.sh b/tests/integration_tests/run_light_it_in_ci.sh index 3a3ec7488c..b3f53dd968 100755 --- a/tests/integration_tests/run_light_it_in_ci.sh +++ b/tests/integration_tests/run_light_it_in_ci.sh @@ -107,7 +107,7 @@ kafka_groups=( # G14 'kafka_simple_basic avro_basic debezium_basic fail_over_ddl_O update_changefeed_check_config' # G15 - 'kafka_simple_basic_avro split_region autorandom gc_safepoint kafka_log_info' + 'kafka_simple_basic_avro split_region autorandom gc_safepoint' ) # Resource allocation for pulsar light integration tests in CI pipelines: