diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index e6d1038b26..90653aec87 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -68,9 +68,91 @@ func (s *sink) SinkType() commonType.SinkType { } func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url.URL, sinkConfig *config.SinkConfig) error { +<<<<<<< HEAD comp, _, err := newKafkaSinkComponent(ctx, changefeedID, uri, sinkConfig) defer comp.close() return err +======= + protocol, err := helper.GetProtocol(util.GetOrZero(sinkConfig.Protocol)) + if err != nil { + return errors.Trace(err) + } + + topic, err := helper.GetTopic(uri) + if err != nil { + return errors.Trace(err) + } + + options := kafka.NewOptions() + if err = options.Apply(changefeedID, uri, sinkConfig); err != nil { + return errors.WrapError(errors.ErrKafkaInvalidConfig, err) + } + options.Topic = topic + + encoderConfig, err := helper.GetEncoderConfig( + changefeedID, uri, protocol, sinkConfig, + options.MaxMessageBytes, options.MaxBatchedBytes, + ) + if err != nil { + return errors.Trace(err) + } + + claimCheck, err := claimcheck.New(ctx, encoderConfig.LargeMessageHandle, changefeedID) + if err != nil { + return err + } + defer claimCheck.Close() + + isAvroLike := protocol == config.ProtocolAvro || protocol == config.ProtocolDebeziumAvro + if _, err = eventrouter.NewEventRouter(sinkConfig, topic, false, isAvroLike); err != nil { + return errors.Trace(err) + } + + if _, err = columnselector.New(sinkConfig); err != nil { + return errors.Trace(err) + } + + factory, err := kafka.NewSaramaFactory(ctx, options, changefeedID) + if err != nil { + return errors.WrapError(errors.ErrKafkaNewProducer, err) + } + + adminClient, err := factory.AdminClient(ctx) + if err != nil { + return errors.WrapError(errors.ErrKafkaNewProducer, err) + } + defer adminClient.Close() + + topics, err := adminClient.GetTopicsMeta([]string{topic}, false) + if err != nil { + return errors.Trace(err) + } + if _, exists := topics[topic]; !exists { + topicConfig := options.DeriveTopicConfig() + if !topicConfig.AutoCreate { + return errors.ErrKafkaInvalidConfig.GenWithStack("`auto-create-topic` is false, and %s not found", topic) + } + if err = topicConfig.ValidateReplicationFactor(adminClient); err != nil { + return err + } + + // the topic is not created, only validate. + err = adminClient.CreateTopic(&kafka.TopicDetail{ + Name: topic, + NumPartitions: topicConfig.PartitionNum, + ReplicationFactor: topicConfig.ReplicationFactor, + }, true) + if err != nil { + return errors.WrapError(errors.ErrKafkaCreateTopic, err) + } + } + + _, err = codec.NewEventEncoder(ctx, encoderConfig, claimCheck) + if err != nil { + return errors.Trace(err) + } + return nil +>>>>>>> 0d4929739 (kafka: verify replication-factor when need to create the topic (#5715)) } func New( diff --git a/downstreamadapter/sink/kafka/sink_test.go b/downstreamadapter/sink/kafka/sink_test.go index 0bb4708f58..8888e1e54b 100644 --- a/downstreamadapter/sink/kafka/sink_test.go +++ b/downstreamadapter/sink/kafka/sink_test.go @@ -16,11 +16,20 @@ package kafka import ( "context" "fmt" + "net/http" + "net/http/httptest" "net/url" "testing" "time" +<<<<<<< HEAD "github.com/pingcap/errors" +======= + "github.com/IBM/sarama" + "github.com/golang/mock/gomock" + "github.com/pingcap/ticdc/downstreamadapter/sink/columnselector" + "github.com/pingcap/ticdc/downstreamadapter/sink/eventrouter" +>>>>>>> 0d4929739 (kafka: verify replication-factor when need to create the topic (#5715)) "github.com/pingcap/ticdc/downstreamadapter/sink/helper" "github.com/pingcap/ticdc/pkg/common" commonEvent "github.com/pingcap/ticdc/pkg/common/event" @@ -32,6 +41,48 @@ import ( "go.uber.org/atomic" ) +<<<<<<< HEAD +======= +const kafkaSinkTestTopic = "mock_topic" + +func TestVerifyInvalidConfig(t *testing.T) { + broker := sarama.NewMockBroker(t, 1) + defer broker.Close() + broker.SetHandlerByMap(map[string]sarama.MockResponse{ + "ApiVersionsRequest": sarama.NewMockApiVersionsResponse(t).SetApiKeys( + []sarama.ApiVersionsResponseKey{ + {ApiKey: 0}, + {ApiKey: 1}, + {ApiKey: 2}, + {ApiKey: 3, MaxVersion: 9}, + }), + "MetadataRequest": sarama.NewMockMetadataResponse(t). + SetController(broker.BrokerID()). + SetBroker(broker.Addr(), broker.BrokerID()). + SetLeader(kafkaSinkTestTopic, 0, broker.BrokerID()), + "DescribeConfigsRequest": sarama.NewMockDescribeConfigsResponse(t), + }) + + schemaRegistry := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + http.Error(w, "invalid response", http.StatusInternalServerError) + })) + defer schemaRegistry.Close() + + avroProtocol := config.ProtocolAvro.String() + sinkConfig := &config.SinkConfig{ + Protocol: &avroProtocol, + SchemaRegistry: &schemaRegistry.URL, + } + sinkURI, err := url.Parse("kafka://" + broker.Addr() + "/" + kafkaSinkTestTopic + + "?required-acks=1&kafka-version=2.4.0") + require.NoError(t, err) + + changefeedID := common.NewChangefeedID4Test("test", "verify-invalid-config") + err = Verify(context.Background(), changefeedID, sinkURI, sinkConfig) + require.ErrorContains(t, err, "ErrAvroSchemaAPIError") +} + +>>>>>>> 0d4929739 (kafka: verify replication-factor when need to create the topic (#5715)) func newKafkaSinkForTestWithProducers(ctx context.Context, asyncProducer kafka.AsyncProducer, syncProducer kafka.SyncProducer, diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go index 8e92167327..168a532be6 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go @@ -15,7 +15,6 @@ package topicmanager import ( "context" - "fmt" "sync" "time" @@ -242,9 +241,17 @@ func (m *kafkaTopicManager) createTopic( topicName string, ) (int32, error) { if !m.cfg.AutoCreate { +<<<<<<< HEAD return 0, cerror.ErrKafkaInvalidConfig.GenWithStack( fmt.Sprintf("`auto-create-topic` is false, "+ "and %s not found", topicName)) +======= + return 0, errors.ErrKafkaInvalidConfig.GenWithStack("`auto-create-topic` is false, and %s not found", topicName) + } + + if err := m.cfg.ValidateReplicationFactor(m.admin); err != nil { + return 0, err +>>>>>>> 0d4929739 (kafka: verify replication-factor when need to create the topic (#5715)) } start := time.Now() diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go index bf02658b24..b77777cfe9 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go @@ -31,6 +31,7 @@ func TestCreateTopic(t *testing.T) { AutoCreate: true, PartitionNum: 2, ReplicationFactor: 1, + RequiredAcks: kafka.WaitForAll, } changefeedID := common.NewChangefeedID4Test("test", "test") @@ -41,6 +42,7 @@ func TestCreateTopic(t *testing.T) { require.NoError(t, err) require.Equal(t, int32(2), partitionNum) + cfg.RequiredAcks = kafka.WaitForLocal partitionNum, err = manager.CreateTopicAndWaitUntilVisible(ctx, "new-topic") require.NoError(t, err) require.Equal(t, int32(2), partitionNum) @@ -49,7 +51,12 @@ func TestCreateTopic(t *testing.T) { require.Equal(t, int32(2), partitionsNum) // Try to create a topic without auto create. - cfg.AutoCreate = false + cfg = &kafka.AutoCreateTopicConfig{ + AutoCreate: false, + PartitionNum: 2, + ReplicationFactor: 1, + RequiredAcks: kafka.WaitForAll, + } manager = newKafkaTopicManager(ctx, "new-topic2", changefeedID, adminClient, cfg) defer manager.Close() _, err = manager.CreateTopicAndWaitUntilVisible(ctx, "new-topic2") @@ -77,7 +84,44 @@ func TestCreateTopic(t *testing.T) { ) } +<<<<<<< HEAD func TestCreateTopicWithDelay(t *testing.T) { +======= +func TestCreateTopicValidatesReplicationFactor(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + adminClient := kafka.NewMockClusterAdminClient(ctrl) + topic := "new-topic" + gomock.InOrder( + adminClient.EXPECT().GetTopicsMeta([]string{topic}, true). + Return(map[string]kafka.TopicDetail{}, nil), + adminClient.EXPECT().GetTopicsMeta([]string{topic}, false). + Return(map[string]kafka.TopicDetail{}, nil), + adminClient.EXPECT().GetBrokerConfig(kafka.MinInsyncReplicasConfigName). + Return("2", nil), + ) + + manager := newKafkaTopicManager( + context.Background(), + topic, + common.NewChangefeedID4Test("test", "test"), + adminClient, + &kafka.AutoCreateTopicConfig{ + AutoCreate: true, + PartitionNum: 2, + ReplicationFactor: 1, + RequiredAcks: kafka.WaitForAll, + }, + ) + defer manager.Close() + + _, err := manager.CreateTopicAndWaitUntilVisible(context.Background(), topic) + require.ErrorContains(t, err, "`replication-factor` 1 is smaller than the `min.insync.replicas` 2 of broker") +} + +func TestCreateTopicWaitsUntilVisible(t *testing.T) { +>>>>>>> 0d4929739 (kafka: verify replication-factor when need to create the topic (#5715)) t.Parallel() adminClient := kafka.NewClusterAdminClientMockImpl() diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go index c9b992814e..48ed5c5677 100644 --- a/pkg/sink/kafka/options.go +++ b/pkg/sink/kafka/options.go @@ -14,7 +14,6 @@ package kafka import ( - "context" "encoding/base64" "fmt" "net/http" @@ -249,6 +248,9 @@ func (o *options) Apply(changefeedID common.ChangeFeedID, } if urlParameter.ReplicationFactor != nil { + if *urlParameter.ReplicationFactor <= 0 { + return errors.ErrKafkaInvalidConfig.GenWithStack("invalid replication-factor %d", *urlParameter.ReplicationFactor) + } o.ReplicationFactor = *urlParameter.ReplicationFactor } @@ -537,11 +539,12 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf return nil } -// AutoCreateTopicConfig is used to create topic configuration. +// AutoCreateTopicConfig contains settings used to create and validate a topic. type AutoCreateTopicConfig struct { AutoCreate bool PartitionNum int32 ReplicationFactor int16 + RequiredAcks RequiredAcks } func (o *options) DeriveTopicConfig() *AutoCreateTopicConfig { @@ -549,7 +552,40 @@ func (o *options) DeriveTopicConfig() *AutoCreateTopicConfig { AutoCreate: o.AutoCreate, PartitionNum: o.PartitionNum, ReplicationFactor: o.ReplicationFactor, + RequiredAcks: o.RequiredAcks, + } +} + +// ValidateReplicationFactor checks whether a topic created with this config +// can satisfy the configured acknowledgment requirement. +func (c *AutoCreateTopicConfig) ValidateReplicationFactor(admin ClusterAdminClient) error { + if c.RequiredAcks != WaitForAll { + return nil + } + + raw, err := admin.GetBrokerConfig(MinInsyncReplicasConfigName) + if err != nil { + log.Warn("cannot get Kafka broker configuration, assume replication factor is valid", + zap.String("configName", MinInsyncReplicasConfigName), + zap.Int16("replicationFactor", c.ReplicationFactor), + zap.Error(err)) + return nil + } + minInsyncReplicas, err := strconv.Atoi(raw) + if err != nil { + return err } + + if int(c.ReplicationFactor) < minInsyncReplicas { + return errors.ErrKafkaInvalidConfig.GenWithStack( + "TiCDC Kafka sink's `request.required.acks` defaults to -1, "+ + "TiCDC cannot deliver messages when the `replication-factor` %d "+ + "is smaller than the `min.insync.replicas` %d of broker", + c.ReplicationFactor, minInsyncReplicas, + ) + } + + return nil } var ( @@ -577,7 +613,11 @@ func NewKafkaClientID(captureAddr string, // adjustOptions adjust the `options` and `sarama.Config` by condition. func adjustOptions( +<<<<<<< HEAD ctx context.Context, +======= + changefeedID common.ChangeFeedID, +>>>>>>> 0d4929739 (kafka: verify replication-factor when need to create the topic (#5715)) admin ClusterAdminClient, options *options, topic string, @@ -587,6 +627,7 @@ func adjustOptions( return errors.Trace(err) } +<<<<<<< HEAD // Only check replicationFactor >= minInsyncReplicas when producer's required acks is -1. // If we don't check it, the producer probably can not send message to the topic. // Because it will wait for the ack from all replicas. But we do not have enough replicas. @@ -597,10 +638,13 @@ func adjustOptions( } } +======= +>>>>>>> 0d4929739 (kafka: verify replication-factor when need to create the topic (#5715)) info, exists := topics[topic] // once we have found the topic, no matter `auto-create-topic`, // make sure user input parameters are valid. if exists { +<<<<<<< HEAD // make sure that producer's `MaxMessageBytes` smaller than topic's `max.message.bytes` topicMaxMessageBytesStr, err := getTopicConfig( ctx, admin, info.Name, @@ -649,6 +693,28 @@ func adjustOptions( return errors.Trace(err) } brokerMessageMaxBytes, err := strconv.Atoi(brokerMessageMaxBytesStr) +======= + err = adjustExistingTopicOption(changefeedID, admin, options, topic, info) + } else { + adjustNewTopicOptions(admin, changefeedID, options, topic) + } + if err != nil { + return err + } + + options.MaxBatchedBytes = min(options.MaxBatchedBytes, options.MaxMessageBytes) + return nil +} + +func adjustExistingTopicOption( + changefeedID common.ChangeFeedID, + admin ClusterAdminClient, + options *options, + topic string, + info TopicDetail, +) error { + maxMessageBytes, err := getTopicMaxMessageBytes(admin, info.Name) +>>>>>>> 0d4929739 (kafka: verify replication-factor when need to create the topic (#5715)) if err != nil { return errors.Trace(err) } @@ -677,6 +743,7 @@ func adjustOptions( log.Warn("partition-num is not set, use the default partition count", zap.String("topic", topic), zap.Int32("partitions", options.PartitionNum)) } +<<<<<<< HEAD return nil } @@ -747,12 +814,46 @@ func validateMinInsyncReplicas( return nil } +======= +} + +func getTopicMaxMessageBytes( + admin ClusterAdminClient, + topic string, +) (int, error) { + raw, err := getTopicConfig( + admin, topic, + TopicMaxMessageBytesConfigName, + BrokerMessageMaxBytesConfigName, + ) + if err != nil { + return 0, errors.Trace(err) + } + maxMessageBytes, err := strconv.Atoi(raw) + if err != nil { + return 0, errors.Trace(err) + } + return maxMessageBytes, nil +} + +func getBrokerMaxMessageBytes(admin ClusterAdminClient) (int, error) { + raw, err := admin.GetBrokerConfig(BrokerMessageMaxBytesConfigName) + if err != nil { + return 0, errors.Trace(err) + } + messageMaxBytes, err := strconv.Atoi(raw) + if err != nil { + return 0, errors.Trace(err) + } + return messageMaxBytes, nil +} + +>>>>>>> 0d4929739 (kafka: verify replication-factor when need to create the topic (#5715)) // getTopicConfig gets topic config by name. // If the topic does not have this configuration, // we will try to get it from the broker's configuration. // NOTICE: The configuration names of topic and broker may be different for the same configuration. func getTopicConfig( - ctx context.Context, admin ClusterAdminClient, topicName string, topicConfigName string, diff --git a/pkg/sink/kafka/options_test.go b/pkg/sink/kafka/options_test.go index 8f64d49762..48cce4d8f6 100644 --- a/pkg/sink/kafka/options_test.go +++ b/pkg/sink/kafka/options_test.go @@ -69,6 +69,18 @@ func TestCompleteOptions(t *testing.T) { options = NewOptions() err = options.Apply(commonType.NewChangefeedID4Test(commonType.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 + sinkURI, err = url.Parse(uri) + require.NoError(t, err) + options = NewOptions() + err = options.Apply( + commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), + sinkURI, + config.GetDefaultReplicaConfig().Sink, + ) + require.ErrorContains(t, err, "invalid replication-factor "+replicationFactor) + } // Illegal max-message-bytes. uri = "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&max-message-bytes=a" @@ -321,9 +333,55 @@ func TestAdjustConfigTopicExist(t *testing.T) { // When the topic exists, but the topic does not have `max.message.bytes` // create a topic without `max.message.bytes` topicName := "test-topic" +<<<<<<< HEAD detail := &TopicDetail{ Name: topicName, NumPartitions: 3, +======= + changefeedID := commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test") + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + adminFixture := newKafkaAdminFixture(t) + adminClient := adminFixture.admin + + detail := &TopicDetail{ + Name: topicName, + NumPartitions: 3, + } + err := adminClient.CreateTopic(detail, false) + require.NoError(t, err) + + configuredMaxMessageBytes := test.configuredMaxMessageBytes(adminFixture) + sinkURI, err := url.Parse(fmt.Sprintf( + "kafka://127.0.0.1:9092/%s?max-message-bytes=%d", + topicName, configuredMaxMessageBytes, + )) + require.NoError(t, err) + + options := NewOptions() + err = options.Apply(changefeedID, sinkURI, config.GetDefaultReplicaConfig().Sink) + require.NoError(t, err) + require.Equal(t, configuredMaxMessageBytes, options.MaxMessageBytes) + require.Equal(t, configuredMaxMessageBytes, options.MaxBatchedBytes) + expectedProducerLimit := adminFixture.brokerMessageMaxBytes() + + ctx := context.Background() + err = adjustOptions(changefeedID, adminClient, options, topicName) + require.NoError(t, err) + + saramaConfig, err := newSaramaConfig(ctx, options) + require.NoError(t, err) + + require.NotEqual(t, configuredMaxMessageBytes, options.MaxMessageBytes) + require.Equal(t, expectedProducerLimit, options.MaxMessageBytes) + require.Equal( + t, + min(configuredMaxMessageBytes, expectedProducerLimit), + options.MaxBatchedBytes, + ) + require.Equal(t, expectedProducerLimit, saramaConfig.Producer.MaxMessageBytes) + }) +>>>>>>> 0d4929739 (kafka: verify replication-factor when need to create the topic (#5715)) } err = adminClient.CreateTopic(detail, false) require.NoError(t, err) @@ -353,6 +411,7 @@ func TestAdjustConfigTopicExist(t *testing.T) { require.Equal(t, maxMessageBytes, saramaConfig.Producer.MaxMessageBytes) } +<<<<<<< HEAD func TestAdjustConfigMinInsyncReplicas(t *testing.T) { adminClient := NewClusterAdminClientMockImpl() defer adminClient.Close() @@ -371,12 +430,26 @@ func TestAdjustConfigMinInsyncReplicas(t *testing.T) { options, "create-new-fail-invalid-min-insync-replicas", ) +======= +func TestValidateReplicationFactor(t *testing.T) { + adminFixture := newKafkaAdminFixture(t) + adminClient := adminFixture.admin + adminFixture.setMinInsyncReplicas("2") + + topicConfig := &AutoCreateTopicConfig{ + AutoCreate: true, + ReplicationFactor: 1, + RequiredAcks: WaitForAll, + } + err := topicConfig.ValidateReplicationFactor(adminClient) +>>>>>>> 0d4929739 (kafka: verify replication-factor when need to create the topic (#5715)) require.Regexp( t, ".*`replication-factor` 1 is smaller than the `min.insync.replicas` 2 of broker.*", errors.Cause(err), ) +<<<<<<< HEAD // topic not exist, and `min.insync.replicas` not found in broker's configuration adminClient.DropBrokerConfig(MinInsyncReplicasConfigName) topicName := "no-topic-no-min-insync-replicas" @@ -384,10 +457,17 @@ func TestAdjustConfigMinInsyncReplicas(t *testing.T) { require.Nil(t, err) err = adminClient.CreateTopic(&TopicDetail{ Name: topicName, +======= + localAcksConfig := &AutoCreateTopicConfig{ + AutoCreate: true, +>>>>>>> 0d4929739 (kafka: verify replication-factor when need to create the topic (#5715)) ReplicationFactor: 1, - }, false) - require.ErrorIs(t, err, sarama.ErrPolicyViolation) + RequiredAcks: WaitForLocal, + } + err = localAcksConfig.ValidateReplicationFactor(adminClient) + require.NoError(t, err) +<<<<<<< HEAD // Report an error if the replication-factor is less than min.insync.replicas // when the topic does exist. @@ -428,6 +508,16 @@ func TestSkipAdjustConfigMinInsyncReplicasWhenRequiredAcksIsNotWailAll(t *testin "skip-check-min-insync-replicas", ) require.Nil(t, err, "Should not report an error when `required-acks` is not `all`") +======= + adminFixture.dropBrokerConfig(MinInsyncReplicasConfigName) + missingBrokerConfig := &AutoCreateTopicConfig{ + AutoCreate: true, + ReplicationFactor: 1, + RequiredAcks: WaitForAll, + } + err = missingBrokerConfig.ValidateReplicationFactor(adminClient) + require.NoError(t, err) +>>>>>>> 0d4929739 (kafka: verify replication-factor when need to create the topic (#5715)) } func TestCreateProducerFailed(t *testing.T) { @@ -640,10 +730,17 @@ func TestConfigurationCombinations(t *testing.T) { sinkURI, err := url.Parse(uri) require.Nil(t, err) +<<<<<<< HEAD ctx := context.Background() options := NewOptions() err = options.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) require.Nil(t, err) +======= + options := NewOptions() + err = options.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) + require.Nil(t, err) + configuredMaxMessageBytes := options.MaxMessageBytes +>>>>>>> 0d4929739 (kafka: verify replication-factor when need to create the topic (#5715)) changefeed := commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "changefeed-test") factory, err := NewMockFactory(ctx, options, changefeed) @@ -652,11 +749,23 @@ func TestConfigurationCombinations(t *testing.T) { adminClient, err := factory.AdminClient(ctx) require.NoError(t, err) +<<<<<<< HEAD topic, ok := a.uriParams[0].(string) require.True(t, ok) require.NotEqual(t, "", topic) err = adjustOptions(ctx, adminClient, options, topic) require.Nil(t, err) +======= + changefeedID := commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test") + err = adjustOptions(changefeedID, adminClient, options, topic) + require.Nil(t, err) + require.Equal(t, sourceMaxMessageBytes, options.MaxMessageBytes) + require.Equal( + t, + min(configuredMaxMessageBytes, sourceMaxMessageBytes), + options.MaxBatchedBytes, + ) +>>>>>>> 0d4929739 (kafka: verify replication-factor when need to create the topic (#5715)) encoderConfig := common.NewConfig(config.ProtocolOpen) err = encoderConfig.Apply(sinkURI, &config.SinkConfig{ diff --git a/pkg/sink/kafka/sarama_factory.go b/pkg/sink/kafka/sarama_factory.go index 650f346b4c..5d9e6882c2 100644 --- a/pkg/sink/kafka/sarama_factory.go +++ b/pkg/sink/kafka/sarama_factory.go @@ -57,7 +57,11 @@ func NewSaramaFactory( admin.Close() }() +<<<<<<< HEAD if err = adjustOptions(ctx, admin, o, o.Topic); err != nil { +======= + if err = adjustOptions(changefeedID, admin, o, o.Topic); err != nil { +>>>>>>> 0d4929739 (kafka: verify replication-factor when need to create the topic (#5715)) return nil, errors.Trace(err) }