From e2d01a2c9bfb2ba94f0a72f9d698f3d36c59a8c7 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Tue, 28 Jul 2026 11:58:30 +0800 Subject: [PATCH 1/7] unify errors --- .../sink/eventrouter/topic/expression.go | 7 +- .../sink/eventrouter/topic/expression_test.go | 4 +- downstreamadapter/sink/kafka/helper.go | 12 +-- downstreamadapter/sink/kafka/sink.go | 12 +-- downstreamadapter/sink/pulsar/helper.go | 2 +- .../sink/topicmanager/kafka_topic_manager.go | 12 +-- .../topicmanager/kafka_topic_manager_test.go | 12 ++- pkg/errors/error.go | 47 ++-------- pkg/errors/error_test.go | 20 +++++ pkg/sink/kafka/admin.go | 28 +++--- pkg/sink/kafka/admin_test.go | 40 +++++++++ pkg/sink/kafka/cluster_admin_client.go | 8 +- pkg/sink/kafka/cluster_admin_client_mock.go | 14 +-- pkg/sink/kafka/options.go | 90 +++++++++++-------- pkg/sink/kafka/options_test.go | 21 ++--- pkg/sink/kafka/sarama_async_producer.go | 4 +- pkg/sink/kafka/sarama_config.go | 8 +- pkg/sink/kafka/sarama_factory.go | 24 ++--- pkg/sink/kafka/sarama_sync_producer.go | 4 +- pkg/sink/kafka/sarama_sync_producer_test.go | 20 ++++- .../http_api/util/test_case.py | 2 +- 21 files changed, 222 insertions(+), 169 deletions(-) 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 30a835d1b8..a817e1fe0a 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 7d2e78ca54..f0cdf23852 100644 --- a/downstreamadapter/sink/kafka/helper.go +++ b/downstreamadapter/sink/kafka/helper.go @@ -72,23 +72,23 @@ func newKafkaSinkComponent( }() protocol, err := helper.GetProtocol(utils.GetOrZero(sinkConfig.Protocol)) if err != nil { - return comp, config.ProtocolUnknown, errors.Trace(err) + return comp, config.ProtocolUnknown, err } topic, err := helper.GetTopic(sinkURI) if err != nil { - return comp, protocol, errors.Trace(err) + return comp, protocol, err } options := kafka.NewOptions() if err = options.Apply(changefeedID, sinkURI, sinkConfig); err != nil { - return comp, protocol, errors.WrapError(errors.ErrKafkaInvalidConfig, err) + return comp, protocol, err } options.Topic = topic comp.factory, err = kafka.NewSaramaFactory(ctx, options, changefeedID) if err != nil { - return comp, protocol, errors.WrapError(errors.ErrKafkaNewProducer, err) + return comp, protocol, err } isAvroLike := protocol == config.ProtocolAvro || protocol == config.ProtocolDebeziumAvro @@ -128,7 +128,7 @@ func newKafkaSinkComponent( comp.adminClient, err = comp.factory.AdminClient(ctx) if err != nil { - return comp, protocol, errors.WrapError(errors.ErrKafkaNewProducer, err) + return comp, protocol, err } comp.topicManager, err = topicmanager.GetTopicManagerAndTryCreateTopic( @@ -139,7 +139,7 @@ func newKafkaSinkComponent( comp.adminClient, ) if err != nil { - return comp, protocol, errors.Trace(err) + return comp, protocol, err } return comp, protocol, nil } diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index 2a03050c17..be75f3ffcc 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -84,7 +84,7 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url. options := kafka.NewOptions() if err = options.Apply(changefeedID, uri, sinkConfig); err != nil { - return errors.WrapError(errors.ErrKafkaInvalidConfig, err) + return err } options.Topic = topic @@ -113,18 +113,18 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url. factory, err := kafka.NewSaramaFactory(ctx, options, changefeedID) if err != nil { - return errors.WrapError(errors.ErrKafkaNewProducer, err) + return err } adminClient, err := factory.AdminClient(ctx) if err != nil { - return errors.WrapError(errors.ErrKafkaNewProducer, err) + return err } defer adminClient.Close() topics, err := adminClient.GetTopicsMeta([]string{topic}, false) if err != nil { - return errors.Trace(err) + return err } if _, exists := topics[topic]; !exists { topicConfig := options.DeriveTopicConfig() @@ -142,7 +142,7 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url. ReplicationFactor: topicConfig.ReplicationFactor, }, true) if err != nil { - return errors.WrapError(errors.ErrKafkaCreateTopic, err) + return err } } @@ -158,7 +158,7 @@ func New( ) (*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, keyspaceID, protocol, comp) } diff --git a/downstreamadapter/sink/pulsar/helper.go b/downstreamadapter/sink/pulsar/helper.go index 12e1beda52..59e592d0de 100644 --- a/downstreamadapter/sink/pulsar/helper.go +++ b/downstreamadapter/sink/pulsar/helper.go @@ -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 6d70c84b52..e33929d484 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go @@ -61,7 +61,7 @@ func GetTopicManagerAndTryCreateTopic( ) if _, err := topicManager.CreateTopicAndWaitUntilVisible(ctx, topic); err != nil { - return nil, errors.WrapError(errors.ErrKafkaCreateTopic, err) + return nil, err } return topicManager, nil @@ -102,7 +102,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 @@ -262,7 +262,7 @@ func (m *kafkaTopicManager) createTopic( zap.Error(err), zap.Duration("duration", time.Since(start)), ) - return 0, errors.WrapError(errors.ErrKafkaCreateTopic, err) + return 0, err } log.Info( @@ -291,7 +291,7 @@ func (m *kafkaTopicManager) CreateTopicAndWaitUntilVisible( if kafka.IsAdminAuthorizationFailed(err) { return m.useConfiguredPartitionNum(topicName, err), nil } - return 0, errors.Trace(err) + return 0, err } if numPartition, ok := m.tryStoreTopicMeta(topicName, topicDetails); ok { return numPartition, nil @@ -311,12 +311,12 @@ func (m *kafkaTopicManager) CreateTopicAndWaitUntilVisible( if kafka.IsAdminAuthorizationFailed(err) { return m.useConfiguredPartitionNum(topicName, err), nil } - return 0, errors.Trace(err) + return 0, err } 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 45ee322c63..4ee0be636a 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go @@ -20,6 +20,7 @@ import ( "github.com/IBM/sarama" "github.com/golang/mock/gomock" "github.com/pingcap/ticdc/pkg/common" + "github.com/pingcap/ticdc/pkg/errors" "github.com/pingcap/ticdc/pkg/sink/kafka" "github.com/stretchr/testify/require" ) @@ -128,7 +129,7 @@ func TestCreateTopic(t *testing.T) { func(detail *kafka.TopicDetail, validateOnly bool) error { gotFailedTopicDetail = detail gotFailedTopicValidateOnly = validateOnly - return sarama.ErrInvalidReplicationFactor + return errors.WrapError(errors.ErrKafkaAdminAPI, sarama.ErrInvalidReplicationFactor, "create-topic", detail.Name) }), ) @@ -179,11 +180,8 @@ func TestCreateTopic(t *testing.T) { manager = newKafkaTopicManager(ctx, topic, changefeedID, adminClient, cfg) defer manager.Close() _, err = manager.CreateTopicAndWaitUntilVisible(ctx, topic) - 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) @@ -201,7 +199,7 @@ func TestCreateTopicValidatesReplicationFactor(t *testing.T) { adminClient.EXPECT().GetTopicsMeta([]string{topic}, false). Return(map[string]kafka.TopicDetail{}, nil), adminClient.EXPECT().GetBrokerConfig(kafka.MinInsyncReplicasConfigName). - Return("2", nil), + Return("2", true, nil), ) manager := newKafkaTopicManager( diff --git a/pkg/errors/error.go b/pkg/errors/error.go index e7925bfa3e..5dcb4160f3 100644 --- a/pkg/errors/error.go +++ b/pkg/errors/error.go @@ -127,50 +127,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 16087c889c..13833516ba 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()), @@ -164,7 +162,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)) } @@ -181,7 +179,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 c2e3f90e37..48d0894f24 100644 --- a/pkg/sink/kafka/admin_test.go +++ b/pkg/sink/kafka/admin_test.go @@ -14,13 +14,53 @@ package kafka import ( + stderrors "errors" "testing" + "github.com/IBM/sarama" "github.com/golang/mock/gomock" "github.com/pingcap/ticdc/pkg/common" + cerror "github.com/pingcap/ticdc/pkg/errors" "github.com/stretchr/testify/require" ) +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 := stderrors.New("describe cluster failed") + 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, cerror.ErrKafkaAdminAPI) + require.ErrorIs(t, err, cause) + }) +} + func TestAdminClientClose(t *testing.T) { tests := []struct { name string 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 index 9a67b63520..dfeebbd773 100644 --- a/pkg/sink/kafka/cluster_admin_client_mock.go +++ b/pkg/sink/kafka/cluster_admin_client_mock.go @@ -74,12 +74,13 @@ func (mr *MockClusterAdminClientMockRecorder) GetAllBrokers() *gomock.Call { } // GetBrokerConfig mocks base method. -func (m *MockClusterAdminClient) GetBrokerConfig(configName string) (string, error) { +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].(error) - return ret0, ret1 + ret1, _ := ret[1].(bool) + ret2, _ := ret[2].(error) + return ret0, ret1, ret2 } // GetBrokerConfig indicates an expected call of GetBrokerConfig. @@ -89,12 +90,13 @@ func (mr *MockClusterAdminClientMockRecorder) GetBrokerConfig(configName interfa } // GetTopicConfig mocks base method. -func (m *MockClusterAdminClient) GetTopicConfig(topicName, configName string) (string, error) { +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].(error) - return ret0, ret1 + ret1, _ := ret[1].(bool) + ret2, _ := ret[2].(error) + return ret0, ret1, ret2 } // GetTopicConfig indicates an expected call of GetTopicConfig. diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go index 506e6287ef..bf76f7f7a5 100644 --- a/pkg/sink/kafka/options.go +++ b/pkg/sink/kafka/options.go @@ -99,7 +99,9 @@ func requireAcksFromString(acks int) (RequiredAcks, error) { case int(NoResponse): return NoResponse, nil default: - return Unknown, errors.ErrKafkaInvalidRequiredAcks.GenWithStackByArgs(acks) + return Unknown, errors.ErrKafkaInvalidConfig.GenWithStack( + "invalid required acks %d, only support these values: "+ + "0(NoResponse), 1(WaitForLocal) and -1(WaitForAll)", acks) } } @@ -213,7 +215,7 @@ func (o *options) setPartitionNum(changefeedID common.ChangeFeedID, realPartitio // the real partition count, since messages would be dispatched to different // partitions, this could prevent potential correctness problems. if o.PartitionNum > realPartitionCount { - return errors.ErrKafkaInvalidPartitionNum.GenWithStack( + return errors.ErrKafkaInvalidConfig.GenWithStack( "the number of partition (%d) specified in sink-uri is more than that of actual topic (%d)", o.PartitionNum, realPartitionCount) } @@ -230,15 +232,15 @@ func (o *options) Apply(changefeedID common.ChangeFeedID, req := &http.Request{URL: sinkURI} urlParameter := &urlConfig{} if err = binding.Query.Bind(req, urlParameter); err != nil { - return errors.WrapError(errors.ErrMySQLInvalidConfig, err) + return errors.WrapError(errors.ErrKafkaInvalidConfig, err) } 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 { - return errors.ErrKafkaInvalidPartitionNum.GenWithStackByArgs(o.PartitionNum) + return errors.ErrKafkaInvalidConfig.GenWithStack("invalid partition num %d", o.PartitionNum) } } @@ -290,7 +292,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 } @@ -298,7 +300,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 } @@ -306,7 +308,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 } @@ -388,8 +390,7 @@ func (o *options) applyTLS(params *urlConfig) error { if o.Credential != nil && !o.Credential.IsEmpty() && !o.Credential.IsTLSEnabled() { - return errors.WrapError(errors.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") } // if enable-tls is not set, but credential files are set, @@ -402,8 +403,7 @@ func (o *options) applyTLS(params *urlConfig) error { enableTLS := *params.EnableTLS if o.Credential != nil && o.Credential.IsTLSEnabled() && !enableTLS { - return errors.WrapError(errors.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") } o.EnableTLS = enableTLS } else { @@ -494,8 +494,7 @@ 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)) - return errors.ErrKafkaInvalidConfig.GenWithStack( - "OAuth2 client secret is not base64 encoded") + return errors.ErrKafkaInvalidConfig.GenWithStack("OAuth2 client secret is not base64 encoded") } o.SASL.OAuth2.ClientSecret = string(decodedClientSecret) } @@ -517,7 +516,7 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf } if err := o.SASL.OAuth2.Validate(); err != nil { - return errors.ErrKafkaInvalidConfig.Wrap(err) + return errors.WrapError(errors.ErrKafkaInvalidConfig, err) } o.SASL.OAuth2.SetDefault() } @@ -562,7 +561,7 @@ func (c *AutoCreateTopicConfig) ValidateReplicationFactor(admin ClusterAdminClie return nil } - raw, err := admin.GetBrokerConfig(MinInsyncReplicasConfigName) + 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), @@ -570,9 +569,15 @@ func (c *AutoCreateTopicConfig) ValidateReplicationFactor(admin ClusterAdminClie 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 err + return errors.WrapError(errors.ErrKafkaAdminAPI, err, "parse-config", MinInsyncReplicasConfigName) } if int(c.ReplicationFactor) < minInsyncReplicas { @@ -605,7 +610,7 @@ func NewKafkaClientID(captureAddr string, clientID = commonInvalidChar.ReplaceAllString(clientID, "_") } if !validClientID.MatchString(clientID) { - return "", errors.ErrKafkaInvalidClientID.GenWithStackByArgs(clientID) + return "", errors.ErrKafkaInvalidConfig.GenWithStack("invalid kafka client ID %q", clientID) } return } @@ -621,7 +626,7 @@ func adjustOptions( ) error { topics, err := admin.GetTopicsMeta([]string{topic}, true) if err != nil { - return errors.Trace(err) + return err } info, exists := topics[topic] @@ -647,8 +652,8 @@ func adjustExistingTopicOption( topic string, info TopicDetail, ) error { - maxMessageBytes, err := getTopicMaxMessageBytes(admin, info.Name) - if err != nil { + 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)) @@ -665,7 +670,7 @@ func adjustExistingTopicOption( } if err = options.setPartitionNum(changefeedID, info.NumPartitions); err != nil { - return errors.Trace(err) + return err } return nil } @@ -678,8 +683,8 @@ func adjustNewTopicOptions( ) { // 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`. - messageMaxBytes, err := getBrokerMaxMessageBytes(admin) - if err != nil { + 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)) @@ -699,32 +704,38 @@ func adjustNewTopicOptions( func getTopicMaxMessageBytes( admin ClusterAdminClient, topic string, -) (int, error) { - raw, err := getTopicConfig( +) (int, bool, error) { + raw, found, err := getTopicConfig( admin, topic, TopicMaxMessageBytesConfigName, BrokerMessageMaxBytesConfigName, ) if err != nil { - return 0, errors.Trace(err) + return 0, false, err + } + if !found { + return 0, false, nil } maxMessageBytes, err := strconv.Atoi(raw) if err != nil { - return 0, errors.Trace(err) + return 0, false, errors.WrapError(errors.ErrKafkaAdminAPI, err, "parse-config", TopicMaxMessageBytesConfigName) } - return maxMessageBytes, nil + return maxMessageBytes, true, nil } -func getBrokerMaxMessageBytes(admin ClusterAdminClient) (int, error) { - raw, err := admin.GetBrokerConfig(BrokerMessageMaxBytesConfigName) +func getBrokerMaxMessageBytes(admin ClusterAdminClient) (int, bool, error) { + raw, found, err := admin.GetBrokerConfig(BrokerMessageMaxBytesConfigName) if err != nil { - return 0, errors.Trace(err) + return 0, false, err + } + if !found { + return 0, false, nil } messageMaxBytes, err := strconv.Atoi(raw) if err != nil { - return 0, errors.Trace(err) + return 0, false, errors.WrapError(errors.ErrKafkaAdminAPI, err, "parse-config", BrokerMessageMaxBytesConfigName) } - return messageMaxBytes, nil + return messageMaxBytes, true, nil } // getTopicConfig gets topic config by name. @@ -736,12 +747,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 d922166c40..ddc8c971bc 100644 --- a/pkg/sink/kafka/options_test.go +++ b/pkg/sink/kafka/options_test.go @@ -110,24 +110,21 @@ func (f *kafkaAdminFixture) getTopicsPartitionsNum( return result, nil } -func (f *kafkaAdminFixture) getBrokerConfig(configName string) (string, error) { +func (f *kafkaAdminFixture) getBrokerConfig(configName string) (string, bool, error) { if value, ok := f.brokerConfig[configName]; ok { - return value, nil + return value, true, nil } - return "", errors.ErrKafkaConfigNotFound.GenWithStack( - "cannot find the `%s` from the broker's configuration", configName) + return "", false, nil } -func (f *kafkaAdminFixture) getTopicConfig(topicName string, configName string) (string, error) { +func (f *kafkaAdminFixture) getTopicConfig(topicName string, configName string) (string, bool, error) { if _, ok := f.topics[topicName]; !ok { - return "", errors.ErrKafkaConfigNotFound.GenWithStack( - "cannot find the `%s` from the topic's configuration", topicName) + return "", false, nil } if value, ok := f.topicConfig[topicName][configName]; ok { - return value, nil + return value, true, nil } - return "", errors.ErrKafkaConfigNotFound.GenWithStack( - "cannot find the `%s` from the topic's configuration", configName) + return "", false, nil } func (f *kafkaAdminFixture) createTopic(detail *TopicDetail, _ bool) error { @@ -264,7 +261,7 @@ func TestCompleteOptions(t *testing.T) { 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)) + 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" @@ -362,7 +359,7 @@ func TestSetPartitionNum(t *testing.T) { options.PartitionNum = 3 err = options.setPartitionNum(changefeedID, 2) - require.True(t, errors.ErrKafkaInvalidPartitionNum.Equal(err)) + require.True(t, errors.ErrKafkaInvalidConfig.Equal(err)) } func TestClientID(t *testing.T) { diff --git a/pkg/sink/kafka/sarama_async_producer.go b/pkg/sink/kafka/sarama_async_producer.go index 9c0bd82104..8b142890b2 100644 --- a/pkg/sink/kafka/sarama_async_producer.go +++ b/pkg/sink/kafka/sarama_async_producer.go @@ -143,7 +143,7 @@ func (p *saramaAsyncProducer) handleProducerError(err *sarama.ProducerError) err extractLogInfo(err.Msg), err.Err, ) - return cerror.WrapError(cerror.ErrKafkaAsyncSendMessage, errWithInfo) + return cerror.WrapError(cerror.ErrKafkaSendMessage, errWithInfo) } // AsyncSend is the input channel for the user to write messages to that they @@ -152,7 +152,7 @@ func (p *saramaAsyncProducer) AsyncSend( ctx context.Context, topic string, partition int32, message *common.Message, ) error { if p.closed.Load() { - return cerror.ErrKafkaProducerClosed.GenWithStackByArgs() + return cerror.ErrKafkaSinkClosed.GenWithStackByArgs() } failpoint.Inject("KafkaSinkAsyncSendError", func() { // simulate sending message to input channel successfully but flushing diff --git a/pkg/sink/kafka/sarama_config.go b/pkg/sink/kafka/sarama_config.go index d18e8b0c56..d22404bf3a 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) } } @@ -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_factory.go b/pkg/sink/kafka/sarama_factory.go index 3be5fd841a..5790322e35 100644 --- a/pkg/sink/kafka/sarama_factory.go +++ b/pkg/sink/kafka/sarama_factory.go @@ -46,19 +46,19 @@ 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() }() if err = adjustOptions(changefeedID, admin, o, o.Topic); err != nil { - return nil, errors.Trace(err) + return nil, err } return &saramaFactory{ @@ -77,7 +77,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 +91,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 +103,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 +113,18 @@ 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) + return nil, errors.WrapError(errors.ErrNewKafkaSink, err) } return &saramaSyncProducer{ @@ -140,18 +140,18 @@ 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) + return nil, errors.WrapError(errors.ErrNewKafkaSink, err) } return &saramaAsyncProducer{ client: client, diff --git a/pkg/sink/kafka/sarama_sync_producer.go b/pkg/sink/kafka/sarama_sync_producer.go index 9d5efdfb0b..ffc8f68d3b 100644 --- a/pkg/sink/kafka/sarama_sync_producer.go +++ b/pkg/sink/kafka/sarama_sync_producer.go @@ -46,7 +46,7 @@ type saramaSyncProducer struct { func (p *saramaSyncProducer) SendMessage(topic string, partitionNum int32, message *common.Message) error { if p.closed.Load() { - return errors.ErrKafkaProducerClosed.GenWithStackByArgs() + return errors.ErrKafkaSinkClosed.GenWithStackByArgs() } msg := &sarama.ProducerMessage{ @@ -73,7 +73,7 @@ func (p *saramaSyncProducer) SendMessage(topic string, partitionNum int32, messa func (p *saramaSyncProducer) SendMessages(topic string, partitionNum int32, message *common.Message) error { if p.closed.Load() { - return errors.ErrKafkaProducerClosed.GenWithStackByArgs() + return errors.ErrKafkaSinkClosed.GenWithStackByArgs() } msgs := make([]*sarama.ProducerMessage, partitionNum) diff --git a/pkg/sink/kafka/sarama_sync_producer_test.go b/pkg/sink/kafka/sarama_sync_producer_test.go index 37285419c6..a0dca9f89c 100644 --- a/pkg/sink/kafka/sarama_sync_producer_test.go +++ b/pkg/sink/kafka/sarama_sync_producer_test.go @@ -14,14 +14,30 @@ package kafka import ( + "context" "errors" "testing" "github.com/golang/mock/gomock" - "github.com/pingcap/ticdc/pkg/common" + commonType "github.com/pingcap/ticdc/pkg/common" + cerror "github.com/pingcap/ticdc/pkg/errors" + "github.com/pingcap/ticdc/pkg/sink/codec/common" + "github.com/stretchr/testify/require" "go.uber.org/atomic" ) +func TestProducerRejectsSendAfterClose(t *testing.T) { + t.Parallel() + + message := &common.Message{} + syncProducer := &saramaSyncProducer{closed: atomic.NewBool(true)} + require.ErrorIs(t, syncProducer.SendMessage("topic", 1, message), cerror.ErrKafkaSinkClosed) + require.ErrorIs(t, syncProducer.SendMessages("topic", 1, message), cerror.ErrKafkaSinkClosed) + + asyncProducer := &saramaAsyncProducer{closed: atomic.NewBool(true)} + require.ErrorIs(t, asyncProducer.AsyncSend(context.Background(), "topic", 0, message), cerror.ErrKafkaSinkClosed) +} + func TestSyncProducerClose(t *testing.T) { tests := []struct { name string @@ -47,7 +63,7 @@ func TestSyncProducerClose(t *testing.T) { ) p := &saramaSyncProducer{ - id: common.NewChangeFeedIDWithName("test", "default"), + id: commonType.NewChangeFeedIDWithName("test", "default"), client: client, producer: producer, closed: atomic.NewBool(false), diff --git a/tests/integration_tests/http_api/util/test_case.py b/tests/integration_tests/http_api/util/test_case.py index 30c2a2e65e..8a3c8e4dbf 100644 --- a/tests/integration_tests/http_api/util/test_case.py +++ b/tests/integration_tests/http_api/util/test_case.py @@ -175,7 +175,7 @@ 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}" + assert "CDC:ErrNewKafkaSink" in resp.text, f"{resp.text}" assert "not found, ResolveEndpointV2" not in resp.text, f"{resp.text}" print("pass test: create changefeed") From a9c425a2a99ef8bedd725c35b546255c0e7ff0fa Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Tue, 28 Jul 2026 12:39:33 +0800 Subject: [PATCH 2/7] fix all error handling code --- downstreamadapter/sink/kafka/helper.go | 21 ++- downstreamadapter/sink/kafka/sink.go | 46 +++--- downstreamadapter/sink/kafka/sink_test.go | 50 ++++++- downstreamadapter/sink/pulsar/helper.go | 12 +- pkg/sink/kafka/admin_test.go | 8 +- pkg/sink/kafka/claimcheck/claim_check.go | 16 +-- pkg/sink/kafka/claimcheck/claim_check_test.go | 22 ++- pkg/sink/kafka/logutil.go | 24 +--- pkg/sink/kafka/logutil_test.go | 46 ++++-- pkg/sink/kafka/oauth2_token_provider.go | 4 +- pkg/sink/kafka/oauth2_token_provider_test.go | 6 +- pkg/sink/kafka/options_test.go | 46 +++--- pkg/sink/kafka/sarama_async_producer.go | 59 +++----- pkg/sink/kafka/sarama_factory.go | 1 - pkg/sink/kafka/sarama_sync_producer.go | 47 +++--- pkg/sink/kafka/sarama_sync_producer_test.go | 91 ++++++++++-- pkg/util/external_storage.go | 6 +- tests/integration_tests/kafka_log_info/run.sh | 134 ------------------ .../mq_sink_error_resume/run.sh | 21 +-- tests/integration_tests/run_heavy_it_in_ci.sh | 2 +- tests/integration_tests/run_light_it_in_ci.sh | 2 +- 21 files changed, 316 insertions(+), 348 deletions(-) delete mode 100755 tests/integration_tests/kafka_log_info/run.sh diff --git a/downstreamadapter/sink/kafka/helper.go b/downstreamadapter/sink/kafka/helper.go index f0cdf23852..37327548d4 100644 --- a/downstreamadapter/sink/kafka/helper.go +++ b/downstreamadapter/sink/kafka/helper.go @@ -21,11 +21,10 @@ 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/ticdc/pkg/sink/kafka/claimcheck" "github.com/pingcap/tidb/br/pkg/utils" @@ -33,7 +32,7 @@ import ( type components struct { encoderGroup codec.EncoderGroup - encoder common.EventEncoder + encoder codecCommon.EventEncoder columnSelector *columnselector.ColumnSelectors eventRouter *eventrouter.EventRouter topicManager topicmanager.TopicManager @@ -56,7 +55,7 @@ func (c components) close() { func newKafkaSinkComponent( ctx context.Context, - changefeedID commonType.ChangeFeedID, + changefeedID common.ChangeFeedID, sinkURI *url.URL, sinkConfig *config.SinkConfig, ) (components, config.Protocol, error) { @@ -95,12 +94,12 @@ func newKafkaSinkComponent( comp.eventRouter, err = eventrouter.NewEventRouter( sinkConfig, topic, false, isAvroLike) if err != nil { - return comp, protocol, errors.Trace(err) + return comp, protocol, err } comp.columnSelector, err = columnselector.New(sinkConfig) if err != nil { - return comp, protocol, errors.Trace(err) + return comp, protocol, err } encoderConfig, err := helper.GetEncoderConfig( @@ -108,22 +107,22 @@ func newKafkaSinkComponent( options.MaxMessageBytes, options.MaxBatchedBytes, ) if err != nil { - return comp, protocol, errors.Trace(err) + return comp, protocol, err } comp.claimCheck, err = claimcheck.New(ctx, encoderConfig.LargeMessageHandle, changefeedID) if err != nil { - return comp, protocol, errors.Trace(err) + return comp, protocol, err } comp.encoderGroup, err = codec.NewEncoderGroup(ctx, sinkConfig, encoderConfig, comp.claimCheck, changefeedID) if err != nil { - return comp, protocol, errors.Trace(err) + return comp, protocol, err } comp.encoder, err = codec.NewEventEncoder(ctx, encoderConfig, comp.claimCheck) if err != nil { - return comp, protocol, errors.Trace(err) + return comp, protocol, err } comp.adminClient, err = comp.factory.AdminClient(ctx) diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index be75f3ffcc..13c4d539b6 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -22,13 +22,13 @@ import ( "github.com/pingcap/ticdc/downstreamadapter/sink/columnselector" "github.com/pingcap/ticdc/downstreamadapter/sink/eventrouter" "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" "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/ticdc/pkg/sink/kafka/claimcheck" "github.com/pingcap/ticdc/pkg/util" @@ -44,7 +44,7 @@ const ( ) type sink struct { - changefeedID commonType.ChangeFeedID + changefeedID common.ChangeFeedID dmlProducer kafka.AsyncProducer ddlProducer kafka.SyncProducer @@ -67,19 +67,19 @@ type sink struct { ctx context.Context } -func (s *sink) SinkType() commonType.SinkType { - return commonType.KafkaSinkType +func (s *sink) SinkType() common.SinkType { + return common.KafkaSinkType } -func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url.URL, sinkConfig *config.SinkConfig) error { +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 errors.Trace(err) + return err } topic, err := helper.GetTopic(uri) if err != nil { - return errors.Trace(err) + return err } options := kafka.NewOptions() @@ -93,7 +93,7 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url. options.MaxMessageBytes, options.MaxBatchedBytes, ) if err != nil { - return errors.Trace(err) + return err } claimCheck, err := claimcheck.New(ctx, encoderConfig.LargeMessageHandle, changefeedID) @@ -104,11 +104,11 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url. isAvroLike := protocol == config.ProtocolAvro || protocol == config.ProtocolDebeziumAvro if _, err = eventrouter.NewEventRouter(sinkConfig, topic, false, isAvroLike); err != nil { - return errors.Trace(err) + return err } if _, err = columnselector.New(sinkConfig); err != nil { - return errors.Trace(err) + return err } factory, err := kafka.NewSaramaFactory(ctx, options, changefeedID) @@ -148,13 +148,13 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url. _, err = codec.NewEventEncoder(ctx, encoderConfig, claimCheck) if err != nil { - return errors.Trace(err) + return err } return nil } func New( - ctx context.Context, changefeedID commonType.ChangeFeedID, sinkURI *url.URL, sinkConfig *config.SinkConfig, keyspaceID uint32, + ctx context.Context, changefeedID common.ChangeFeedID, sinkURI *url.URL, sinkConfig *config.SinkConfig, keyspaceID uint32, ) (*sink, error) { comp, protocol, err := newKafkaSinkComponent(ctx, changefeedID, sinkURI, sinkConfig) if err != nil { @@ -165,7 +165,7 @@ func New( func newWithComponents( ctx context.Context, - changefeedID commonType.ChangeFeedID, + changefeedID common.ChangeFeedID, keyspaceID uint32, protocol config.Protocol, comp components, @@ -236,7 +236,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 { @@ -307,7 +307,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 { @@ -328,7 +328,7 @@ func (s *sink) calculateKeyPartitions(ctx context.Context) error { selector := s.comp.columnSelector.GetForTableInfo(event.TableInfo) events, err := helper.NewMQRowEvents(event, topic, partitionNum, partitionGenerator, selector) if err != nil { - return errors.Trace(err) + return err } s.rowChan.Push(events...) } @@ -339,7 +339,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 { @@ -432,7 +432,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", @@ -481,7 +481,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. @@ -531,14 +531,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", @@ -555,7 +555,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 892b3e7c92..0ffbd3b3b3 100644 --- a/downstreamadapter/sink/kafka/sink_test.go +++ b/downstreamadapter/sink/kafka/sink_test.go @@ -32,7 +32,7 @@ import ( commonEvent "github.com/pingcap/ticdc/pkg/common/event" "github.com/pingcap/ticdc/pkg/config" "github.com/pingcap/ticdc/pkg/sink/codec" - codeccommon "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/stretchr/testify/require" "go.uber.org/atomic" @@ -40,6 +40,52 @@ import ( 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() @@ -249,7 +295,7 @@ func TestKafkaSinkBasicFunctionality(t *testing.T) { _ context.Context, _ string, _ int32, - message *codeccommon.Message, + message *codecCommon.Message, ) error { if message.Callback != nil { message.Callback() diff --git a/downstreamadapter/sink/pulsar/helper.go b/downstreamadapter/sink/pulsar/helper.go index 59e592d0de..b2676b1daa 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, diff --git a/pkg/sink/kafka/admin_test.go b/pkg/sink/kafka/admin_test.go index 48d0894f24..3bcd3d0468 100644 --- a/pkg/sink/kafka/admin_test.go +++ b/pkg/sink/kafka/admin_test.go @@ -14,13 +14,13 @@ package kafka import ( - stderrors "errors" + "io" "testing" "github.com/IBM/sarama" "github.com/golang/mock/gomock" "github.com/pingcap/ticdc/pkg/common" - cerror "github.com/pingcap/ticdc/pkg/errors" + "github.com/pingcap/ticdc/pkg/errors" "github.com/stretchr/testify/require" ) @@ -47,7 +47,7 @@ func TestGetBrokerConfig(t *testing.T) { t.Run("admin error", func(t *testing.T) { ctrl := gomock.NewController(t) admin := NewMocksaramaClusterAdmin(ctrl) - cause := stderrors.New("describe cluster failed") + cause := io.ErrUnexpectedEOF admin.EXPECT().DescribeCluster().Return(nil, int32(0), cause) client := &saramaAdminClient{ @@ -56,7 +56,7 @@ func TestGetBrokerConfig(t *testing.T) { } _, _, err := client.GetBrokerConfig("missing") - require.ErrorIs(t, err, cerror.ErrKafkaAdminAPI) + require.ErrorIs(t, err, errors.ErrKafkaAdminAPI) require.ErrorIs(t, err, cause) }) } diff --git a/pkg/sink/kafka/claimcheck/claim_check.go b/pkg/sink/kafka/claimcheck/claim_check.go index e38b01c97f..5ed01ec016 100644 --- a/pkg/sink/kafka/claimcheck/claim_check.go +++ b/pkg/sink/kafka/claimcheck/claim_check.go @@ -21,10 +21,10 @@ import ( "github.com/google/uuid" "github.com/pingcap/log" - 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/common" + codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/pingcap/ticdc/pkg/util" "github.com/pingcap/tidb/pkg/objstore/storeapi" "github.com/prometheus/client_golang/prometheus" @@ -36,7 +36,7 @@ type ClaimCheck struct { storage storeapi.Storage 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 @@ -44,7 +44,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 } @@ -58,7 +58,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 } return &ClaimCheck{ @@ -73,19 +73,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 index e51cfc5626..080e62de0f 100644 --- a/pkg/sink/kafka/claimcheck/claim_check_test.go +++ b/pkg/sink/kafka/claimcheck/claim_check_test.go @@ -16,10 +16,12 @@ package claimcheck import ( "context" "fmt" + "strings" "testing" - 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/tidb/pkg/objstore" "github.com/pingcap/tidb/pkg/objstore/mockobjstore" "github.com/stretchr/testify/require" @@ -32,7 +34,7 @@ func TestClaimCheck(t *testing.T) { ctx := context.Background() - changefeedID := commonType.NewChangeFeedIDWithName("test", "") + changefeedID := common.NewChangeFeedIDWithName("test", "") largeHandleConfig := config.NewDefaultLargeMessageHandleConfig() claimCheck, err := New(ctx, largeHandleConfig, changefeedID) @@ -49,6 +51,18 @@ func TestClaimCheck(t *testing.T) { 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) @@ -58,7 +72,7 @@ func TestClaimCheckCloseClosesStorage(t *testing.T) { storage.EXPECT().Close().Times(1) claimCheck := &ClaimCheck{ storage: storage, - changefeedID: commonType.NewChangeFeedIDWithName("test", "default"), + changefeedID: common.NewChangeFeedIDWithName("test", "default"), } claimCheck.Close() @@ -67,7 +81,7 @@ func TestClaimCheckCloseClosesStorage(t *testing.T) { func TestClaimCheckConcurrentWrites(t *testing.T) { ctx := context.Background() storage := objstore.NewMemStorage() - changefeedID := commonType.NewChangeFeedIDWithName("test", "default") + changefeedID := common.NewChangeFeedIDWithName("test", "default") claimCheck := &ClaimCheck{ storage: storage, rawValue: true, 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_test.go b/pkg/sink/kafka/options_test.go index ddc8c971bc..a77b10b319 100644 --- a/pkg/sink/kafka/options_test.go +++ b/pkg/sink/kafka/options_test.go @@ -25,7 +25,7 @@ import ( "github.com/IBM/sarama" "github.com/aws/aws-sdk-go-v2/aws" "github.com/golang/mock/gomock" - 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/stretchr/testify/require" @@ -175,7 +175,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) @@ -190,7 +190,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) @@ -200,7 +200,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)) for _, replicationFactor := range []string{"0", "-1"} { uri = "kafka://127.0.0.1:9092/abc?replication-factor=" + replicationFactor @@ -208,7 +208,7 @@ func TestCompleteOptions(t *testing.T) { require.NoError(t, err) options = NewOptions() err = options.Apply( - commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), + common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink, ) @@ -220,7 +220,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 max-retry. @@ -228,7 +228,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. @@ -236,7 +236,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. @@ -244,7 +244,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. @@ -252,7 +252,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 @@ -260,7 +260,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.True(t, errors.ErrKafkaInvalidConfig.Equal(err)) // max-retry accepts non-negative sink-uri values. @@ -268,7 +268,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, 7, options.MaxRetry) @@ -276,7 +276,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) @@ -285,7 +285,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, defaultMaxRetry, options.MaxRetry) } @@ -321,7 +321,7 @@ func TestApplyRejectsNonPositiveMaxMessageBytes(t *testing.T) { }, } - changefeedID := commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test") + changefeedID := common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test") for _, test := range tests { t.Run(test.name, func(t *testing.T) { sinkURI, err := url.Parse(test.uri) @@ -347,7 +347,7 @@ func TestApplyRejectsNonPositiveMaxMessageBytes(t *testing.T) { func TestSetPartitionNum(t *testing.T) { options := NewOptions() - changefeedID := commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test") + changefeedID := common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test") err := options.setPartitionNum(changefeedID, 2) require.NoError(t, err) require.Equal(t, int32(2), options.PartitionNum) @@ -397,7 +397,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 { @@ -418,7 +418,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) @@ -452,7 +452,7 @@ func TestAdjustConfigFallsBackToBrokerMessageMaxBytesWhenTopicConfigMissing(t *t } topicName := "test-topic" - changefeedID := commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test") + changefeedID := common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test") for _, test := range tests { t.Run(test.name, func(t *testing.T) { adminFixture := newKafkaAdminFixture(t) @@ -701,7 +701,7 @@ func TestConfigurationCombinations(t *testing.T) { require.Nil(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.Nil(t, err) configuredMaxMessageBytes := options.MaxMessageBytes @@ -714,7 +714,7 @@ func TestConfigurationCombinations(t *testing.T) { sourceMaxMessageBytes = adminFixture.topicMaxMessageBytes(topic) } - changefeedID := commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test") + changefeedID := common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test") err = adjustOptions(changefeedID, adminClient, options, topic) require.Nil(t, err) require.Equal(t, sourceMaxMessageBytes, options.MaxMessageBytes) @@ -761,7 +761,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) @@ -843,7 +843,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 8b142890b2..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.ErrKafkaSendMessage, 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.ErrKafkaSinkClosed.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_factory.go b/pkg/sink/kafka/sarama_factory.go index 5790322e35..59c052b547 100644 --- a/pkg/sink/kafka/sarama_factory.go +++ b/pkg/sink/kafka/sarama_factory.go @@ -158,7 +158,6 @@ func (f *saramaFactory) AsyncProducer(ctx context.Context) (AsyncProducer, error 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 ffc8f68d3b..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,13 +37,13 @@ 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.ErrKafkaSinkClosed.GenWithStackByArgs() } @@ -56,22 +55,18 @@ 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.ErrKafkaSinkClosed.GenWithStackByArgs() } @@ -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 a0dca9f89c..522dc13bcb 100644 --- a/pkg/sink/kafka/sarama_sync_producer_test.go +++ b/pkg/sink/kafka/sarama_sync_producer_test.go @@ -15,13 +15,15 @@ package kafka import ( "context" - "errors" + "io" + "strings" "testing" + "github.com/IBM/sarama" "github.com/golang/mock/gomock" - 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" "github.com/stretchr/testify/require" "go.uber.org/atomic" ) @@ -29,13 +31,13 @@ import ( func TestProducerRejectsSendAfterClose(t *testing.T) { t.Parallel() - message := &common.Message{} + message := &codecCommon.Message{} syncProducer := &saramaSyncProducer{closed: atomic.NewBool(true)} - require.ErrorIs(t, syncProducer.SendMessage("topic", 1, message), cerror.ErrKafkaSinkClosed) - require.ErrorIs(t, syncProducer.SendMessages("topic", 1, message), cerror.ErrKafkaSinkClosed) + 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), cerror.ErrKafkaSinkClosed) + require.ErrorIs(t, asyncProducer.AsyncSend(context.Background(), "topic", 0, message), errors.ErrKafkaSinkClosed) } func TestSyncProducerClose(t *testing.T) { @@ -48,7 +50,7 @@ func TestSyncProducerClose(t *testing.T) { }, { name: "still closes producer when client close fails", - clientCloseErr: errors.New("boom"), + clientCloseErr: io.ErrClosedPipe, }, } @@ -63,7 +65,7 @@ func TestSyncProducerClose(t *testing.T) { ) p := &saramaSyncProducer{ - id: commonType.NewChangeFeedIDWithName("test", "default"), + id: common.NewChangeFeedIDWithName("test", "default"), client: client, producer: producer, closed: atomic.NewBool(false), @@ -73,3 +75,72 @@ func TestSyncProducerClose(t *testing.T) { }) } } + +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 5817c25480..02c0bcf154 100644 --- a/pkg/util/external_storage.go +++ b/pkg/util/external_storage.go @@ -66,7 +66,7 @@ func getExternalStorage( ) (storeapi.Storage, error) { backEnd, err := objstore.ParseBackend(uri, opts) if err != nil { - return nil, errors.Trace(err) + return nil, errors.WrapError(errors.ErrExternalStorageAPI, err) } ret, err := objstore.New(ctx, backEnd, &storeapi.Options{ @@ -74,7 +74,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 { @@ -85,7 +85,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/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 34c6d4af32..4eea519108 100755 --- a/tests/integration_tests/run_heavy_it_in_ci.sh +++ b/tests/integration_tests/run_heavy_it_in_ci.sh @@ -85,7 +85,7 @@ kafka_groups=( # G08 'kafka_simple_claim_check kafka_simple_claim_check_avro tidb_mysql_test' # 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 '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 1bfb0b83b6..995583e1f2 100755 --- a/tests/integration_tests/run_light_it_in_ci.sh +++ b/tests/integration_tests/run_light_it_in_ci.sh @@ -101,7 +101,7 @@ kafka_groups=( # G14 'kafka_simple_basic avro_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: From 6503a19ae2129a46e4a24e1d169f54ebe2fe9c86 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Tue, 28 Jul 2026 13:57:22 +0800 Subject: [PATCH 3/7] add test --- downstreamadapter/sink/kafka/sink_test.go | 24 ++++++++++++++++++- .../kafka_big_messages/run.sh | 19 +++++++++++---- 2 files changed, 38 insertions(+), 5 deletions(-) diff --git a/downstreamadapter/sink/kafka/sink_test.go b/downstreamadapter/sink/kafka/sink_test.go index 0ffbd3b3b3..8b205c1649 100644 --- a/downstreamadapter/sink/kafka/sink_test.go +++ b/downstreamadapter/sink/kafka/sink_test.go @@ -31,6 +31,7 @@ import ( "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/sink/codec" codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/pingcap/ticdc/pkg/sink/kafka" @@ -231,10 +232,30 @@ func newKafkaSinkForTestWithProducers(ctx context.Context, if err != nil { return nil, err } - go s.Run(ctx) return s, nil } +func TestKafkaSinkRunReturnsAsyncProducerError(t *testing.T) { + ctx := t.Context() + + ctrl := gomock.NewController(t) + producerErr := errors.ErrKafkaSendMessage.GenWithStackByArgs() + asyncProducer := kafka.NewMockAsyncProducer(ctrl) + syncProducer := kafka.NewMockSyncProducer(ctrl) + 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()) +} + func TestKafkaSinkBasicFunctionality(t *testing.T) { helper := commonEvent.NewEventTestHelper(t) defer helper.Close() @@ -309,6 +330,7 @@ func TestKafkaSinkBasicFunctionality(t *testing.T) { kafkaSink, err := newKafkaSinkForTestWithProducers(ctx, t, ctrl, asyncProducer, syncProducer) require.NoError(t, err) defer cancel() + go kafkaSink.Run(ctx) err = kafkaSink.WriteBlockEvent(ddlEvent) require.NoError(t, err) diff --git a/tests/integration_tests/kafka_big_messages/run.sh b/tests/integration_tests/kafka_big_messages/run.sh index 7225decaaa..8f126a914b 100755 --- a/tests/integration_tests/kafka_big_messages/run.sh +++ b/tests/integration_tests/kafka_big_messages/run.sh @@ -129,10 +129,16 @@ function run_protocol_case() { 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 "$SMALL_TOPIC_LIMIT" + 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") @@ -145,13 +151,17 @@ function run_protocol_case() { 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 + 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" - # The encoded row is larger than the topic limit, so the changefeed must - # enter the retryable warning state with ErrMessageTooLarge. - wait_changefeed_state "$pd_addr" "$changefeed_id" "warning" "ErrMessageTooLarge" + 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. @@ -173,6 +183,7 @@ function run() { 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" From ebc524837030e83a646f68d45300783f66cc5f86 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Tue, 28 Jul 2026 13:59:44 +0800 Subject: [PATCH 4/7] close the client --- pkg/sink/kafka/sarama_factory.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/pkg/sink/kafka/sarama_factory.go b/pkg/sink/kafka/sarama_factory.go index 59c052b547..c9678ce784 100644 --- a/pkg/sink/kafka/sarama_factory.go +++ b/pkg/sink/kafka/sarama_factory.go @@ -124,6 +124,7 @@ func (f *saramaFactory) SyncProducer(ctx context.Context) (SyncProducer, error) p, err := sarama.NewSyncProducerFromClient(client) if err != nil { + _ = client.Close() return nil, errors.WrapError(errors.ErrNewKafkaSink, err) } @@ -151,6 +152,7 @@ func (f *saramaFactory) AsyncProducer(ctx context.Context) (AsyncProducer, error p, err := sarama.NewAsyncProducerFromClient(client) if err != nil { + _ = client.Close() return nil, errors.WrapError(errors.ErrNewKafkaSink, err) } return &saramaAsyncProducer{ From effa50d77e18c8fa6754f81741d0ebb960f2fc34 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Tue, 28 Jul 2026 14:39:57 +0800 Subject: [PATCH 5/7] fix http api test --- tests/integration_tests/http_api/util/test_case.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/tests/integration_tests/http_api/util/test_case.py b/tests/integration_tests/http_api/util/test_case.py index 8a3c8e4dbf..c959682b7e 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:ErrNewKafkaSink" 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") From 4438456bf0e2c78263312aaae83aeb3979d3c69c Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Tue, 28 Jul 2026 15:30:29 +0800 Subject: [PATCH 6/7] adjust the code --- tests/integration_tests/kafka_big_messages/run.sh | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tests/integration_tests/kafka_big_messages/run.sh b/tests/integration_tests/kafka_big_messages/run.sh index 8f126a914b..b9a4b646fb 100755 --- a/tests/integration_tests/kafka_big_messages/run.sh +++ b/tests/integration_tests/kafka_big_messages/run.sh @@ -154,6 +154,9 @@ function run_protocol_case() { # 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 From c9e57f1bb89e09ef0417424065f0fc7ca8bbad58 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Tue, 28 Jul 2026 15:34:36 +0800 Subject: [PATCH 7/7] adjust the code --- pkg/sink/kafka/sarama_config.go | 2 +- pkg/sink/kafka/sarama_config_test.go | 21 ++++++++++++++++++--- 2 files changed, 19 insertions(+), 4 deletions(-) diff --git a/pkg/sink/kafka/sarama_config.go b/pkg/sink/kafka/sarama_config.go index d22404bf3a..6f2d56456e 100644 --- a/pkg/sink/kafka/sarama_config.go +++ b/pkg/sink/kafka/sarama_config.go @@ -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) diff --git a/pkg/sink/kafka/sarama_config_test.go b/pkg/sink/kafka/sarama_config_test.go index 89f3e004c6..3e1520b1e8 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" ) @@ -86,6 +86,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() @@ -127,7 +142,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, )