diff --git a/downstreamadapter/sink/cloudstorage/encoder_group_test.go b/downstreamadapter/sink/cloudstorage/encoder_group_test.go index f5df7b6454..e77cfc59bb 100644 --- a/downstreamadapter/sink/cloudstorage/encoder_group_test.go +++ b/downstreamadapter/sink/cloudstorage/encoder_group_test.go @@ -230,7 +230,6 @@ func newTestTxnEncoderConfig(t *testing.T) *common.Config { config.ProtocolCsv, replicaConfig.Sink, config.DefaultMaxMessageBytes, - config.DefaultMaxMessageBytes, ) require.NoError(t, err) return encoderConfig diff --git a/downstreamadapter/sink/cloudstorage/sink.go b/downstreamadapter/sink/cloudstorage/sink.go index 0766816315..ff6920307a 100644 --- a/downstreamadapter/sink/cloudstorage/sink.go +++ b/downstreamadapter/sink/cloudstorage/sink.go @@ -86,7 +86,7 @@ func Verify(ctx context.Context, changefeedID common.ChangeFeedID, sinkURI *url. if err != nil { return err } - _, err = helper.GetEncoderConfig(changefeedID, sinkURI, protocol, sinkConfig, math.MaxInt, math.MaxInt) + _, err = helper.GetEncoderConfig(changefeedID, sinkURI, protocol, sinkConfig, math.MaxInt) if err != nil { return err } @@ -116,9 +116,9 @@ func New( } // get cloud storage file extension according to the specific protocol. ext := helper.GetFileExtension(protocol) - // Message size limits are mainly for MQ batch protocols. Cloud storage uses - // max int for both the final message limit and the batch threshold. - encoderConfig, err := helper.GetEncoderConfig(changefeedID, sinkURI, protocol, sinkConfig, math.MaxInt, math.MaxInt) + // the last param maxMsgBytes is mainly to limit the size of a single message for + // batch protocols in mq scenario. In cloud storage sink, we just set it to max int. + encoderConfig, err := helper.GetEncoderConfig(changefeedID, sinkURI, protocol, sinkConfig, math.MaxInt) if err != nil { return nil, err } diff --git a/downstreamadapter/sink/helper/helper.go b/downstreamadapter/sink/helper/helper.go index 3911ea6ab5..47cd949220 100644 --- a/downstreamadapter/sink/helper/helper.go +++ b/downstreamadapter/sink/helper/helper.go @@ -50,16 +50,17 @@ func GetEncoderConfig( sinkURI *url.URL, protocol config.Protocol, sinkConfig *config.SinkConfig, - maxMessageBytes int, - maxBatchedBytes int, + maxMsgBytes int, ) (*common.Config, error) { encoderConfig := common.NewConfig(protocol) if err := encoderConfig.Apply(sinkURI, sinkConfig); err != nil { return nil, errors.WrapError(errors.ErrSinkInvalidConfig, err) } + // Always set encoder's `MaxMessageBytes` equal to producer's `MaxMessageBytes` + // to prevent that the encoder generate batched message too large + // then cause producer meet `message too large`. encoderConfig = encoderConfig. - WithMaxMessageBytes(maxMessageBytes). - WithMaxBatchedBytes(maxBatchedBytes). + WithMaxMessageBytes(maxMsgBytes). WithChangefeedID(changefeedID) tz, err := util.GetTimezone(config.GetGlobalServerConfig().TZ) diff --git a/downstreamadapter/sink/kafka/helper.go b/downstreamadapter/sink/kafka/helper.go index ebd5b915e4..733278a59a 100644 --- a/downstreamadapter/sink/kafka/helper.go +++ b/downstreamadapter/sink/kafka/helper.go @@ -102,10 +102,7 @@ func newKafkaSinkComponent( return comp, protocol, err } - encoderConfig, err := helper.GetEncoderConfig( - changefeedID, sinkURI, protocol, sinkConfig, - options.MaxMessageBytes, options.MaxBatchedBytes, - ) + encoderConfig, err := helper.GetEncoderConfig(changefeedID, sinkURI, protocol, sinkConfig, options.MaxMessageBytes) if err != nil { return comp, protocol, err } diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index f6351fb269..6433f40d7e 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -89,10 +89,7 @@ func Verify(ctx context.Context, changefeedID common.ChangeFeedID, uri *url.URL, } options.Topic = topic - encoderConfig, err := helper.GetEncoderConfig( - changefeedID, uri, protocol, sinkConfig, - options.MaxMessageBytes, options.MaxBatchedBytes, - ) + encoderConfig, err := helper.GetEncoderConfig(changefeedID, uri, protocol, sinkConfig, options.MaxMessageBytes) if err != nil { return err } diff --git a/downstreamadapter/sink/kafka/sink_test.go b/downstreamadapter/sink/kafka/sink_test.go index 9c6e49b508..330b981908 100644 --- a/downstreamadapter/sink/kafka/sink_test.go +++ b/downstreamadapter/sink/kafka/sink_test.go @@ -41,19 +41,6 @@ import ( const kafkaSinkTestTopic = "mock_topic" -func TestVerifyValidatesEncoderConfigBeforeKafkaConnection(t *testing.T) { - openProtocol := config.ProtocolOpen.String() - sinkConfig := &config.SinkConfig{Protocol: &openProtocol} - sinkURI, err := url.Parse("kafka://127.0.0.1:1/" + kafkaSinkTestTopic + "?max-batch-size=0") - require.NoError(t, err) - - changefeedID := common.NewChangefeedID4Test("test", "verify-existing-topic") - ctx, cancel := context.WithTimeout(context.Background(), time.Second) - defer cancel() - err = Verify(ctx, changefeedID, sinkURI, sinkConfig) - require.ErrorContains(t, err, "invalid max-batch-size 0") -} - func TestSinkWorkersReturnContextError(t *testing.T) { contexts := []struct { name string @@ -197,10 +184,7 @@ func newKafkaSinkForTestWithProducers(ctx context.Context, if err != nil { return nil, err } - encoderConfig, err := helper.GetEncoderConfig( - changefeedID, sinkURI, protocol, sinkConfig, - options.MaxMessageBytes, options.MaxBatchedBytes, - ) + encoderConfig, err := helper.GetEncoderConfig(changefeedID, sinkURI, protocol, sinkConfig, options.MaxMessageBytes) if err != nil { return nil, err } diff --git a/downstreamadapter/sink/pulsar/helper.go b/downstreamadapter/sink/pulsar/helper.go index a2d82f7af6..ffb503a8f3 100644 --- a/downstreamadapter/sink/pulsar/helper.go +++ b/downstreamadapter/sink/pulsar/helper.go @@ -122,10 +122,7 @@ func newPulsarSinkComponentWithFactory(ctx context.Context, return pulsarComponent, protocol, errors.Trace(err) } - encoderConfig, err := helper.GetEncoderConfig( - changefeedID, sinkURI, protocol, sinkConfig, - config.DefaultMaxMessageBytes, config.DefaultMaxMessageBytes, - ) + encoderConfig, err := helper.GetEncoderConfig(changefeedID, sinkURI, protocol, sinkConfig, config.DefaultMaxMessageBytes) if err != nil { return pulsarComponent, protocol, errors.Trace(err) } diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go index b00bf947d0..13b371f697 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go @@ -212,42 +212,50 @@ func TestEnsureTopicExistsWaitsUntilVisible(t *testing.T) { ctrl := gomock.NewController(t) adminClient := kafka.NewMockClusterAdminClient(ctrl) - cfg := &kafka.AutoCreateTopicConfig{ - AutoCreate: true, - PartitionNum: 2, - ReplicationFactor: 1, - } - - topic := "delayed-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().CreateTopic(gomock.Any()).DoAndReturn( - func(detail *kafka.TopicDetail) error { - require.Equal(t, &kafka.TopicDetail{ - Name: topic, - NumPartitions: 2, - ReplicationFactor: 1, - }, detail) - return nil - }), - adminClient.EXPECT().GetTopicsMeta([]string{topic}, false).Return( - map[string]kafka.TopicDetail{}, nil), - adminClient.EXPECT().GetTopicsMeta([]string{topic}, false).Return( - map[string]kafka.TopicDetail{ - topic: { - Name: topic, + created := false + postCreateDescribeCount := 0 + adminClient.EXPECT().GetTopicsMeta([]string{"delayed-topic"}, true).Return(map[string]kafka.TopicDetail{}, nil) + adminClient.EXPECT().GetTopicsMeta([]string{"delayed-topic"}, false).DoAndReturn( + func([]string, bool) (map[string]kafka.TopicDetail, error) { + if !created { + return map[string]kafka.TopicDetail{}, nil + } + postCreateDescribeCount++ + if postCreateDescribeCount == 1 { + return map[string]kafka.TopicDetail{}, nil + } + return map[string]kafka.TopicDetail{ + "delayed-topic": { + Name: "delayed-topic", NumPartitions: 2, }, - }, nil), + }, nil + }).Times(3) + adminClient.EXPECT().CreateTopic(gomock.Any()).DoAndReturn( + func(detail *kafka.TopicDetail) error { + require.Equal(t, &kafka.TopicDetail{ + Name: "delayed-topic", + NumPartitions: 2, + ReplicationFactor: 1, + }, detail) + created = true + return nil + }) + + err := EnsureTopic( + context.Background(), + common.NewChangefeedID4Test("test", "test"), + "delayed-topic", + &kafka.AutoCreateTopicConfig{ + AutoCreate: true, + PartitionNum: 2, + ReplicationFactor: 1, + }, + adminClient, ) - ctx := context.Background() - changefeedID := common.NewChangefeedID4Test("test", "test") - err := EnsureTopic(ctx, changefeedID, topic, cfg, adminClient) require.NoError(t, err) + require.Equal(t, 2, postCreateDescribeCount) } func TestGetTopicManagerStartsBackgroundRefreshAfterTopicReady(t *testing.T) { diff --git a/pkg/sink/codec/canal/canal_json_txn_encoder.go b/pkg/sink/codec/canal/canal_json_txn_encoder.go index 3e10d98229..0af6f4f3f2 100644 --- a/pkg/sink/codec/canal/canal_json_txn_encoder.go +++ b/pkg/sink/codec/canal/canal_json_txn_encoder.go @@ -65,6 +65,7 @@ func (j *JSONTxnEventEncoder) AppendTxnEvent(event *commonEvent.DMLEvent) error return err } length := len(value) + common.MaxRecordOverhead + // For single message that is longer than max-message-bytes, do not send it. if length > j.config.MaxMessageBytes { log.Warn("Single message is too large for canal-json", zap.Int("maxMessageBytes", j.config.MaxMessageBytes), diff --git a/pkg/sink/codec/common/config.go b/pkg/sink/codec/common/config.go index 415e864a05..c55c30fac0 100644 --- a/pkg/sink/codec/common/config.go +++ b/pkg/sink/codec/common/config.go @@ -41,12 +41,9 @@ type Config struct { Protocol config.Protocol + // control batch behavior, only for `open-protocol` and `craft` at the moment. MaxMessageBytes int - - // MaxBatchedBytes controls open-protocol encoder's maximum number of bytes for a batched message. - MaxBatchedBytes int - // MaxBatchedSize controls open-protocol encoder's maximum number of events for a batched message. - MaxBatchSize int + MaxBatchSize int // DeleteOnlyHandleKeyColumns is true, for the delete event only output the handle key columns. DeleteOnlyHandleKeyColumns bool @@ -117,7 +114,6 @@ func NewConfig(protocol config.Protocol) *Config { Protocol: protocol, MaxMessageBytes: config.DefaultMaxMessageBytes, - MaxBatchedBytes: config.DefaultMaxMessageBytes, MaxBatchSize: defaultMaxBatchSize, EnableTiDBExtension: false, @@ -348,12 +344,6 @@ func (c *Config) WithMaxMessageBytes(bytes int) *Config { return c } -// WithMaxBatchedBytes sets the maximum batched message bytes. -func (c *Config) WithMaxBatchedBytes(bytes int) *Config { - c.MaxBatchedBytes = bytes - return c -} - // WithChangefeedID set the `changefeedID` func (c *Config) WithChangefeedID(id common.ChangeFeedID) *Config { c.ChangefeedID = id @@ -423,12 +413,6 @@ func (c *Config) Validate() error { if c.MaxMessageBytes <= 0 { return errors.ErrCodecInvalidConfig.GenWithStack("invalid max-message-bytes %d", c.MaxMessageBytes) } - if c.MaxBatchedBytes < 0 { - return errors.ErrCodecInvalidConfig.GenWithStack("invalid max-batch-message-bytes %d", c.MaxBatchedBytes) - } - if c.MaxBatchedBytes > c.MaxMessageBytes { - return errors.ErrCodecInvalidConfig.GenWithStack("max-batch-message-bytes %d cannot be greater than max-message-bytes %d", c.MaxBatchedBytes, c.MaxMessageBytes) - } if c.MaxBatchSize <= 0 { return errors.ErrCodecInvalidConfig.GenWithStack("invalid max-batch-size %d", c.MaxBatchSize) diff --git a/pkg/sink/codec/common/config_test.go b/pkg/sink/codec/common/config_test.go index 8a49fb3ff1..4824bbfaec 100644 --- a/pkg/sink/codec/common/config_test.go +++ b/pkg/sink/codec/common/config_test.go @@ -33,7 +33,7 @@ func TestApplyReturnsSinkInvalidConfigForQueryBindingError(t *testing.T) { require.Equal(t, errors.ErrSinkInvalidConfig.RFCCode(), errCode) } -func TestValidateMaxBatchMessageBytes(t *testing.T) { +func TestValidateMessageLimits(t *testing.T) { tests := []struct { name string adjust func(*Config) @@ -46,21 +46,6 @@ func TestValidateMaxBatchMessageBytes(t *testing.T) { }, expected: "invalid max-message-bytes 0", }, - { - name: "negative max batched bytes", - adjust: func(cfg *Config) { - cfg.MaxBatchedBytes = -1 - }, - expected: "invalid max-batch-message-bytes -1", - }, - { - name: "max batched bytes exceeds max message bytes", - adjust: func(cfg *Config) { - cfg.MaxMessageBytes = 100 - cfg.MaxBatchedBytes = 101 - }, - expected: "max-batch-message-bytes 101 cannot be greater than max-message-bytes 100", - }, { name: "non-positive max batch size", adjust: func(cfg *Config) { diff --git a/pkg/sink/codec/open/encoder.go b/pkg/sink/codec/open/encoder.go index b8e608a298..567ce8b60f 100644 --- a/pkg/sink/codec/open/encoder.go +++ b/pkg/sink/codec/open/encoder.go @@ -38,7 +38,7 @@ var ( ) // batchEncoder for open protocol will batch multiple row changed events into a single message. -// One message can contain at most MaxBatchSize events, and the total size of the message cannot exceed MaxBatchedBytes. +// One message can contain at most MaxBatchSize events, and the total size of the message cannot exceed MaxMessageBytes. type batchEncoder struct { messages []*common.Message // buff the callback of the latest message @@ -164,7 +164,7 @@ func (d *batchEncoder) pushMessage(key, value []byte, callback func()) { binary.BigEndian.PutUint64(keyLenByte[:], uint64(len(key))) binary.BigEndian.PutUint64(valueLenByte[:], uint64(len(value))) - if len(d.messages) == 0 || d.messages[len(d.messages)-1].Length()+length > d.config.MaxBatchedBytes || d.messages[len(d.messages)-1].GetRowsCount() >= d.config.MaxBatchSize { + if len(d.messages) == 0 || d.messages[len(d.messages)-1].Length()+length > d.config.MaxMessageBytes || d.messages[len(d.messages)-1].GetRowsCount() >= d.config.MaxBatchSize { d.finalizeCallback() // create a new message versionHead := make([]byte, 8) diff --git a/pkg/sink/codec/open/encoder_test.go b/pkg/sink/codec/open/encoder_test.go index 026190d2e2..acd79528e2 100644 --- a/pkg/sink/codec/open/encoder_test.go +++ b/pkg/sink/codec/open/encoder_test.go @@ -778,9 +778,7 @@ func TestEncoderMultipleMessage(t *testing.T) { `insert into test.t values (3, 333)`) ctx := context.Background() - codecConfig := common.NewConfig(config.ProtocolOpen). - WithMaxMessageBytes(1000). - WithMaxBatchedBytes(400) + codecConfig := common.NewConfig(config.ProtocolOpen).WithMaxMessageBytes(400) encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) @@ -810,13 +808,11 @@ func TestEncoderMultipleMessage(t *testing.T) { require.Equal(t, 2, len(messages)) require.Equal(t, 2, messages[0].GetRowsCount()) require.Equal(t, 1, messages[1].GetRowsCount()) - require.LessOrEqual(t, messages[0].Length(), codecConfig.MaxBatchedBytes) - require.LessOrEqual(t, messages[1].Length(), codecConfig.MaxBatchedBytes) - require.Equal(t, 0, count) - messages[0].Callback() - require.Equal(t, 2, count) - messages[1].Callback() + for _, message := range messages { + message.Callback() + } + require.Equal(t, 3, count) decoder, err := NewDecoder(ctx, 0, codecConfig, nil) @@ -889,48 +885,6 @@ func TestMessageTooLarge(t *testing.T) { require.Equal(t, count, 0) } -func TestMessageLargerThanBatchLimit(t *testing.T) { - ctx := context.Background() - codecConfig := common.NewConfig(config.ProtocolOpen). - WithMaxMessageBytes(400). - WithMaxBatchedBytes(100) - encoder, err := NewBatchEncoder(codecConfig, nil) - require.NoError(t, err) - - helper := commonEvent.NewEventTestHelper(t) - defer helper.Close() - helper.Tk().MustExec("use test") - - job := helper.DDL2Job(`create table test.t(a tinyint primary key, b int)`) - tableInfo := helper.GetTableInfo(job) - dmlEvent := helper.DML2Event("test", "t", `insert into test.t values (1, 123)`) - require.NotNil(t, dmlEvent) - insertRow, ok := dmlEvent.GetNextRow() - require.True(t, ok) - - count := 0 - insertRowEvent := &commonEvent.RowEvent{ - TableInfo: tableInfo, - CommitTs: dmlEvent.GetCommitTs(), - Event: insertRow, - ColumnSelector: columnselector.NewDefaultColumnSelector(), - Callback: func() { count++ }, - } - - err = encoder.AppendRowChangedEvent(ctx, "", insertRowEvent) - require.NoError(t, err) - - messages := encoder.Build() - require.Len(t, messages, 1) - require.Equal(t, 1, messages[0].GetRowsCount()) - require.Greater(t, messages[0].Length(), codecConfig.MaxBatchedBytes) - require.LessOrEqual(t, messages[0].Length(), codecConfig.MaxMessageBytes) - require.Equal(t, 0, count) - - messages[0].Callback() - require.Equal(t, 1, count) -} - func TestLargeMessageWithHandleEnableHandleKeyOnly(t *testing.T) { helper := commonEvent.NewEventTestHelper(t) defer helper.Close() diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go index 1e64238f9b..5d42c58dd5 100644 --- a/pkg/sink/kafka/options.go +++ b/pkg/sink/kafka/options.go @@ -40,6 +40,13 @@ const ( defaultMaxRetry = 5 // defaultTimeout is the default timeout for Kafka connections. defaultTimeout = 10 * time.Second + + // the `max-message-bytes` is set equal to topic's `max.message.bytes`, and is used to check + // whether the message is larger than the max size limit. It's found some message pass the message + // size limit check at the client side and failed at the broker side since message enlarged during + // the network transmission. so we set the `max-message-bytes` to a smaller value to avoid this problem. + // maxMessageBytesOverhead is used to reduce the `max-message-bytes`. + maxMessageBytesOverhead = 128 ) const ( @@ -151,15 +158,14 @@ type options struct { Version string IsAssignedVersion bool RequestVersion int16 - // MaxMessageBytes controls the byte size limit of the producer and encoded messages. - MaxMessageBytes int - // MaxBatchedBytes controls the byte size limit when batching messages. - MaxBatchedBytes int - - MaxRetry int - Compression string - ClientID string - RequiredAcks RequiredAcks + MaxMessageBytes int + MaxRetry int + Compression string + ClientID string + RequiredAcks RequiredAcks + // Only for test. User can not set this value. + // The current prod default value is 0. + MaxMessages int // Credential is used to connect to kafka cluster. EnableTLS bool @@ -176,9 +182,9 @@ type options struct { // NewOptions returns a default Kafka configuration func NewOptions() *options { return &options{ - Version: "2.4.0", + Version: "2.4.0", + // MaxMessageBytes will be used to initialize producer MaxMessageBytes: config.DefaultMaxMessageBytes, - MaxBatchedBytes: config.DefaultMaxMessageBytes, MaxRetry: defaultMaxRetry, ReplicationFactor: 1, Compression: "none", @@ -255,13 +261,8 @@ func (o *options) Apply(changefeedID common.ChangeFeedID, } if urlParameter.MaxMessageBytes != nil { - if *urlParameter.MaxMessageBytes <= 0 { - return errors.ErrKafkaInvalidConfig.GenWithStack( - "invalid max-message-bytes %d", *urlParameter.MaxMessageBytes) - } o.MaxMessageBytes = *urlParameter.MaxMessageBytes } - o.MaxBatchedBytes = o.MaxMessageBytes if urlParameter.MaxRetry != nil && *urlParameter.MaxRetry >= 0 { o.MaxRetry = *urlParameter.MaxRetry @@ -622,9 +623,7 @@ func NewKafkaClientID(captureAddr string, return } -// adjustOptions adjusts options with Kafka runtime metadata. -// It overwrites MaxMessageBytes with the final producer message limit derived -// from the topic or broker configuration. +// adjustOptions adjust the `options` and `sarama.Config` by condition. func adjustOptions( changefeedID common.ChangeFeedID, admin ClusterAdminClient, @@ -640,98 +639,83 @@ func adjustOptions( // once we have found the topic, no matter `auto-create-topic`, // make sure user input parameters are valid. if exists { - err = adjustExistingTopicOption(changefeedID, admin, options, info) - } else { - adjustNewTopicOptions(changefeedID, admin, options) - } - if err != nil { - return err - } + // make sure that producer's `MaxMessageBytes` smaller than topic's `max.message.bytes` + topicMaxMessageBytesStr, found, err := getTopicConfig( + admin, info.Name, + TopicMaxMessageBytesConfigName, + BrokerMessageMaxBytesConfigName, + ) + if err != nil { + return err + } + if !found { + return errors.ErrKafkaAdminAPI.GenWithStack( + "Kafka configuration %s not found in topic %s or broker", + TopicMaxMessageBytesConfigName, info.Name) + } + topicMaxMessageBytes, err := strconv.Atoi(topicMaxMessageBytesStr) + if err != nil { + return errors.WrapError(errors.ErrKafkaAdminAPI, err, + "parse-config", TopicMaxMessageBytesConfigName) + } - options.MaxBatchedBytes = min(options.MaxBatchedBytes, options.MaxMessageBytes) - return nil -} + maxMessageBytes := topicMaxMessageBytes - maxMessageBytesOverhead + if topicMaxMessageBytes <= options.MaxMessageBytes { + log.Warn("topic's `max.message.bytes` less than the `max-message-bytes`,"+ + "use topic's `max.message.bytes` to initialize the Kafka producer", + zap.Int("max.message.bytes", topicMaxMessageBytes), + zap.Int("max-message-bytes", options.MaxMessageBytes), + zap.Int("real-max-message-bytes", maxMessageBytes)) + options.MaxMessageBytes = maxMessageBytes + } else if maxMessageBytes < options.MaxMessageBytes { + options.MaxMessageBytes = maxMessageBytes + } -func adjustExistingTopicOption( - changefeedID common.ChangeFeedID, - admin ClusterAdminClient, - options *options, - info TopicDetail, -) error { - maxMessageBytes, found, err := getTopicMaxMessageBytes(admin, info.Name) - if err != nil || !found { - log.Warn("kafka topic `max.message.bytes` unavailable, using configured value", - zap.String("namespace", changefeedID.Keyspace()), zap.String("changefeed", changefeedID.Name()), - zap.Int("maxMessageBytes", options.MaxMessageBytes), zap.Error(err)) - maxMessageBytes = options.MaxMessageBytes + // no need to create the topic, + // but we would have to log user if they found enter wrong topic name later + if err = options.setPartitionNum(changefeedID, info.NumPartitions); err != nil { + return err + } + + return nil } - options.MaxMessageBytes = maxMessageBytes - if err = options.setPartitionNum(changefeedID, info.NumPartitions); err != nil { + brokerMessageMaxBytesStr, found, err := admin.GetBrokerConfig(BrokerMessageMaxBytesConfigName) + if err != nil { + log.Warn("TiCDC cannot find `message.max.bytes` from broker's configuration") return err } - return nil -} + if !found { + return errors.ErrKafkaAdminAPI.GenWithStack( + "Kafka broker configuration %s not found", BrokerMessageMaxBytesConfigName) + } + brokerMessageMaxBytes, err := strconv.Atoi(brokerMessageMaxBytesStr) + if err != nil { + return errors.WrapError(errors.ErrKafkaAdminAPI, err, + "parse-config", BrokerMessageMaxBytesConfigName) + } -func adjustNewTopicOptions( - changefeedID common.ChangeFeedID, - admin ClusterAdminClient, - options *options, -) { // 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, found, err := getBrokerMaxMessageBytes(admin) - if err != nil || !found { - log.Warn("kafka broker `message.max.bytes` unavailable, using configured value", - zap.String("namespace", changefeedID.Keyspace()), zap.String("changefeed", changefeedID.Name()), - zap.Int("maxMessageBytes", options.MaxMessageBytes), zap.Error(err)) - messageMaxBytes = options.MaxMessageBytes + // TiCDC need to make sure that the producer's `MaxMessageBytes` won't larger than + // broker's `message.max.bytes`. + maxMessageBytes := brokerMessageMaxBytes - maxMessageBytesOverhead + if brokerMessageMaxBytes <= options.MaxMessageBytes { + log.Warn("broker's `message.max.bytes` less than the `max-message-bytes`,"+ + "use broker's `message.max.bytes` to initialize the Kafka producer", + zap.Int("message.max.bytes", brokerMessageMaxBytes), + zap.Int("max-message-bytes", options.MaxMessageBytes), + zap.Int("real-max-message-bytes", maxMessageBytes)) + options.MaxMessageBytes = maxMessageBytes + } else if maxMessageBytes < options.MaxMessageBytes { + options.MaxMessageBytes = maxMessageBytes } - options.MaxMessageBytes = messageMaxBytes // topic not exists yet, and user does not specify the `partition-num` in the sink uri. if options.PartitionNum == 0 { options.PartitionNum = defaultPartitionNum } -} - -func getTopicMaxMessageBytes( - admin ClusterAdminClient, - topic string, -) (int, bool, error) { - raw, found, err := getTopicConfig( - admin, topic, - TopicMaxMessageBytesConfigName, - BrokerMessageMaxBytesConfigName, - ) - if err != nil { - return 0, false, err - } - if !found { - return 0, false, nil - } - maxMessageBytes, err := strconv.Atoi(raw) - if err != nil { - return 0, false, errors.WrapError( - errors.ErrKafkaAdminAPI, err, "parse-config", TopicMaxMessageBytesConfigName) - } - return maxMessageBytes, true, nil -} - -func getBrokerMaxMessageBytes(admin ClusterAdminClient) (int, bool, error) { - raw, found, err := admin.GetBrokerConfig(BrokerMessageMaxBytesConfigName) - if err != nil { - return 0, false, err - } - if !found { - return 0, false, nil - } - messageMaxBytes, err := strconv.Atoi(raw) - if err != nil { - return 0, false, errors.WrapError( - errors.ErrKafkaAdminAPI, err, "parse-config", BrokerMessageMaxBytesConfigName) - } - return messageMaxBytes, true, nil + return nil } // getTopicConfig gets topic config by name. diff --git a/pkg/sink/kafka/options_test.go b/pkg/sink/kafka/options_test.go index 9c0ca3c6ed..66d6e6cee1 100644 --- a/pkg/sink/kafka/options_test.go +++ b/pkg/sink/kafka/options_test.go @@ -28,6 +28,7 @@ import ( "github.com/pingcap/ticdc/pkg/common" "github.com/pingcap/ticdc/pkg/config" "github.com/pingcap/ticdc/pkg/errors" + codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/stretchr/testify/require" ) @@ -154,6 +155,14 @@ func (f *kafkaAdminFixture) setMessageMaxBytes(brokerValue, topicValue string) { f.topicConfig[defaultMockTopicName][TopicMaxMessageBytesConfigName] = topicValue } +func expectedAdjustedMaxMessageBytes(configuredMaxMessageBytes, sourceMaxMessageBytes int) int { + sourceMaxMessageBytes -= maxMessageBytesOverhead + if configuredMaxMessageBytes < sourceMaxMessageBytes { + return configuredMaxMessageBytes + } + return sourceMaxMessageBytes +} + func (f *kafkaAdminFixture) setMinInsyncReplicas(minInsyncReplicas string) { f.topicConfig[defaultMockTopicName][MinInsyncReplicasConfigName] = minInsyncReplicas f.brokerConfig[MinInsyncReplicasConfigName] = minInsyncReplicas @@ -181,7 +190,6 @@ func TestCompleteOptions(t *testing.T) { require.Equal(t, int16(3), options.ReplicationFactor) require.Equal(t, "2.6.0", options.Version) require.Equal(t, 4096, options.MaxMessageBytes) - require.Equal(t, 4096, options.MaxBatchedBytes) require.Equal(t, WaitForLocal, options.RequiredAcks) require.Equal(t, defaultMaxRetry, options.MaxRetry) @@ -290,60 +298,6 @@ func TestCompleteOptions(t *testing.T) { require.Equal(t, defaultMaxRetry, options.MaxRetry) } -func TestApplyRejectsNonPositiveMaxMessageBytes(t *testing.T) { - tests := []struct { - name string - uri string - configValue *int - expected int - }{ - { - name: "zero from URI", - uri: "kafka://127.0.0.1:9092/test-topic?max-message-bytes=0", - expected: 0, - }, - { - name: "negative from URI", - uri: "kafka://127.0.0.1:9092/test-topic?max-message-bytes=-1", - expected: -1, - }, - { - name: "zero from sink config", - uri: "kafka://127.0.0.1:9092/test-topic", - configValue: aws.Int(0), - expected: 0, - }, - { - name: "negative from sink config", - uri: "kafka://127.0.0.1:9092/test-topic", - configValue: aws.Int(-1), - expected: -1, - }, - } - - changefeedID := common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test") - for _, test := range tests { - t.Run(test.name, func(t *testing.T) { - sinkURI, err := url.Parse(test.uri) - require.NoError(t, err) - - sinkConfig := config.GetDefaultReplicaConfig().Sink - if test.configValue != nil { - sinkConfig.KafkaConfig = &config.KafkaConfig{ - MaxMessageBytes: test.configValue, - } - } - - options := NewOptions() - err = options.Apply(changefeedID, sinkURI, sinkConfig) - require.ErrorContains(t, err, fmt.Sprintf("invalid max-message-bytes %d", test.expected)) - errCode, ok := errors.RFCCode(err) - require.True(t, ok) - require.Equal(t, errors.ErrKafkaInvalidConfig.RFCCode(), errCode) - }) - } -} - func TestSetPartitionNum(t *testing.T) { options := NewOptions() changefeedID := common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test") @@ -455,13 +409,13 @@ func TestAdjustConfigFallsBackToBrokerMessageMaxBytesWhenTopicConfigMissing(t *t configuredMaxMessageBytes func(*kafkaAdminFixture) int }{ { - name: "uses broker limit when configured value is below broker", + name: "keeps configured value below broker limit", configuredMaxMessageBytes: func(*kafkaAdminFixture) int { return 1024 }, }, { - name: "uses broker limit when configured value is below broker by one byte", + name: "uses broker limit when configured value is within overhead", configuredMaxMessageBytes: func(f *kafkaAdminFixture) int { return f.brokerMessageMaxBytes() - 1 }, @@ -488,23 +442,13 @@ func TestAdjustConfigFallsBackToBrokerMessageMaxBytesWhenTopicConfigMissing(t *t err := adminClient.CreateTopic(detail) 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, + options.BrokerEndpoints = []string{"127.0.0.1:9092"} + options.MaxMessageBytes = test.configuredMaxMessageBytes(adminFixture) + expectedMaxMessageBytes := expectedAdjustedMaxMessageBytes( + options.MaxMessageBytes, + adminFixture.brokerMessageMaxBytes(), ) - 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) @@ -513,13 +457,8 @@ func TestAdjustConfigFallsBackToBrokerMessageMaxBytesWhenTopicConfigMissing(t *t saramaConfig, err := newSaramaConfig(ctx, options) require.NoError(t, err) - require.Equal(t, expectedProducerLimit, options.MaxMessageBytes) - require.Equal( - t, - min(configuredMaxMessageBytes, expectedProducerLimit), - options.MaxBatchedBytes, - ) - require.Equal(t, expectedProducerLimit, saramaConfig.Producer.MaxMessageBytes) + require.Equal(t, expectedMaxMessageBytes, options.MaxMessageBytes) + require.Equal(t, expectedMaxMessageBytes, saramaConfig.Producer.MaxMessageBytes) }) } } @@ -612,7 +551,7 @@ func TestConfigurationCombinations(t *testing.T) { mockTopicMessageMaxBytes, }, { - "new topic broker below user", + "new topic broker overhead below user", "kafka://127.0.0.1:9092/%s?max-message-bytes=%s", []any{"not-created-topic", strconv.Itoa(1024*1024 + 1)}, mockBrokerMessageMaxBytes, @@ -678,7 +617,7 @@ func TestConfigurationCombinations(t *testing.T) { strconv.Itoa(config.DefaultMaxMessageBytes + 1), }, { - "existing topic topic below user", + "existing topic topic overhead below user", "kafka://127.0.0.1:9092/%s?max-message-bytes=%s", []any{defaultMockTopicName, strconv.Itoa(1024*1024 + 1)}, mockBrokerMessageMaxBytes, @@ -729,7 +668,6 @@ func TestConfigurationCombinations(t *testing.T) { options := NewOptions() err = options.Apply(common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) require.Nil(t, err) - configuredMaxMessageBytes := options.MaxMessageBytes topic, ok := a.uriParams[0].(string) require.True(t, ok) @@ -739,15 +677,31 @@ func TestConfigurationCombinations(t *testing.T) { if _, exists := adminFixture.topics[topic]; exists { sourceMaxMessageBytes = adminFixture.topicMaxMessageBytes(topic) } + expectedMaxMessageBytes := expectedAdjustedMaxMessageBytes(options.MaxMessageBytes, sourceMaxMessageBytes) + changefeedID := common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test") err = adjustOptions(changefeedID, adminClient, options, topic) require.Nil(t, err) - require.Equal(t, sourceMaxMessageBytes, options.MaxMessageBytes) - require.Equal( - t, - min(configuredMaxMessageBytes, sourceMaxMessageBytes), - options.MaxBatchedBytes, - ) + require.Equal(t, expectedMaxMessageBytes, options.MaxMessageBytes) + + saramaConfig, err := newSaramaConfig(context.Background(), options) + require.Nil(t, err) + require.Equal(t, expectedMaxMessageBytes, saramaConfig.Producer.MaxMessageBytes) + + encoderConfig := codecCommon.NewConfig(config.ProtocolOpen) + err = encoderConfig.Apply(sinkURI, &config.SinkConfig{ + KafkaConfig: &config.KafkaConfig{ + LargeMessageHandle: config.NewDefaultLargeMessageHandleConfig(), + }, + }) + require.Nil(t, err) + encoderConfig.WithMaxMessageBytes(options.MaxMessageBytes) + + err = encoderConfig.Validate() + require.Nil(t, err) + + // producer's `MaxMessageBytes` = encoder's `MaxMessageBytes`. + require.Equal(t, expectedMaxMessageBytes, encoderConfig.MaxMessageBytes) adminClient.Close() }) @@ -792,7 +746,6 @@ func TestMerge(t *testing.T) { require.Equal(t, int16(5), c.ReplicationFactor) require.Equal(t, "3.1.2", c.Version) require.Equal(t, 1024*1024, c.MaxMessageBytes) - require.Equal(t, 1024*1024, c.MaxBatchedBytes) require.Equal(t, "gzip", c.Compression) require.Equal(t, "test-id", c.ClientID) require.Equal(t, true, c.AutoCreate) @@ -874,7 +827,6 @@ func TestMerge(t *testing.T) { require.Equal(t, int16(5), c.ReplicationFactor) require.Equal(t, "3.1.2", c.Version) require.Equal(t, 1024*1024, c.MaxMessageBytes) - require.Equal(t, 1024*1024, c.MaxBatchedBytes) require.Equal(t, "gzip", c.Compression) require.Equal(t, "test-id", c.ClientID) require.Equal(t, true, c.AutoCreate) diff --git a/pkg/sink/kafka/sarama_config.go b/pkg/sink/kafka/sarama_config.go index 51dbd2384e..8ca31e5025 100644 --- a/pkg/sink/kafka/sarama_config.go +++ b/pkg/sink/kafka/sarama_config.go @@ -62,7 +62,7 @@ func newSaramaConfig(ctx context.Context, o *options) (*sarama.Config, error) { config.Producer.Flush.Bytes = 0 config.Producer.Flush.Messages = 0 config.Producer.Flush.Frequency = time.Duration(0) - config.Producer.Flush.MaxMessages = 0 + config.Producer.Flush.MaxMessages = o.MaxMessages config.Net.MaxOpenRequests = 1 config.Net.DialTimeout = o.DialTimeout diff --git a/pkg/sink/kafka/sarama_config_test.go b/pkg/sink/kafka/sarama_config_test.go index f85d688edc..f8970aa659 100644 --- a/pkg/sink/kafka/sarama_config_test.go +++ b/pkg/sink/kafka/sarama_config_test.go @@ -58,7 +58,6 @@ func TestNewSaramaConfig(t *testing.T) { cfg, err := newSaramaConfig(ctx, options) require.NoError(t, err) require.Equal(t, defaultMaxRetry, cfg.Producer.Retry.Max) - require.Equal(t, options.MaxMessageBytes, cfg.Producer.MaxMessageBytes) options.EnableTLS = true options.Credential = &security.Credential{ diff --git a/pkg/sink/kafka/sarama_factory.go b/pkg/sink/kafka/sarama_factory.go index 8f73ca70b5..5d141f3d1f 100644 --- a/pkg/sink/kafka/sarama_factory.go +++ b/pkg/sink/kafka/sarama_factory.go @@ -68,7 +68,6 @@ func NewSaramaFactory( zap.String("topic", o.Topic), zap.Int32("partitionNum", o.PartitionNum), zap.Int("maxMessageBytes", o.MaxMessageBytes), - zap.Int("maxBatchedBytes", o.MaxBatchedBytes), zap.String("compression", config.Producer.Compression.String()), zap.Int16("requiredAcks", int16(o.RequiredAcks)), zap.Int("maxRetry", o.MaxRetry), diff --git a/tests/integration_tests/_utils/kafka_topic b/tests/integration_tests/_utils/kafka_topic deleted file mode 100755 index 9f9a79a1cd..0000000000 --- a/tests/integration_tests/_utils/kafka_topic +++ /dev/null @@ -1,12 +0,0 @@ -#!/bin/bash - -set -eu - -CUR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) -UTILITY_DIR="$CUR/../../utils/kafka_topic" - -if [ ! -f "$UTILITY_DIR/kafka_topic" ]; then - (cd "$UTILITY_DIR" && GO111MODULE=on go build) -fi - -"$UTILITY_DIR/kafka_topic" "$@" diff --git a/tests/integration_tests/canal_json_claim_check/run.sh b/tests/integration_tests/canal_json_claim_check/run.sh index a354dc589a..10f822c98b 100755 --- a/tests/integration_tests/canal_json_claim_check/run.sh +++ b/tests/integration_tests/canal_json_claim_check/run.sh @@ -18,10 +18,7 @@ function run() { start_tidb_cluster --workdir $WORK_DIR - TOPIC_NAME="canal-json-claim-check-$RANDOM" - CLAIM_CHECK_DIR="/tmp/canal-json-claim-check" - rm -rf "$CLAIM_CHECK_DIR" - kafka_topic --topic "$TOPIC_NAME" --max-message-bytes 1000 + TOPIC_NAME="canal-json-claim-check" # record tso before we create tables to skip the system table DDLs start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1}) @@ -39,10 +36,6 @@ function run() { # sync_diff can't check non-exist table, so we check expected tables are created in downstream first check_table_exists test.finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200 check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml - if ! find "$CLAIM_CHECK_DIR" -type f -print -quit | grep -q .; then - echo "claim-check did not write any file to $CLAIM_CHECK_DIR" - exit 1 - fi cleanup_process $CDC_BINARY } diff --git a/tests/integration_tests/canal_json_handle_key_only/run.sh b/tests/integration_tests/canal_json_handle_key_only/run.sh index 93ae5adfba..372dff24fe 100755 --- a/tests/integration_tests/canal_json_handle_key_only/run.sh +++ b/tests/integration_tests/canal_json_handle_key_only/run.sh @@ -19,7 +19,6 @@ function run() { start_tidb_cluster --workdir $WORK_DIR TOPIC_NAME="canal-json-handle-key-only-$RANDOM" - kafka_topic --topic "$TOPIC_NAME" --max-message-bytes 1000 # record tso before we create tables to skip the system table DDLs start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1}) diff --git a/tests/integration_tests/kafka_big_messages/conf/diff_config.toml b/tests/integration_tests/kafka_big_messages/conf/diff_config.toml index bf73beb595..0082a37028 100644 --- a/tests/integration_tests/kafka_big_messages/conf/diff_config.toml +++ b/tests/integration_tests/kafka_big_messages/conf/diff_config.toml @@ -13,7 +13,7 @@ source-instances = ["mysql1"] target-instance = "tidb0" -target-check-tables = ["database_name.*"] +target-check-tables = ["kafka_big_messages.test"] [data-sources] [data-sources.mysql1] diff --git a/tests/integration_tests/kafka_big_messages/run.sh b/tests/integration_tests/kafka_big_messages/run.sh index b9a4b646fb..0628eaa92e 100755 --- a/tests/integration_tests/kafka_big_messages/run.sh +++ b/tests/integration_tests/kafka_big_messages/run.sh @@ -3,212 +3,53 @@ set -eu CUR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) -source "$CUR/../_utils/test_prepare" +source $CUR/../_utils/test_prepare WORK_DIR=$OUT_DIR/$TEST_NAME CDC_BINARY=cdc.test SINK_TYPE=$1 -STATE_WAIT_TIMEOUT_SECONDS=30 -STATE_CHECK_INTERVAL_SECONDS=1 -TABLE_CHECK_RETRIES=15 -BATCH_LIMIT=262144 -SMALL_TOPIC_LIMIT=524288 -LARGE_TOPIC_LIMIT=2097152 -ROW_BYTES=1048576 -SCHEMA_REGISTRY_URI=http://127.0.0.1:8088 -GENERATOR_DIR=$CUR/../../utils/gen_kafka_big_messages -consumer_pid="" - -function start_schema_registry() { - if curl -o /dev/null -s "$SCHEMA_REGISTRY_URI"; then +function run() { + # test kafka sink only in this case + if [ "$SINK_TYPE" != "kafka" ]; then return fi + rm -rf $WORK_DIR && mkdir -p $WORK_DIR - echo "Starting schema registry..." - ./bin/bin/schema-registry-start -daemon ./bin/etc/schema-registry/schema-registry.properties - local i=0 - while ! curl -o /dev/null -s "$SCHEMA_REGISTRY_URI"; do - i=$((i + 1)) - if [ "$i" -gt 30 ]; then - echo "Failed to start schema registry" - exit 1 - fi - sleep 2 - done - curl -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" --data '{"compatibility": "NONE"}' "$SCHEMA_REGISTRY_URI/config" -} + start_tidb_cluster --workdir $WORK_DIR -function build_message_generator() { - if [ ! -f "$GENERATOR_DIR/gen_kafka_big_messages" ]; then - (cd "$GENERATOR_DIR" && GO111MODULE=on go build) - fi -} + TOPIC_NAME="big-message-test-$RANDOM" -function kafka_sink_uri() { - local topic_name=$1 - local protocol=$2 - local extra_params=$3 - local sink_uri="kafka://127.0.0.1:9092/${topic_name}?protocol=${protocol}&partition-num=1&kafka-version=${KAFKA_VERSION}&max-message-bytes=${BATCH_LIMIT}" - if [ "$extra_params" != "" ]; then - sink_uri="${sink_uri}&${extra_params}" - fi - echo "$sink_uri" -} + # record tso before we create tables to skip the system table DDLs + start_ts=$(run_cdc_cli_tso_query $UP_PD_HOST_1 $UP_PD_PORT_1) -function start_kafka_consumer() { - local work_dir=$1 - local sink_uri=$2 - local schema_registry_uri=$3 - local protocol_case=$4 - local downstream_uri="mysql://root@${DOWN_TIDB_HOST}:${DOWN_TIDB_PORT}/?safe-mode=true&batch-dml-enable=false&enable-ddl-ts=false" - local args=( - --log-file "$work_dir/cdc_kafka_consumer.log" - --log-level debug - --upstream-uri "$sink_uri" - --downstream-uri "$downstream_uri" - ) - if [ "$schema_registry_uri" != "" ]; then - args+=(--schema-registry-uri "$schema_registry_uri") - fi - if [[ "$protocol_case" == simple_* ]]; then - args+=(--upstream-tidb-dsn "root@tcp(${UP_TIDB_HOST}:${UP_TIDB_PORT})/?") - fi + run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY - cdc_kafka_consumer "${args[@]}" >>"$work_dir/cdc_kafka_consumer_stdout.log" 2>&1 & - consumer_pid=$! -} + # Use a max-message-bytes parameter that is larger than the kafka topic max message bytes. + # Test if TiCDC automatically uses the max-message-bytes of the topic. + # See: https://github.com/PingCAP-QE/ci/blob/ddde195ebf4364a0028d53405d1194aa37a4d853/jenkins/pipelines/ci/ticdc/cdc_ghpr_kafka_integration_test.groovy#L178 + # Use a topic that has already been created. + # See: https://github.com/PingCAP-QE/ci/blob/ddde195ebf4364a0028d53405d1194aa37a4d853/jenkins/pipelines/ci/ticdc/cdc_ghpr_kafka_integration_test.groovy#L180 + SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=open-protocol&partition-num=1&kafka-version=${KAFKA_VERSION}&max-message-bytes=12582912" + cdc_cli_changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" + run_kafka_consumer $WORK_DIR "kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=open-protocol&partition-num=1&version=${KAFKA_VERSION}" -function stop_kafka_consumer() { - if [ "$consumer_pid" != "" ]; then - kill -9 "$consumer_pid" 2>/dev/null || true - wait "$consumer_pid" 2>/dev/null || true - consumer_pid="" + echo "Starting generate kafka big messages..." + cd $CUR/../../utils/gen_kafka_big_messages + if [ ! -f ./gen_kafka_big_messages ]; then + GO111MODULE=on go build fi -} - -function wait_changefeed_state() { - local pd_addr=$1 - local changefeed_id=$2 - local expected_state=$3 - local expected_error=$4 - local deadline=$((SECONDS + STATE_WAIT_TIMEOUT_SECONDS)) - - while true; do - if check_changefeed_state "$pd_addr" "$changefeed_id" "$expected_state" "$expected_error" ""; then - return - fi - if [ "$SECONDS" -ge "$deadline" ]; then - echo "changefeed $changefeed_id did not reach state $expected_state within ${STATE_WAIT_TIMEOUT_SECONDS}s" - return 1 - fi - sleep "$STATE_CHECK_INTERVAL_SECONDS" - done -} - -function render_diff_config() { - local work_dir=$1 - local database_name=$2 - local diff_config=$3 - - sed -e "s/database_name/${database_name}/g" \ - -e "s|/tmp/tidb_cdc_test/kafka_big_messages/sync_diff/output|${work_dir}/sync_diff/output|g" \ - "$CUR/conf/diff_config.toml" >"$diff_config" -} - -function run_protocol_case() { - local protocol_case=$1 - local protocol=$2 - local schema_registry_uri=$3 - local extra_params=$4 - local topic_case=${protocol_case//_/-} - local topic_name="big-message-${topic_case}-${RANDOM}" - local changefeed_id="kafka-big-messages-${topic_case}" - local database_name="kafka_big_messages_${protocol_case}" - local work_dir="$WORK_DIR/$protocol_case" - local sql_file="$work_dir/test.sql" - local diff_config="$work_dir/diff_config.toml" - local pd_addr="http://${UP_PD_HOST_1}:${UP_PD_PORT_1}" - local sink_uri - local initial_topic_limit=$SMALL_TOPIC_LIMIT - local expected_error=ErrMessageTooLarge - if [ "$protocol_case" = "async_error" ]; then - initial_topic_limit=$LARGE_TOPIC_LIMIT - expected_error=ErrKafkaSendMessage - fi - - mkdir -p "$work_dir" - render_diff_config "$work_dir" "$database_name" "$diff_config" - kafka_topic --topic "$topic_name" --max-message-bytes "$initial_topic_limit" - local start_ts - start_ts=$(run_cdc_cli_tso_query "$UP_PD_HOST_1" "$UP_PD_PORT_1") - sink_uri=$(kafka_sink_uri "$topic_name" "$protocol" "$extra_params") - - if [ "$schema_registry_uri" != "" ]; then - cdc_cli_changefeed create --start-ts="$start_ts" --sink-uri="$sink_uri" --schema-registry="$schema_registry_uri" -c "$changefeed_id" - else - cdc_cli_changefeed create --start-ts="$start_ts" --sink-uri="$sink_uri" -c "$changefeed_id" - fi - start_kafka_consumer "$work_dir" "$sink_uri" "$schema_registry_uri" "$protocol_case" - wait_changefeed_state "$pd_addr" "$changefeed_id" "normal" "null" - - # Lower the topic limit after the producer has started. The encoder and - # producer still accept the message, then Kafka rejects it asynchronously. - if [ "$protocol_case" = "async_error" ]; then - local ready_database="${database_name}_ready" - run_sql "CREATE DATABASE ${ready_database}; CREATE TABLE ${ready_database}.ready(id INT PRIMARY KEY); INSERT INTO ${ready_database}.ready VALUES (1)" "$UP_TIDB_HOST" "$UP_TIDB_PORT" - ensure "$TABLE_CHECK_RETRIES" "run_sql 'SELECT id FROM ${ready_database}.ready' '$DOWN_TIDB_HOST' '$DOWN_TIDB_PORT' && check_contains 'id: 1'" - kafka_topic --topic "$topic_name" --max-message-bytes "$SMALL_TOPIC_LIMIT" --alter - fi - - "$GENERATOR_DIR/gen_kafka_big_messages" --row-bytes="$ROW_BYTES" --row-count=1 --database-name="$database_name" --table-name=test --sql-file-path="$sql_file" - run_sql_file "$sql_file" "$UP_TIDB_HOST" "$UP_TIDB_PORT" - run_sql "CREATE TABLE ${database_name}.finish_mark(id INT PRIMARY KEY)" "$UP_TIDB_HOST" "$UP_TIDB_PORT" - - wait_changefeed_state "$pd_addr" "$changefeed_id" "warning" "$expected_error" - - # Only increase Kafka's topic limit. TiCDC must recreate the sink, read the - # new limit, and resume without updating, pausing, or resuming the changefeed. - kafka_topic --topic "$topic_name" --max-message-bytes "$LARGE_TOPIC_LIMIT" --alter - wait_changefeed_state "$pd_addr" "$changefeed_id" "normal" "null" - check_table_exists "${database_name}.finish_mark" "$DOWN_TIDB_HOST" "$DOWN_TIDB_PORT" "$TABLE_CHECK_RETRIES" - check_sync_diff "$work_dir" "$diff_config" - - cdc_cli_changefeed remove -c "$changefeed_id" - stop_kafka_consumer -} - -function run() { - # Test Kafka sink only in this case. - if [ "$SINK_TYPE" != "kafka" ]; then - return - fi - - 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" - ) - - rm -rf "$WORK_DIR" && mkdir -p "$WORK_DIR" - start_schema_registry - build_message_generator - start_tidb_cluster --workdir "$WORK_DIR" - run_cdc_server --workdir "$WORK_DIR" --binary "$CDC_BINARY" + # Generate data larger than kafka broker max.message.bytes. We can send this data correctly. + ./gen_kafka_big_messages --row-count=15 --sql-file-path=$CUR/test.sql - local case_entry - for case_entry in "${cases[@]}"; do - local protocol_case protocol schema_registry_uri extra_params - IFS='|' read -r protocol_case protocol schema_registry_uri extra_params <<<"$case_entry" - run_protocol_case "$protocol_case" "$protocol" "$schema_registry_uri" "$extra_params" - done + run_sql_file $CUR/test.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT} + table="kafka_big_messages.test" + check_table_exists $table ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} + check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml - cleanup_process "$CDC_BINARY" + cleanup_process $CDC_BINARY } -trap 'stop_kafka_consumer; stop_test "$WORK_DIR"' EXIT -run "$@" -check_logs "$WORK_DIR" +trap 'stop_test $WORK_DIR' EXIT +run $* +check_logs $WORK_DIR echo "[$(date)] <<<<<< run test case $TEST_NAME success! >>>>>>" diff --git a/tests/integration_tests/kafka_simple_claim_check/data/data.sql b/tests/integration_tests/kafka_simple_claim_check/data/data.sql index c753734f2e..2730d96d71 100644 --- a/tests/integration_tests/kafka_simple_claim_check/data/data.sql +++ b/tests/integration_tests/kafka_simple_claim_check/data/data.sql @@ -1,6 +1,4 @@ use test; --- Keep the encoded row larger than max-message-bytes after Snappy compression, --- so this case exercises claim-check instead of sending the full row to Kafka. insert into t values ( 1, 1, 2, 3, 4, 5, @@ -9,7 +7,7 @@ insert into t values ( 3.1415, 2.7182, 8000, 179394.233, '2020-02-20', '2020-02-20 02:20:20', '2020-02-20 02:20:20', '02:20:20', '2020', '89504E470D0A1A0A', '89504E470D0A1A0A', '89504E470D0A1A0A', '89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A', - x'89504E470D0A1A0A', RANDOM_BYTES(1024), RANDOM_BYTES(1024), RANDOM_BYTES(1024), + x'89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A', '89504E470D0A1A0A', '89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A', 'b', 'b,c', b'1000001', '{ "key1": "value1", @@ -30,7 +28,7 @@ insert into t values ( 3.1415, 2.7182, 8000, 179394.233, '2020-02-20', '2020-02-20 02:20:20', '2020-02-20 02:20:20', '02:20:20', '2020', '89504E470D0A1A0A', '89504E470D0A1A0A', '89504E470D0A1A0A', '89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A', - x'89504E470D0A1A0A', RANDOM_BYTES(1024), RANDOM_BYTES(1024), RANDOM_BYTES(1024), + x'89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A', '89504E470D0A1A0A', '89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A', 'b', 'b,c', b'1000001', '{ "key1": "value1", diff --git a/tests/integration_tests/kafka_simple_claim_check/run.sh b/tests/integration_tests/kafka_simple_claim_check/run.sh index a0ff8de268..c8414aee56 100755 --- a/tests/integration_tests/kafka_simple_claim_check/run.sh +++ b/tests/integration_tests/kafka_simple_claim_check/run.sh @@ -24,8 +24,6 @@ function run() { run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY TOPIC_NAME="kafka-simple-claim-check-$RANDOM" - CLAIM_CHECK_DIR="/tmp/kafka-simple-claim-check" - rm -rf "$CLAIM_CHECK_DIR" # record tso before we create tables to skip the system table DDLs start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1}) @@ -39,7 +37,6 @@ function run() { cdc_cli_changefeed pause -c ${changefeed_id} - kafka_topic --topic "$TOPIC_NAME" --max-message-bytes 2048 --alter SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=simple&max-message-bytes=2048" cdc_cli_changefeed update -c ${changefeed_id} --sink-uri="$SINK_URI" --config="$CUR/conf/changefeed.toml" --no-confirm cdc_cli_changefeed resume -c ${changefeed_id} @@ -51,10 +48,6 @@ function run() { # sync_diff can't check non-exist table, so we check expected tables are created in downstream first check_table_exists test.finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200 check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml - if ! find "$CLAIM_CHECK_DIR" -type f -print -quit | grep -q .; then - echo "claim-check did not write any file to $CLAIM_CHECK_DIR" - exit 1 - fi cleanup_process $CDC_BINARY } diff --git a/tests/integration_tests/kafka_simple_claim_check_avro/data/data.sql b/tests/integration_tests/kafka_simple_claim_check_avro/data/data.sql index c753734f2e..2730d96d71 100644 --- a/tests/integration_tests/kafka_simple_claim_check_avro/data/data.sql +++ b/tests/integration_tests/kafka_simple_claim_check_avro/data/data.sql @@ -1,6 +1,4 @@ use test; --- Keep the encoded row larger than max-message-bytes after Snappy compression, --- so this case exercises claim-check instead of sending the full row to Kafka. insert into t values ( 1, 1, 2, 3, 4, 5, @@ -9,7 +7,7 @@ insert into t values ( 3.1415, 2.7182, 8000, 179394.233, '2020-02-20', '2020-02-20 02:20:20', '2020-02-20 02:20:20', '02:20:20', '2020', '89504E470D0A1A0A', '89504E470D0A1A0A', '89504E470D0A1A0A', '89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A', - x'89504E470D0A1A0A', RANDOM_BYTES(1024), RANDOM_BYTES(1024), RANDOM_BYTES(1024), + x'89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A', '89504E470D0A1A0A', '89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A', 'b', 'b,c', b'1000001', '{ "key1": "value1", @@ -30,7 +28,7 @@ insert into t values ( 3.1415, 2.7182, 8000, 179394.233, '2020-02-20', '2020-02-20 02:20:20', '2020-02-20 02:20:20', '02:20:20', '2020', '89504E470D0A1A0A', '89504E470D0A1A0A', '89504E470D0A1A0A', '89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A', - x'89504E470D0A1A0A', RANDOM_BYTES(1024), RANDOM_BYTES(1024), RANDOM_BYTES(1024), + x'89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A', '89504E470D0A1A0A', '89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A', 'b', 'b,c', b'1000001', '{ "key1": "value1", diff --git a/tests/integration_tests/kafka_simple_claim_check_avro/run.sh b/tests/integration_tests/kafka_simple_claim_check_avro/run.sh index ac3626634e..259d526d59 100755 --- a/tests/integration_tests/kafka_simple_claim_check_avro/run.sh +++ b/tests/integration_tests/kafka_simple_claim_check_avro/run.sh @@ -24,8 +24,6 @@ function run() { run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY TOPIC_NAME="kafka-simple-claim-check-avro-$RANDOM" - CLAIM_CHECK_DIR="/tmp/kafka-simple-avro-claim-check" - rm -rf "$CLAIM_CHECK_DIR" # record tso before we create tables to skip the system table DDLs start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1}) @@ -39,7 +37,6 @@ function run() { cdc_cli_changefeed pause -c ${changefeed_id} - kafka_topic --topic "$TOPIC_NAME" --max-message-bytes 2048 --alter SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=simple&encoding-format=avro&max-message-bytes=2048" cdc_cli_changefeed update -c ${changefeed_id} --sink-uri="$SINK_URI" --config="$CUR/conf/changefeed.toml" --no-confirm cdc_cli_changefeed resume -c ${changefeed_id} @@ -51,10 +48,6 @@ function run() { # sync_diff can't check non-exist table, so we check expected tables are created in downstream first check_table_exists test.finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200 check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml - if ! find "$CLAIM_CHECK_DIR" -type f -print -quit | grep -q .; then - echo "claim-check did not write any file to $CLAIM_CHECK_DIR" - exit 1 - fi cleanup_process $CDC_BINARY } diff --git a/tests/integration_tests/kafka_simple_handle_key_only/run.sh b/tests/integration_tests/kafka_simple_handle_key_only/run.sh index e7b9f38884..32f7ecc6e3 100755 --- a/tests/integration_tests/kafka_simple_handle_key_only/run.sh +++ b/tests/integration_tests/kafka_simple_handle_key_only/run.sh @@ -36,7 +36,6 @@ function run() { cdc_cli_changefeed pause -c ${changefeed_id} - kafka_topic --topic "$TOPIC_NAME" --max-message-bytes 700 --alter SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=simple&max-message-bytes=700" cdc_cli_changefeed update -c ${changefeed_id} --sink-uri="$SINK_URI" --config="$CUR/conf/changefeed.toml" --no-confirm cdc_cli_changefeed resume -c ${changefeed_id} diff --git a/tests/integration_tests/kafka_simple_handle_key_only_avro/run.sh b/tests/integration_tests/kafka_simple_handle_key_only_avro/run.sh index 94ea60d97b..717d3924d3 100755 --- a/tests/integration_tests/kafka_simple_handle_key_only_avro/run.sh +++ b/tests/integration_tests/kafka_simple_handle_key_only_avro/run.sh @@ -36,7 +36,6 @@ function run() { cdc_cli_changefeed pause -c ${changefeed_id} - kafka_topic --topic "$TOPIC_NAME" --max-message-bytes 650 --alter SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=simple&encoding-format=avro&max-message-bytes=650" cdc_cli_changefeed update -c ${changefeed_id} --sink-uri="$SINK_URI" --config="$CUR/conf/changefeed.toml" --no-confirm cdc_cli_changefeed resume -c ${changefeed_id} diff --git a/tests/integration_tests/open_protocol_claim_check/data/data.sql b/tests/integration_tests/open_protocol_claim_check/data/data.sql index dbacc55cf2..14ae2db77e 100644 --- a/tests/integration_tests/open_protocol_claim_check/data/data.sql +++ b/tests/integration_tests/open_protocol_claim_check/data/data.sql @@ -93,7 +93,6 @@ insert into t values ( ); update t set c_float = 3.1415, c_double = 2.7182, c_decimal = 8000, c_decimal_2 = 179394.233 where id = 2; -update t set c_longblob = concat(random_bytes(1024), random_bytes(1024), random_bytes(1024)) where id = 2; begin; diff --git a/tests/integration_tests/open_protocol_claim_check/run.sh b/tests/integration_tests/open_protocol_claim_check/run.sh index 7f9f94e3be..2b262fd3df 100755 --- a/tests/integration_tests/open_protocol_claim_check/run.sh +++ b/tests/integration_tests/open_protocol_claim_check/run.sh @@ -18,10 +18,7 @@ function run() { start_tidb_cluster --workdir $WORK_DIR - TOPIC_NAME="open-protocol-claim-check-$RANDOM" - CLAIM_CHECK_DIR="/tmp/open-protocol-claim-check" - rm -rf "$CLAIM_CHECK_DIR" - kafka_topic --topic "$TOPIC_NAME" --max-message-bytes 2048 + TOPIC_NAME="open-protocol-claim-check" # record tso before we create tables to skip the system table DDLs start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1}) @@ -43,10 +40,6 @@ function run() { # sync_diff can't check non-exist table, so we check expected tables are created in downstream first check_table_exists test.finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200 check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml - if ! find "$CLAIM_CHECK_DIR" -type f -print -quit | grep -q .; then - echo "claim-check did not write any file to $CLAIM_CHECK_DIR" - exit 1 - fi cleanup_process $CDC_BINARY } diff --git a/tests/integration_tests/open_protocol_handle_key_only/data/data.sql b/tests/integration_tests/open_protocol_handle_key_only/data/data.sql index 2c79413baf..2977b9aa12 100644 --- a/tests/integration_tests/open_protocol_handle_key_only/data/data.sql +++ b/tests/integration_tests/open_protocol_handle_key_only/data/data.sql @@ -93,7 +93,6 @@ insert into t values ( ); update t set c_float = 3.1415, c_double = 2.7182, c_decimal = 8000, c_decimal_2 = 179394.233 where id = 2; -update t set c_longblob = concat(random_bytes(1024), random_bytes(1024), random_bytes(1024)) where id = 2; create table finish_mark ( diff --git a/tests/integration_tests/open_protocol_handle_key_only/run.sh b/tests/integration_tests/open_protocol_handle_key_only/run.sh index 01aab001dc..a416274e1d 100755 --- a/tests/integration_tests/open_protocol_handle_key_only/run.sh +++ b/tests/integration_tests/open_protocol_handle_key_only/run.sh @@ -19,7 +19,6 @@ function run() { start_tidb_cluster --workdir $WORK_DIR TOPIC_NAME="open-protocol-handle-key-only-$RANDOM" - kafka_topic --topic "$TOPIC_NAME" --max-message-bytes 2048 # record tso before we create tables to skip the system table DDLs start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1}) diff --git a/tests/utils/kafka_topic/main.go b/tests/utils/kafka_topic/main.go deleted file mode 100644 index 6227492cf9..0000000000 --- a/tests/utils/kafka_topic/main.go +++ /dev/null @@ -1,68 +0,0 @@ -// Copyright 2026 PingCAP, Inc. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// See the License for the specific language governing permissions and -// limitations under the License. - -package main - -import ( - "flag" - "log" - "strconv" - "strings" - - "github.com/IBM/sarama" -) - -func main() { - brokers := flag.String("brokers", "127.0.0.1:9092", "Comma-separated Kafka broker addresses.") - topic := flag.String("topic", "", "Kafka topic name.") - maxMessageBytes := flag.Int("max-message-bytes", 0, "Topic max.message.bytes value.") - alter := flag.Bool("alter", false, "Alter an existing topic instead of creating it.") - flag.Parse() - - if *topic == "" { - log.Fatal("topic must not be empty") - } - if *maxMessageBytes <= 0 { - log.Fatal("max-message-bytes must be greater than zero") - } - - value := strconv.Itoa(*maxMessageBytes) - config := sarama.NewConfig() - config.ClientID = "ticdc-integration-test-kafka-topic" - admin, err := sarama.NewClusterAdmin(strings.Split(*brokers, ","), config) - if err != nil { - log.Fatalf("create Kafka admin client: %v", err) - } - defer func() { - if err := admin.Close(); err != nil { - log.Printf("close Kafka admin client: %v", err) - } - }() - - configEntries := map[string]*string{"max.message.bytes": &value} - if *alter { - if err := admin.AlterConfig(sarama.TopicResource, *topic, configEntries, false); err != nil { - log.Fatalf("alter Kafka topic %s: %v", *topic, err) - } - return - } - - detail := &sarama.TopicDetail{ - NumPartitions: 1, - ReplicationFactor: 1, - ConfigEntries: configEntries, - } - if err := admin.CreateTopic(*topic, detail, false); err != nil { - log.Fatalf("create Kafka topic %s: %v", *topic, err) - } -}