From e417fd3d3cfa51e62bdac0bcb080bae254b2b4a0 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Wed, 29 Jul 2026 19:05:38 +0800 Subject: [PATCH 01/11] adjust the code --- .../sink/topicmanager/kafka_topic_manager.go | 6 ++++- .../topicmanager/kafka_topic_manager_test.go | 4 +--- pkg/sink/kafka/options.go | 9 +++++++ pkg/sink/kafka/options_test.go | 24 +++++++++++++++++++ 4 files changed, 39 insertions(+), 4 deletions(-) diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go index e33929d484..56ed725781 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go @@ -216,11 +216,15 @@ func (m *kafkaTopicManager) waitUntilTopicVisible( ) return err } + detail, ok := meta[topicName] + if !ok { + return errors.ErrKafkaAdminAPI.GenWithStackByArgs("describe-topic", topicName) + } log.Info("topic found", zap.String("keyspace", m.changefeedID.Keyspace()), zap.String("changefeed", m.changefeedID.Name()), zap.String("topic", topicName), - zap.Int32("partitionNumber", meta[topicName].NumPartitions), + zap.Int32("partitionNumber", detail.NumPartitions), zap.Duration("duration", time.Since(start))) return nil }, retry.WithBackoffBaseDelay(500), diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go index 4ee0be636a..3708d77668 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go @@ -248,9 +248,7 @@ func TestCreateTopicWaitsUntilVisible(t *testing.T) { return nil }), adminClient.EXPECT().GetTopicsMeta([]string{topic}, false).Return( - nil, sarama.ErrUnknownTopicOrPartition), - adminClient.EXPECT().GetTopicsMeta([]string{topic}, false).Return( - nil, sarama.ErrUnknownTopicOrPartition), + map[string]kafka.TopicDetail{}, nil), adminClient.EXPECT().GetTopicsMeta([]string{topic}, false).Return( map[string]kafka.TopicDetail{ topic: { diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go index bf76f7f7a5..becd9ac32a 100644 --- a/pkg/sink/kafka/options.go +++ b/pkg/sink/kafka/options.go @@ -294,6 +294,9 @@ func (o *options) Apply(changefeedID common.ChangeFeedID, if err != nil { return errors.WrapError(errors.ErrKafkaInvalidConfig, err) } + if a <= 0 { + return errors.ErrKafkaInvalidConfig.GenWithStack("dial-timeout must be greater than zero") + } o.DialTimeout = a } @@ -302,6 +305,9 @@ func (o *options) Apply(changefeedID common.ChangeFeedID, if err != nil { return errors.WrapError(errors.ErrKafkaInvalidConfig, err) } + if a <= 0 { + return errors.ErrKafkaInvalidConfig.GenWithStack("write-timeout must be greater than zero") + } o.WriteTimeout = a } @@ -310,6 +316,9 @@ func (o *options) Apply(changefeedID common.ChangeFeedID, if err != nil { return errors.WrapError(errors.ErrKafkaInvalidConfig, err) } + if a <= 0 { + return errors.ErrKafkaInvalidConfig.GenWithStack("read-timeout must be greater than zero") + } o.ReadTimeout = a } diff --git a/pkg/sink/kafka/options_test.go b/pkg/sink/kafka/options_test.go index a77b10b319..025df9086b 100644 --- a/pkg/sink/kafka/options_test.go +++ b/pkg/sink/kafka/options_test.go @@ -426,6 +426,30 @@ func TestTimeout(t *testing.T) { require.Equal(t, 2*time.Minute, options.WriteTimeout) } +func TestApplyRejectsNonPositiveTimeout(t *testing.T) { + t.Parallel() + + changefeedID := common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test") + for _, parameter := range []string{"dial-timeout", "read-timeout", "write-timeout"} { + for _, value := range []string{"0s", "-1s"} { + t.Run(parameter+"="+value, func(t *testing.T) { + t.Parallel() + + sinkURI, err := url.Parse( + "kafka://127.0.0.1:9092/kafka-test?" + parameter + "=" + value) + require.NoError(t, err) + + err = NewOptions().Apply( + changefeedID, sinkURI, config.GetDefaultReplicaConfig().Sink) + require.ErrorContains(t, err, parameter+" must be greater than zero") + errCode, ok := errors.RFCCode(err) + require.True(t, ok) + require.Equal(t, errors.ErrKafkaInvalidConfig.RFCCode(), errCode) + }) + } + } +} + func TestAdjustConfigFallsBackToBrokerMessageMaxBytesWhenTopicConfigMissing(t *testing.T) { tests := []struct { name string From 08747bb9cb91c97a6d46a571bd887ee6ee9666bb Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Wed, 29 Jul 2026 19:16:47 +0800 Subject: [PATCH 02/11] add default timeout --- pkg/sink/kafka/options.go | 8 +++++--- pkg/sink/kafka/options_test.go | 6 +++--- 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go index becd9ac32a..1592adac9f 100644 --- a/pkg/sink/kafka/options.go +++ b/pkg/sink/kafka/options.go @@ -38,6 +38,8 @@ const ( defaultPartitionNum = 3 // defaultMaxRetry is the default retry budget for Kafka producers. defaultMaxRetry = 5 + // defaultTimeout is the default timeout for Kafka connections. + defaultTimeout = 10 * time.Second ) const ( @@ -186,9 +188,9 @@ func NewOptions() *options { InsecureSkipVerify: false, SASL: &security.SASL{}, AutoCreate: true, - DialTimeout: 10 * time.Second, - WriteTimeout: 10 * time.Second, - ReadTimeout: 10 * time.Second, + DialTimeout: defaultTimeout, + WriteTimeout: defaultTimeout, + ReadTimeout: defaultTimeout, } } diff --git a/pkg/sink/kafka/options_test.go b/pkg/sink/kafka/options_test.go index 025df9086b..484ca52f97 100644 --- a/pkg/sink/kafka/options_test.go +++ b/pkg/sink/kafka/options_test.go @@ -409,9 +409,9 @@ func TestClientID(t *testing.T) { func TestTimeout(t *testing.T) { options := NewOptions() - require.Equal(t, 10*time.Second, options.DialTimeout) - require.Equal(t, 10*time.Second, options.ReadTimeout) - require.Equal(t, 10*time.Second, options.WriteTimeout) + require.Equal(t, defaultTimeout, options.DialTimeout) + require.Equal(t, defaultTimeout, options.ReadTimeout) + require.Equal(t, defaultTimeout, options.WriteTimeout) uri := "kafka://127.0.0.1:9092/kafka-test?dial-timeout=5s&read-timeout=1000ms" + "&write-timeout=2m" From 03b4643c1af271691b1a6b67c7293643f737d102 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Thu, 30 Jul 2026 10:32:15 +0800 Subject: [PATCH 03/11] remove useless log --- .../sink/topicmanager/kafka_topic_manager.go | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go index 56ed725781..defd158d59 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go @@ -216,16 +216,10 @@ func (m *kafkaTopicManager) waitUntilTopicVisible( ) return err } - detail, ok := meta[topicName] + _, ok := meta[topicName] if !ok { return errors.ErrKafkaAdminAPI.GenWithStackByArgs("describe-topic", topicName) } - log.Info("topic found", - zap.String("keyspace", m.changefeedID.Keyspace()), - zap.String("changefeed", m.changefeedID.Name()), - zap.String("topic", topicName), - zap.Int32("partitionNumber", detail.NumPartitions), - zap.Duration("duration", time.Since(start))) return nil }, retry.WithBackoffBaseDelay(500), retry.WithBackoffMaxDelay(1000), From 0140249cb1a7de98774778ff5cec43a806c3f6a4 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Thu, 30 Jul 2026 11:31:00 +0800 Subject: [PATCH 04/11] remove useless logs --- downstreamadapter/sink/kafka/sink.go | 24 ----- .../sink/topicmanager/kafka_topic_manager.go | 25 ++--- pkg/sink/kafka/admin.go | 15 --- pkg/sink/kafka/sarama_async_producer.go | 3 - tests/integration_tests/log_redaction/run.sh | 91 ------------------- tests/integration_tests/run_light_it_in_ci.sh | 2 +- 6 files changed, 7 insertions(+), 153 deletions(-) diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index 13c4d539b6..b54eedfd29 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -311,9 +311,6 @@ func (s *sink) calculateKeyPartitions(ctx context.Context) error { default: event, ok := s.eventChan.Get() if !ok { - log.Info("kafka sink event channel closed", - zap.String("keyspace", s.changefeedID.Keyspace()), - zap.String("changefeed", s.changefeedID.Name())) return nil } schema := event.TableInfo.GetSchemaName() @@ -343,9 +340,6 @@ func (s *sink) nonBatchEncodeRun(ctx context.Context) error { default: event, ok := s.rowChan.Get() if !ok { - log.Info("kafka sink event channel closed", - zap.String("keyspace", s.changefeedID.Keyspace()), - zap.String("changefeed", s.changefeedID.Name())) return nil } if err := s.comp.encoderGroup.AddEvents(ctx, event.Key, &event.RowEvent); err != nil { @@ -368,10 +362,6 @@ func (s *sink) batchEncodeRun(ctx context.Context) error { start := time.Now() msgs, err := s.batch(ctx, msgsBuf) if err != nil { - log.Error("kafka sink batch dml events failed", - zap.String("keyspace", s.changefeedID.Keyspace()), - zap.String("changefeed", s.changefeedID.Name()), - zap.Error(err)) return err } if len(msgs) == 0 { @@ -401,9 +391,6 @@ func (s *sink) batch(ctx context.Context, buffer []*commonEvent.MQRowEvent) ([]* default: msgs, ok := s.rowChan.GetMultipleNoGroup(buffer) if !ok { - log.Info("kafka sink event channel closed", - zap.String("keyspace", s.changefeedID.Keyspace()), - zap.String("changefeed", s.changefeedID.Name())) return nil, nil } buffer = buffer[:0] @@ -435,9 +422,6 @@ func (s *sink) sendMessages(ctx context.Context) error { return context.Cause(ctx) case future, ok := <-outCh: if !ok { - log.Info("kafka sink encoder's output channel closed", - zap.String("keyspace", s.changefeedID.Keyspace()), - zap.String("changefeed", s.changefeedID.Name())) return nil } if err = future.Ready(ctx); err != nil { @@ -447,7 +431,6 @@ func (s *sink) sendMessages(ctx context.Context) error { start := time.Now() if err = s.statistics.RecordBatchExecution(func() (int, int64, error) { message.SetPartitionKey(future.Key.PartitionKey) - log.Debug("send message to kafka", zap.String("messageKey", util.RedactBytes(message.Key)), zap.String("messageValue", util.RedactBytes(message.Value))) if err = s.dmlProducer.AsyncSend( ctx, future.Key.Topic, @@ -476,9 +459,6 @@ func (s *sink) sendDDLEvent(event *commonEvent.DDLEvent) error { return err } if message == nil { - log.Info("Skip ddl event", zap.Uint64("startTs", event.GetStartTs()), zap.Uint64("commitTs", e.GetCommitTs()), - zap.String("query", e.Query), - zap.Stringer("changefeed", s.changefeedID)) continue } codecCommon.SetDDLMessageLogInfo(message, e) @@ -596,10 +576,6 @@ func (s *sink) SetTableSchemaStore(tableSchemaStore *commonEvent.TableSchemaStor func (s *sink) getAllTableNames(ts uint64) []*commonEvent.SchemaTableName { if s.tableSchemaStore == nil { - log.Warn("kafka sink table schema store is not set", - zap.String("keyspace", s.changefeedID.Keyspace()), - zap.String("changefeed", s.changefeedID.Name()), - zap.Uint64("ts", ts)) return nil } return s.tableSchemaStore.GetAllTableNames(ts, true) diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go index defd158d59..313f415be7 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go @@ -114,10 +114,6 @@ func (m *kafkaTopicManager) backgroundRefreshMeta(ctx context.Context) { for { select { case <-ctx.Done(): - log.Info("Background refresh Kafka metadata goroutine exit.", - zap.String("keyspace", m.changefeedID.Keyspace()), - zap.String("changefeed", m.changefeedID.Name()), - ) return case <-ticker.C: // We ignore the error here, because the error may be caused by the @@ -147,13 +143,6 @@ func (m *kafkaTopicManager) tryUpdatePartitionsAndLogging(topic string, partitio } } else { m.topics.Store(topic, partitions) - log.Info( - "store topic partition number", - zap.String("keyspace", m.changefeedID.Keyspace()), - zap.String("changefeed", m.changefeedID.Name()), - zap.String("topic", topic), - zap.Int32("partitionNumber", partitions), - ) } } @@ -201,19 +190,13 @@ func (m *kafkaTopicManager) waitUntilTopicVisible( ctx context.Context, topicName string, ) error { + start := time.Now() topics := []string{topicName} err := retry.Do(ctx, func() error { - start := time.Now() // ignoreTopicError is set to false since we just create the topic, // make sure the topic is visible. meta, err := m.admin.GetTopicsMeta(topics, false) if err != nil { - log.Warn("topic not found, retry it", - zap.String("keyspace", m.changefeedID.Keyspace()), - zap.String("changefeed", m.changefeedID.Name()), - zap.Error(err), - zap.Duration("duration", time.Since(start)), - ) return err } _, ok := meta[topicName] @@ -225,7 +208,11 @@ func (m *kafkaTopicManager) waitUntilTopicVisible( retry.WithBackoffMaxDelay(1000), retry.WithMaxTries(6), ) - + log.Warn("kafka topic not found for too long", + zap.String("keyspace", m.changefeedID.Keyspace()), + zap.String("changefeed", m.changefeedID.Name()), + zap.Duration("duration", time.Since(start)), + zap.Error(err)) return err } diff --git a/pkg/sink/kafka/admin.go b/pkg/sink/kafka/admin.go index 13833516ba..22839e7314 100644 --- a/pkg/sink/kafka/admin.go +++ b/pkg/sink/kafka/admin.go @@ -81,11 +81,6 @@ func (a *saramaAdminClient) GetBrokerConfig(configName string) (string, bool, er return entry.Value, true, nil } } - - log.Warn("Kafka config item not found", - zap.String("keyspace", a.changefeed.Keyspace()), - zap.String("changefeed", a.changefeed.Name()), - zap.String("configName", configName)) return "", false, nil } @@ -104,19 +99,9 @@ func (a *saramaAdminClient) GetTopicConfig(topicName string, configName string) // 2. Kop returns all configs. for _, entry := range configEntries { if entry.Name == configName { - log.Info("Kafka config item found", - zap.String("keyspace", a.changefeed.Keyspace()), - zap.String("changefeed", a.changefeed.Name()), - zap.String("configName", configName), - zap.String("configValue", entry.Value)) return entry.Value, true, nil } } - - log.Warn("Kafka config item not found", - zap.String("keyspace", a.changefeed.Keyspace()), - zap.String("changefeed", a.changefeed.Name()), - zap.String("configName", configName)) return "", false, nil } diff --git a/pkg/sink/kafka/sarama_async_producer.go b/pkg/sink/kafka/sarama_async_producer.go index f0c0f6b5d1..cc26e23531 100644 --- a/pkg/sink/kafka/sarama_async_producer.go +++ b/pkg/sink/kafka/sarama_async_producer.go @@ -97,9 +97,6 @@ func (p *saramaAsyncProducer) AsyncRunCallback( for { select { case <-ctx.Done(): - log.Info("async producer exit since context is done", - zap.String("keyspace", p.changefeedID.Keyspace()), - zap.String("changefeed", p.changefeedID.Name())) return context.Cause(ctx) case ack := <-p.producer.Successes(): if ack != nil { diff --git a/tests/integration_tests/log_redaction/run.sh b/tests/integration_tests/log_redaction/run.sh index b99fe5da90..71b7e1e656 100755 --- a/tests/integration_tests/log_redaction/run.sh +++ b/tests/integration_tests/log_redaction/run.sh @@ -372,97 +372,6 @@ function run() { echo "[$(date)] ✓ MySQL sink: All redaction modes validated" fi - # ========================================================================== - # Test 4b: Kafka sink validation (tests Kafka-specific redaction) - # ========================================================================== - if [ "$SINK_TYPE" = "kafka" ]; then - echo "" - echo "=== Test 4b: Kafka sink redaction validation ===" - - # Kafka sink logs message key/value at DEBUG level - # Log message: "send message to kafka" with messageKey and messageValue fields - - # Test ON mode with Kafka sink (most important - full redaction) - echo " [4b-1] ON mode with Kafka sink:" - run_sql "DROP DATABASE IF EXISTS log_redaction_test;" - run_sql "CREATE DATABASE log_redaction_test;" - - KAFKA_TOPIC="log-redaction-test-$RANDOM" - KAFKA_SINK_URI="kafka://127.0.0.1:9092/$KAFKA_TOPIC?protocol=open-protocol" - - run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY --redact-info-log on --logsuffix "_on_kafka" - - cdc_cli_changefeed create --sink-uri="$KAFKA_SINK_URI" --changefeed-id="kafka-on-test" --config=$CUR/conf/changefeed.toml - - run_sql_file $CUR/data/test.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT} - - echo " Waiting for Kafka sink to process events..." - wait_for_log_content "$WORK_DIR/cdc_on_kafka.log" "send message to kafka" "Kafka message logs" 30 - - echo " [Validation] ON mode with Kafka sink:" - echo "" - - # Capture Kafka logs once for all validations - captured_logs=$(grep "send message to kafka" "$WORK_DIR/cdc_on_kafka.log" 2>/dev/null || echo "") - log_raw_content "Kafka message logs (ON mode)" "$captured_logs" - - # STRICT POSITIVE VALIDATION: messageKey and messageValue must show redacted format - echo " [1/2] Verifying Kafka logs show redacted '?' placeholder:" - require_log_pattern "$WORK_DIR/cdc_on_kafka.log" \ - "send message to kafka.*messageKey.*\?.*messageValue.*\?" \ - "Kafka messageKey and messageValue redacted to '?'" \ - "ON mode should redact both messageKey and messageValue" - - # STRICT NEGATIVE VALIDATION: No sensitive data should leak in Kafka logs - echo " [2/2] Verifying NO sensitive data leaks in Kafka logs:" - sensitive_patterns=("Password1!" "SecretPass1!" "user1@example.com" "4532-1000-1000") - for pattern in "${sensitive_patterns[@]}"; do - require_no_log_pattern "$WORK_DIR/cdc_on_kafka.log" \ - "$pattern" \ - "No leak of sensitive value in Kafka logs: $pattern" - done - - captured_logs="" - cleanup_process $CDC_BINARY - - # Test MARKER mode with Kafka sink - echo "" - echo " [4b-2] MARKER mode with Kafka sink:" - run_sql "DROP DATABASE IF EXISTS log_redaction_test;" - run_sql "CREATE DATABASE log_redaction_test;" - - KAFKA_TOPIC="log-redaction-marker-$RANDOM" - KAFKA_SINK_URI="kafka://127.0.0.1:9092/$KAFKA_TOPIC?protocol=open-protocol" - - run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY --redact-info-log marker --logsuffix "_marker_kafka" - - cdc_cli_changefeed create --sink-uri="$KAFKA_SINK_URI" --changefeed-id="kafka-marker-test" --config=$CUR/conf/changefeed.toml - - run_sql_file $CUR/data/test.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT} - - echo " Waiting for Kafka sink to process events..." - wait_for_log_content "$WORK_DIR/cdc_marker_kafka.log" "send message to kafka" "Kafka message logs" 30 - - echo " [Validation] MARKER mode with Kafka sink:" - echo "" - - # Capture Kafka logs once for all validations - captured_logs=$(grep "send message to kafka" "$WORK_DIR/cdc_marker_kafka.log" 2>/dev/null || echo "") - log_raw_content "Kafka message logs (MARKER mode)" "$captured_logs" - - # STRICT POSITIVE VALIDATION: messageKey and messageValue must have markers - echo " [1/1] Verifying Kafka logs have ‹› markers:" - require_log_pattern "$WORK_DIR/cdc_marker_kafka.log" \ - "send message to kafka.*‹" \ - "Kafka message values wrapped with ‹› markers" \ - "MARKER mode should wrap Kafka message data with ‹› markers" - - captured_logs="" - cleanup_process $CDC_BINARY - - echo "[$(date)] ✓ Kafka sink: Redaction modes validated" - fi - # ========================================================================== # Test 5: API mode switching # ========================================================================== diff --git a/tests/integration_tests/run_light_it_in_ci.sh b/tests/integration_tests/run_light_it_in_ci.sh index 995583e1f2..2bff0a327a 100755 --- a/tests/integration_tests/run_light_it_in_ci.sh +++ b/tests/integration_tests/run_light_it_in_ci.sh @@ -89,7 +89,7 @@ kafka_groups=( # G08 'capture_session_done_during_task fail_over_ddl_I table_route' # G09 - 'cdc_server_tips ddl_sequence log_redaction fail_over_ddl_J' + 'cdc_server_tips ddl_sequence fail_over_ddl_J' # G10 'changefeed_error batch_add_table fail_over_ddl_K' # G11 From 33f9c5c2fc487cef9c92521ae048e183ff669186 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Thu, 30 Jul 2026 12:14:58 +0800 Subject: [PATCH 05/11] adjust kafka version detection log --- pkg/sink/kafka/sarama_config.go | 70 ++++++++++++++-------------- pkg/sink/kafka/sarama_config_test.go | 57 ++++++++++++++++++++++ 2 files changed, 92 insertions(+), 35 deletions(-) diff --git a/pkg/sink/kafka/sarama_config.go b/pkg/sink/kafka/sarama_config.go index 6f2d56456e..8c194ee224 100644 --- a/pkg/sink/kafka/sarama_config.go +++ b/pkg/sink/kafka/sarama_config.go @@ -120,27 +120,27 @@ func newSaramaConfig(ctx context.Context, o *options) (*sarama.Config, error) { return nil, err } - kafkaVersion, err := getKafkaVersion(config, o) + err = completeSaramaKafkaVersion(config, o) if err != nil { - log.Warn("Can't get Kafka version by broker. ticdc will use default version", - zap.String("defaultVersion", kafkaVersion.String())) + return nil, err } - config.Version = kafkaVersion + return config, nil +} - if o.IsAssignedVersion { - version, err := sarama.ParseKafkaVersion(o.Version) - if err != nil { - return nil, errors.WrapError(errors.ErrKafkaInvalidConfig, err) - } - config.Version = version - if !version.IsAtLeast(maxKafkaVersion) && version.String() != kafkaVersion.String() { - log.Warn("The Kafka version you assigned may not be correct. "+ - "Please assign a version equal to or less than the specified version", - zap.String("assignedVersion", version.String()), - zap.String("desiredVersion", kafkaVersion.String())) - } +func completeSaramaKafkaVersion(config *sarama.Config, o *options) error { + detectedVersion, err := detectKafkaVersion(config, o) + if err != nil { + log.Warn("kafka version detection failed, using fallback version", + zap.Strings("brokers", o.BrokerEndpoints), + zap.String("fallbackVersion", detectedVersion.String()), + zap.Error(err)) } - return config, nil + kafkaVersion, err := selectKafkaVersion(detectedVersion, o) + if err != nil { + return err + } + config.Version = kafkaVersion + return nil } func completeSaramaSASLConfig(ctx context.Context, config *sarama.Config, o *options) error { @@ -186,7 +186,7 @@ func completeSaramaSASLConfig(ctx context.Context, config *sarama.Config, o *opt return nil } -func getKafkaVersion(config *sarama.Config, o *options) (sarama.KafkaVersion, error) { +func detectKafkaVersion(config *sarama.Config, o *options) (sarama.KafkaVersion, error) { addrs := o.BrokerEndpoints if len(addrs) > 1 { // Shuffle the list of addresses to randomize the order in which @@ -208,25 +208,27 @@ func getKafkaVersion(config *sarama.Config, o *options) (sarama.KafkaVersion, er } } if err != nil { - log.Warn("kafka sink use the default kafka version since cannot find it from the brokers", - zap.String("defaultVersion", defaultKafkaVersion.String())) targetVersion = defaultKafkaVersion } + return targetVersion, err +} - if o.IsAssignedVersion { - assignedVersion, err := sarama.ParseKafkaVersion(o.Version) - if err != nil { - return assignedVersion, errors.WrapError(errors.ErrKafkaInvalidConfig, err) - } - if !assignedVersion.IsAtLeast(maxKafkaVersion) && assignedVersion.String() != targetVersion.String() { - log.Warn("The Kafka version you assigned may not be correct. "+ - "Please assign a version equal to or less than the specified version", - zap.String("assignedVersion", assignedVersion.String()), - zap.String("desiredVersion", targetVersion.String())) - } - targetVersion = assignedVersion +func selectKafkaVersion(detectedVersion sarama.KafkaVersion, o *options) (sarama.KafkaVersion, error) { + if !o.IsAssignedVersion { + return detectedVersion, nil + } + assignedVersion, err := sarama.ParseKafkaVersion(o.Version) + if err != nil { + return assignedVersion, errors.WrapError(errors.ErrKafkaInvalidConfig, err) + } + if !assignedVersion.IsAtLeast(maxKafkaVersion) && + assignedVersion.String() != detectedVersion.String() { + log.Warn("The Kafka version you assigned may not be correct. "+ + "Please assign a version equal to or less than the specified version", + zap.String("assignedVersion", assignedVersion.String()), + zap.String("desiredVersion", detectedVersion.String())) } - return targetVersion, nil + return assignedVersion, nil } func getKafkaVersionFromBroker(config *sarama.Config, requestVersion int16, addr string) (sarama.KafkaVersion, error) { @@ -237,12 +239,10 @@ func getKafkaVersionFromBroker(config *sarama.Config, requestVersion int16, addr _ = broker.Close() }() if err != nil { - log.Warn("Kafka fail to open broker", zap.String("addr", addr), zap.Error(err)) return KafkaVersion, err } apiResponse, err := broker.ApiVersions(&sarama.ApiVersionsRequest{Version: requestVersion}) if err != nil { - log.Warn("Kafka fail to get ApiVersions", zap.String("addr", addr), zap.Error(err)) return KafkaVersion, err } // ApiKey method diff --git a/pkg/sink/kafka/sarama_config_test.go b/pkg/sink/kafka/sarama_config_test.go index 3e1520b1e8..65ebb088ad 100644 --- a/pkg/sink/kafka/sarama_config_test.go +++ b/pkg/sink/kafka/sarama_config_test.go @@ -86,6 +86,63 @@ func TestNewSaramaConfig(t *testing.T) { require.Equal(t, sarama.SASLMechanism("SCRAM-SHA-256"), cfg.Net.SASL.Mechanism) } +func TestSelectKafkaVersion(t *testing.T) { + tests := []struct { + name string + detectedVersion sarama.KafkaVersion + assignedVersion string + expectedVersion sarama.KafkaVersion + expectedErr error + }{ + { + name: "use detected version", + detectedVersion: sarama.V2_4_0_0, + expectedVersion: sarama.V2_4_0_0, + }, + { + name: "use fallback version", + detectedVersion: defaultKafkaVersion, + expectedVersion: defaultKafkaVersion, + }, + { + name: "assigned version overrides detected version", + detectedVersion: sarama.V2_4_0_0, + assignedVersion: "2.6.0", + expectedVersion: sarama.V2_6_0_0, + }, + { + name: "assigned version overrides fallback version", + detectedVersion: defaultKafkaVersion, + assignedVersion: "2.6.0", + expectedVersion: sarama.V2_6_0_0, + }, + { + name: "reject invalid assigned version", + detectedVersion: sarama.V2_4_0_0, + assignedVersion: "invalid", + expectedErr: errors.ErrKafkaInvalidConfig, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + options := NewOptions() + if test.assignedVersion != "" { + options.IsAssignedVersion = true + options.Version = test.assignedVersion + } + + version, err := selectKafkaVersion(test.detectedVersion, options) + if test.expectedErr != nil { + require.ErrorIs(t, err, test.expectedErr) + return + } + require.NoError(t, err) + require.Equal(t, test.expectedVersion, version) + }) + } +} + func TestNewSaramaConfigInvalidOAuthTokenURL(t *testing.T) { options := NewOptions() options.SASL = &security.SASL{ From 80f00d8ccf2cf7d2b489ed71763977350365bec9 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Thu, 30 Jul 2026 12:26:29 +0800 Subject: [PATCH 06/11] unify the log format --- downstreamadapter/sink/kafka/sink.go | 10 ++---- .../sink/topicmanager/kafka_topic_manager.go | 23 +++++++------ pkg/sink/kafka/admin.go | 6 ++-- pkg/sink/kafka/claimcheck/claim_check.go | 2 +- pkg/sink/kafka/options.go | 33 +++++-------------- pkg/sink/kafka/sarama_async_producer.go | 12 +++---- pkg/sink/kafka/sarama_config.go | 8 ++--- pkg/sink/kafka/sarama_factory.go | 6 ++-- pkg/sink/kafka/sarama_sync_producer.go | 12 +++---- 9 files changed, 45 insertions(+), 67 deletions(-) diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index b54eedfd29..1e04dc67f2 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -257,7 +257,7 @@ func (s *sink) WriteBlockEvent(event commonEvent.BlockEvent) error { case *commonEvent.DDLEvent: err = s.sendDDLEvent(v) default: - log.Error("kafka sink doesn't support this type of block event", + log.Error("unsupported kafka sink block event type", zap.String("namespace", s.changefeedID.Keyspace()), zap.String("changefeed", s.changefeedID.Name()), zap.String("eventType", commonEvent.TypeToString(event.GetType()))) @@ -436,10 +436,6 @@ func (s *sink) sendMessages(ctx context.Context) error { future.Key.Topic, future.Key.Partition, message); err != nil { - log.Error("kafka sink send message failed", - zap.String("keyspace", s.changefeedID.Keyspace()), - zap.String("changefeed", s.changefeedID.Name()), - zap.Error(err)) return 0, 0, err } return message.GetRowsCount(), int64(message.Length()), nil @@ -485,7 +481,7 @@ func (s *sink) sendDDLEvent(event *commonEvent.DDLEvent) error { return err } } - log.Info("kafka sink send DDL event", + log.Info("kafka ddl event sent", zap.String("keyspace", s.changefeedID.Keyspace()), zap.String("changefeed", s.changefeedID.Name()), zap.Any("startTs", event.GetStartTs()), zap.Any("commitTs", event.GetCommitTs()), zap.Any("event", event.GetDDLQuery()), zap.String("schema", event.GetSchemaName()), zap.String("table", event.GetTableName())) @@ -521,7 +517,7 @@ func (s *sink) sendCheckpoint(ctx context.Context) error { return context.Cause(ctx) case ts, ok := <-s.checkpointChan: if !ok { - log.Warn("kafka sink checkpoint channel closed", + log.Warn("kafka checkpoint channel closed", zap.String("keyspace", s.changefeedID.Keyspace()), zap.String("changefeed", s.changefeedID.Name())) return nil diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go index 313f415be7..b31e29610f 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go @@ -133,7 +133,7 @@ func (m *kafkaTopicManager) tryUpdatePartitionsAndLogging(topic string, partitio if oldPartitions.(int32) != partitions { m.topics.Store(topic, partitions) log.Info( - "update topic partition number", + "kafka topic partition count changed", zap.String("keyspace", m.changefeedID.Keyspace()), zap.String("changefeed", m.changefeedID.Name()), zap.String("topic", topic), @@ -161,7 +161,7 @@ func (m *kafkaTopicManager) fetchAllTopicsPartitionsNum() (map[string]int32, err numPartitions, err := m.admin.GetTopicsPartitionsNum(topics) if err != nil { log.Warn( - "Kafka admin client describe topics failed", + "kafka topic metadata refresh failed", zap.String("keyspace", m.changefeedID.Keyspace()), zap.String("changefeed", m.changefeedID.Name()), zap.Duration("duration", time.Since(start)), @@ -208,11 +208,14 @@ func (m *kafkaTopicManager) waitUntilTopicVisible( retry.WithBackoffMaxDelay(1000), retry.WithMaxTries(6), ) - log.Warn("kafka topic not found for too long", - zap.String("keyspace", m.changefeedID.Keyspace()), - zap.String("changefeed", m.changefeedID.Name()), - zap.Duration("duration", time.Since(start)), - zap.Error(err)) + if err != nil { + log.Warn("kafka topic metadata refresh failed", + zap.String("keyspace", m.changefeedID.Keyspace()), + zap.String("changefeed", m.changefeedID.Name()), + zap.String("topic", topicName), + zap.Duration("duration", time.Since(start)), + zap.Error(err)) + } return err } @@ -238,7 +241,7 @@ func (m *kafkaTopicManager) createTopic( }, false) if err != nil { log.Error( - "Kafka admin client create the topic failed", + "kafka topic creation failed", zap.String("keyspace", m.changefeedID.Keyspace()), zap.String("changefeed", m.changefeedID.Name()), zap.String("topic", topicName), @@ -251,7 +254,7 @@ func (m *kafkaTopicManager) createTopic( } log.Info( - "Kafka admin client create the topic success", + "kafka topic created", zap.String("keyspace", m.changefeedID.Keyspace()), zap.String("changefeed", m.changefeedID.Name()), zap.String("topic", topicName), @@ -323,7 +326,7 @@ func (m *kafkaTopicManager) tryStoreTopicMeta( } func (m *kafkaTopicManager) useConfiguredPartitionNum(topicName string, cause error) int32 { - log.Warn("skip Kafka topic creation because topic authorization failed", + log.Warn("kafka topic creation skipped due to authorization failure", zap.String("keyspace", m.changefeedID.Keyspace()), zap.String("changefeed", m.changefeedID.Name()), zap.String("topic", topicName), diff --git a/pkg/sink/kafka/admin.go b/pkg/sink/kafka/admin.go index 22839e7314..98e3b154d1 100644 --- a/pkg/sink/kafka/admin.go +++ b/pkg/sink/kafka/admin.go @@ -121,7 +121,7 @@ func (a *saramaAdminClient) GetTopicsMeta(topics []string, ignoreTopicError bool if !ignoreTopicError { return nil, errors.WrapError(errors.ErrKafkaAdminAPI, meta.Err, "describe-topic", meta.Name) } - log.Warn("fetch topic meta failed", + log.Warn("kafka topic metadata refresh failed", zap.String("keyspace", a.changefeed.Keyspace()), zap.String("changefeed", a.changefeed.Name()), zap.String("topic", meta.Name), @@ -175,7 +175,7 @@ func (a *saramaAdminClient) Close() { // only when admin is unexpectedly nil. if a.admin != nil { if err := a.admin.Close(); err != nil { - log.Warn("close admin client meet error", + log.Warn("kafka admin client close failed", zap.String("keyspace", a.changefeed.Keyspace()), zap.String("changefeed", a.changefeed.Name()), zap.Error(err)) @@ -184,7 +184,7 @@ func (a *saramaAdminClient) Close() { } if a.client != nil { if err := a.client.Close(); err != nil { - log.Warn("close kafka client meet error", + log.Warn("kafka client close failed", zap.String("keyspace", a.changefeed.Keyspace()), zap.String("changefeed", a.changefeed.Name()), zap.Error(err)) diff --git a/pkg/sink/kafka/claimcheck/claim_check.go b/pkg/sink/kafka/claimcheck/claim_check.go index 5ed01ec016..21a53b58c3 100644 --- a/pkg/sink/kafka/claimcheck/claim_check.go +++ b/pkg/sink/kafka/claimcheck/claim_check.go @@ -52,7 +52,7 @@ func New(ctx context.Context, config *config.LargeMessageHandleConfig, changefee start := time.Now() externalStorage, err := util.GetExternalStorageWithDefaultTimeout(ctx, config.ClaimCheckStorageURI) if err != nil { - log.Error("create external storage failed", + log.Error("external storage creation failed", zap.String("keyspace", changefeedID.Keyspace()), zap.String("changefeed", changefeedID.Name()), zap.String("storageURI", util.MaskSensitiveDataInURI(config.ClaimCheckStorageURI)), diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go index 1592adac9f..523bcf73ea 100644 --- a/pkg/sink/kafka/options.go +++ b/pkg/sink/kafka/options.go @@ -199,15 +199,14 @@ func (o *options) setPartitionNum(changefeedID common.ChangeFeedID, realPartitio // user does not specify the `partition-num` in the sink-uri if o.PartitionNum == 0 { o.PartitionNum = realPartitionCount - log.Info("partitionNum is not set, set by topic's partition-num", + log.Info("kafka partition count set from topic metadata", zap.String("namespace", changefeedID.Keyspace()), zap.String("changefeed", changefeedID.Name()), zap.Int32("partitionNum", realPartitionCount)) return nil } if o.PartitionNum < realPartitionCount { - log.Warn("number of partition specified in sink-uri is less than that of the actual topic. "+ - "Some partitions will not have messages dispatched to", + log.Warn("configured kafka partition count is lower than topic partition count", zap.String("namespace", changefeedID.Keyspace()), zap.String("changefeed", changefeedID.Name()), zap.Int32("sinkUriPartitions", o.PartitionNum), zap.Int32("topicPartitions", realPartitionCount)) return nil @@ -504,7 +503,6 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf // BASE64 decode the client secret decodedClientSecret, err := base64.StdEncoding.DecodeString(clientSecret) if err != nil { - log.Error("OAuth2 client secret is not base64 encoded", zap.Error(err)) return errors.ErrKafkaInvalidConfig.GenWithStack("OAuth2 client secret is not base64 encoded") } o.SASL.OAuth2.ClientSecret = string(decodedClientSecret) @@ -574,14 +572,14 @@ func (c *AutoCreateTopicConfig) ValidateReplicationFactor(admin ClusterAdminClie raw, found, err := admin.GetBrokerConfig(MinInsyncReplicasConfigName) if err != nil { - log.Warn("cannot get Kafka broker configuration, assume replication factor is valid", + log.Warn("kafka broker configuration lookup failed, skipping replication factor validation", zap.String("configName", MinInsyncReplicasConfigName), zap.Int16("replicationFactor", c.ReplicationFactor), zap.Error(err)) return nil } if !found { - log.Warn("Kafka broker configuration not found, assume replication factor is valid", + log.Warn("kafka broker configuration not found, skipping replication factor validation", zap.String("configName", MinInsyncReplicasConfigName), zap.Int16("replicationFactor", c.ReplicationFactor)) return nil @@ -644,9 +642,9 @@ 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, topic, info) + err = adjustExistingTopicOption(changefeedID, admin, options, info) } else { - adjustNewTopicOptions(admin, changefeedID, options, topic) + adjustNewTopicOptions(admin, changefeedID, options) } if err != nil { return err @@ -660,26 +658,17 @@ func adjustExistingTopicOption( changefeedID common.ChangeFeedID, admin ClusterAdminClient, options *options, - topic string, info TopicDetail, ) error { maxMessageBytes, found, err := getTopicMaxMessageBytes(admin, info.Name) if err != nil || !found { - log.Warn("`max.message.bytes` not found from topic's configuration, use the option `MaxMessageBytes` as default", + 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 } options.MaxMessageBytes = maxMessageBytes - // no need to create the topic, - // but we would have to log user if they found enter wrong topic name later - if options.AutoCreate { - log.Warn("topic already exist, TiCDC will not create the topic", - zap.String("namespace", changefeedID.Keyspace()), zap.String("changefeed", changefeedID.Name()), - zap.String("topic", topic), zap.Any("detail", info)) - } - if err = options.setPartitionNum(changefeedID, info.NumPartitions); err != nil { return err } @@ -690,13 +679,12 @@ func adjustNewTopicOptions( admin ClusterAdminClient, changefeedID common.ChangeFeedID, options *options, - topic string, ) { // 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("`message.max.bytes` not found from broker's configuration, use the option `MaxMessageBytes` as default", + log.Warn("kafka broker 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)) messageMaxBytes = options.MaxMessageBytes @@ -706,9 +694,6 @@ func adjustNewTopicOptions( // topic not exists yet, and user does not specify the `partition-num` in the sink uri. if options.PartitionNum == 0 { options.PartitionNum = defaultPartitionNum - log.Warn("partition-num is not set, use the default partition count", - zap.String("namespace", changefeedID.Keyspace()), zap.String("changefeed", changefeedID.Name()), - zap.String("topic", topic), zap.Int32("partitions", options.PartitionNum)) } } @@ -764,7 +749,5 @@ func getTopicConfig( return c, true, nil } - log.Info("kafka sink cannot get the configuration from topic, try to get it from broker", - zap.String("topic", topicName), zap.String("config", topicConfigName), zap.Error(err)) return admin.GetBrokerConfig(brokerConfigName) } diff --git a/pkg/sink/kafka/sarama_async_producer.go b/pkg/sink/kafka/sarama_async_producer.go index cc26e23531..d3e1781c71 100644 --- a/pkg/sink/kafka/sarama_async_producer.go +++ b/pkg/sink/kafka/sarama_async_producer.go @@ -62,13 +62,13 @@ func (p *saramaAsyncProducer) Close() { // To prevent the scenario mentioned above, close the client first. start := time.Now() if err := p.client.Close(); err != nil { - log.Warn("Close kafka async producer client error", + log.Warn("kafka async producer client close failed", zap.String("keyspace", p.changefeedID.Keyspace()), zap.String("changefeed", p.changefeedID.Name()), zap.Duration("duration", time.Since(start)), zap.Error(err)) } else { - log.Info("Close kafka async producer client success", + log.Info("kafka async producer client closed", zap.String("keyspace", p.changefeedID.Keyspace()), zap.String("changefeed", p.changefeedID.Name()), zap.Duration("duration", time.Since(start))) @@ -76,13 +76,13 @@ func (p *saramaAsyncProducer) Close() { start = time.Now() if err := p.producer.Close(); err != nil { - log.Warn("Close kafka async producer error", + log.Warn("kafka async producer close failed", zap.String("keyspace", p.changefeedID.Keyspace()), zap.String("changefeed", p.changefeedID.Name()), zap.Duration("duration", time.Since(start)), zap.Error(err)) } else { - log.Info("Close kafka async producer success", + log.Info("kafka async producer closed", zap.String("keyspace", p.changefeedID.Keyspace()), zap.String("changefeed", p.changefeedID.Name()), zap.Duration("duration", time.Since(start))) @@ -106,7 +106,7 @@ func (p *saramaAsyncProducer) AsyncRunCallback( meta.callback() } default: - log.Error("unknown message metadata type in async producer", + log.Error("kafka producer received unknown message metadata type", zap.Any("metadata", ack.Metadata)) } } @@ -125,7 +125,7 @@ func (p *saramaAsyncProducer) AsyncRunCallback( } func (p *saramaAsyncProducer) handleProducerError(err *sarama.ProducerError) error { - log.Error("send message to kafka failed", + log.Error("kafka message send failed", zap.String("keyspace", p.changefeedID.Keyspace()), zap.String("changefeed", p.changefeedID.Name()), zap.String("eventContext", BuildEventLogContext( diff --git a/pkg/sink/kafka/sarama_config.go b/pkg/sink/kafka/sarama_config.go index 8c194ee224..51dbd2384e 100644 --- a/pkg/sink/kafka/sarama_config.go +++ b/pkg/sink/kafka/sarama_config.go @@ -87,12 +87,9 @@ func newSaramaConfig(ctx context.Context, o *options) (*sarama.Config, error) { case "zstd": config.Producer.Compression = sarama.CompressionZSTD default: - log.Warn("Unsupported compression algorithm", zap.String("compression", o.Compression)) + log.Warn("unsupported kafka compression algorithm", zap.String("compression", o.Compression)) config.Producer.Compression = sarama.CompressionNone } - if config.Producer.Compression != sarama.CompressionNone { - log.Info("Kafka producer uses " + compression + " compression algorithm") - } if o.EnableTLS { // for SSL encryption with a trust CA certificate, we must populate the @@ -223,8 +220,7 @@ func selectKafkaVersion(detectedVersion sarama.KafkaVersion, o *options) (sarama } if !assignedVersion.IsAtLeast(maxKafkaVersion) && assignedVersion.String() != detectedVersion.String() { - log.Warn("The Kafka version you assigned may not be correct. "+ - "Please assign a version equal to or less than the specified version", + log.Warn("configured kafka version differs from detected version", zap.String("assignedVersion", assignedVersion.String()), zap.String("desiredVersion", detectedVersion.String())) } diff --git a/pkg/sink/kafka/sarama_factory.go b/pkg/sink/kafka/sarama_factory.go index c9678ce784..6f0af88aa6 100644 --- a/pkg/sink/kafka/sarama_factory.go +++ b/pkg/sink/kafka/sarama_factory.go @@ -42,7 +42,7 @@ func NewSaramaFactory( config, err := newSaramaConfig(ctx, o) duration := time.Since(start).Seconds() if duration > 2 { - log.Warn("new sarama config cost too much time", + log.Warn("kafka configuration initialization is slow", zap.Stringer("changefeedID", changefeedID), zap.Any("duration", duration)) } if err != nil { @@ -73,7 +73,7 @@ func newAdminClient(changefeedID common.ChangeFeedID, endpoints []string, config client, err := sarama.NewClient(endpoints, config) duration := time.Since(start).Seconds() if duration > 2 { - log.Warn("new sarama client cost too much time", + log.Warn("kafka client initialization is slow", zap.Any("duration", duration), zap.Stringer("changefeedID", changefeedID)) } if err != nil { @@ -84,7 +84,7 @@ func newAdminClient(changefeedID common.ChangeFeedID, endpoints []string, config admin, err := sarama.NewClusterAdminFromClient(client) duration = time.Since(start).Seconds() if duration > 2 { - log.Warn("new sarama cluster admin cost too much time", + log.Warn("kafka admin client initialization is slow", zap.Any("duration", duration), zap.Stringer("changefeedID", changefeedID)) } if err != nil { diff --git a/pkg/sink/kafka/sarama_sync_producer.go b/pkg/sink/kafka/sarama_sync_producer.go index 754d1db9e1..fcf1c9c258 100644 --- a/pkg/sink/kafka/sarama_sync_producer.go +++ b/pkg/sink/kafka/sarama_sync_producer.go @@ -58,7 +58,7 @@ func (p *saramaSyncProducer) SendMessage(topic string, partitionNum int32, messa if err == nil { return nil } - log.Error("send message to kafka failed", + log.Error("kafka message send failed", zap.String("keyspace", p.id.Keyspace()), zap.String("changefeed", p.id.Name()), zap.String("eventContext", BuildEventLogContext(p.id.Keyspace(), p.id.Name(), message.LogInfo)), @@ -84,7 +84,7 @@ func (p *saramaSyncProducer) SendMessages(topic string, partitionNum int32, mess if err == nil { return nil } - log.Error("send message to kafka failed", + log.Error("kafka message send failed", zap.String("keyspace", p.id.Keyspace()), zap.String("changefeed", p.id.Name()), zap.String("eventContext", BuildEventLogContext(p.id.Keyspace(), p.id.Name(), message.LogInfo)), @@ -94,7 +94,7 @@ func (p *saramaSyncProducer) SendMessages(topic string, partitionNum int32, mess func (p *saramaSyncProducer) Close() { if p.closed.Load() { - log.Warn("kafka DDL producer already closed", + log.Warn("kafka ddl producer already closed", zap.String("keyspace", p.id.Keyspace()), zap.String("changefeed", p.id.Name())) return @@ -106,7 +106,7 @@ func (p *saramaSyncProducer) Close() { // so producer.Close() alone won't release the underlying client resources. if p.client != nil { if err := p.client.Close(); err != nil { - log.Warn("Close Kafka DDL producer client with error", + log.Warn("kafka ddl producer client close failed", zap.String("keyspace", p.id.Keyspace()), zap.String("changefeed", p.id.Name()), zap.Duration("duration", time.Since(start)), @@ -115,7 +115,7 @@ func (p *saramaSyncProducer) Close() { } if p.producer != nil { if err := p.producer.Close(); err != nil { - log.Error("Close Kafka DDL producer with error", + log.Error("kafka ddl producer close failed", zap.String("keyspace", p.id.Keyspace()), zap.String("changefeed", p.id.Name()), zap.Duration("duration", time.Since(start)), @@ -123,7 +123,7 @@ func (p *saramaSyncProducer) Close() { return } } - log.Info("Kafka DDL producer closed", + log.Info("kafka ddl producer closed", zap.String("keyspace", p.id.Keyspace()), zap.String("changefeed", p.id.Name()), zap.Duration("duration", time.Since(start))) From 549f420dc57062778da41fc632bbfdcbcf4589c0 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Thu, 30 Jul 2026 12:30:06 +0800 Subject: [PATCH 07/11] unify the log fields --- .../sink/topicmanager/kafka_topic_manager.go | 10 ++++---- pkg/sink/kafka/options.go | 3 ++- pkg/sink/kafka/sarama_factory.go | 24 ++++++++++++------- 3 files changed, 22 insertions(+), 15 deletions(-) diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go index b31e29610f..878ffe24ef 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go @@ -137,8 +137,8 @@ func (m *kafkaTopicManager) tryUpdatePartitionsAndLogging(topic string, partitio zap.String("keyspace", m.changefeedID.Keyspace()), zap.String("changefeed", m.changefeedID.Name()), zap.String("topic", topic), - zap.Int32("oldPartitionNumber", oldPartitions.(int32)), - zap.Int32("newPartitionNumber", partitions), + zap.Int32("oldPartitionNum", oldPartitions.(int32)), + zap.Int32("newPartitionNum", partitions), ) } } else { @@ -245,7 +245,7 @@ func (m *kafkaTopicManager) createTopic( zap.String("keyspace", m.changefeedID.Keyspace()), zap.String("changefeed", m.changefeedID.Name()), zap.String("topic", topicName), - zap.Int32("partitionNumber", m.cfg.PartitionNum), + zap.Int32("partitionNum", m.cfg.PartitionNum), zap.Int16("replicationFactor", m.cfg.ReplicationFactor), zap.Error(err), zap.Duration("duration", time.Since(start)), @@ -258,7 +258,7 @@ func (m *kafkaTopicManager) createTopic( zap.String("keyspace", m.changefeedID.Keyspace()), zap.String("changefeed", m.changefeedID.Name()), zap.String("topic", topicName), - zap.Int32("partitionNumber", m.cfg.PartitionNum), + zap.Int32("partitionNum", m.cfg.PartitionNum), zap.Int16("replicationFactor", m.cfg.ReplicationFactor), zap.Duration("duration", time.Since(start)), ) @@ -330,7 +330,7 @@ func (m *kafkaTopicManager) useConfiguredPartitionNum(topicName string, cause er zap.String("keyspace", m.changefeedID.Keyspace()), zap.String("changefeed", m.changefeedID.Name()), zap.String("topic", topicName), - zap.Int32("partitionNumber", m.cfg.PartitionNum), + zap.Int32("partitionNum", m.cfg.PartitionNum), zap.Error(cause)) m.tryUpdatePartitionsAndLogging(topicName, m.cfg.PartitionNum) return m.cfg.PartitionNum diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go index 523bcf73ea..104e962067 100644 --- a/pkg/sink/kafka/options.go +++ b/pkg/sink/kafka/options.go @@ -208,7 +208,8 @@ func (o *options) setPartitionNum(changefeedID common.ChangeFeedID, realPartitio if o.PartitionNum < realPartitionCount { log.Warn("configured kafka partition count is lower than topic partition count", zap.String("namespace", changefeedID.Keyspace()), zap.String("changefeed", changefeedID.Name()), - zap.Int32("sinkUriPartitions", o.PartitionNum), zap.Int32("topicPartitions", realPartitionCount)) + zap.Int32("configuredPartitionNum", o.PartitionNum), + zap.Int32("topicPartitionNum", realPartitionCount)) return nil } diff --git a/pkg/sink/kafka/sarama_factory.go b/pkg/sink/kafka/sarama_factory.go index 6f0af88aa6..762e4ac930 100644 --- a/pkg/sink/kafka/sarama_factory.go +++ b/pkg/sink/kafka/sarama_factory.go @@ -40,10 +40,12 @@ func NewSaramaFactory( ) (Factory, error) { start := time.Now() config, err := newSaramaConfig(ctx, o) - duration := time.Since(start).Seconds() - if duration > 2 { + duration := time.Since(start) + if duration > 2*time.Second { log.Warn("kafka configuration initialization is slow", - zap.Stringer("changefeedID", changefeedID), zap.Any("duration", duration)) + zap.String("keyspace", changefeedID.Keyspace()), + zap.String("changefeed", changefeedID.Name()), + zap.Duration("duration", duration)) } if err != nil { return nil, err @@ -71,10 +73,12 @@ func NewSaramaFactory( func newAdminClient(changefeedID common.ChangeFeedID, endpoints []string, config *sarama.Config) (ClusterAdminClient, error) { start := time.Now() client, err := sarama.NewClient(endpoints, config) - duration := time.Since(start).Seconds() - if duration > 2 { + duration := time.Since(start) + if duration > 2*time.Second { log.Warn("kafka client initialization is slow", - zap.Any("duration", duration), zap.Stringer("changefeedID", changefeedID)) + zap.String("keyspace", changefeedID.Keyspace()), + zap.String("changefeed", changefeedID.Name()), + zap.Duration("duration", duration)) } if err != nil { return nil, errors.WrapError(errors.ErrNewKafkaSink, err) @@ -82,10 +86,12 @@ func newAdminClient(changefeedID common.ChangeFeedID, endpoints []string, config start = time.Now() admin, err := sarama.NewClusterAdminFromClient(client) - duration = time.Since(start).Seconds() - if duration > 2 { + duration = time.Since(start) + if duration > 2*time.Second { log.Warn("kafka admin client initialization is slow", - zap.Any("duration", duration), zap.Stringer("changefeedID", changefeedID)) + zap.String("keyspace", changefeedID.Keyspace()), + zap.String("changefeed", changefeedID.Name()), + zap.Duration("duration", duration)) } if err != nil { // `sarama.NewClusterAdminFromClient` does not take ownership of the client, From a5f93e426638ae996d9c5f0a56d884a8f287d174 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Thu, 30 Jul 2026 15:10:40 +0800 Subject: [PATCH 08/11] fix code --- downstreamadapter/sink/kafka/sink.go | 5 +---- .../sink/topicmanager/kafka_topic_manager.go | 20 ++++++++++--------- pkg/sink/kafka/options_test.go | 6 +++--- .../kafka_compression/run.sh | 5 ----- 4 files changed, 15 insertions(+), 21 deletions(-) diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index 1e04dc67f2..4851dbfe67 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -483,7 +483,7 @@ func (s *sink) sendDDLEvent(event *commonEvent.DDLEvent) error { } log.Info("kafka ddl event sent", zap.String("keyspace", s.changefeedID.Keyspace()), zap.String("changefeed", s.changefeedID.Name()), - zap.Any("startTs", event.GetStartTs()), zap.Any("commitTs", event.GetCommitTs()), zap.Any("event", event.GetDDLQuery()), + zap.Any("startTs", event.GetStartTs()), zap.Any("commitTs", event.GetCommitTs()), zap.Any("DDL", event.GetDDLQuery()), zap.String("schema", event.GetSchemaName()), zap.String("table", event.GetTableName())) return nil } @@ -517,9 +517,6 @@ func (s *sink) sendCheckpoint(ctx context.Context) error { return context.Cause(ctx) case ts, ok := <-s.checkpointChan: if !ok { - log.Warn("kafka checkpoint channel closed", - zap.String("keyspace", s.changefeedID.Keyspace()), - zap.String("changefeed", s.changefeedID.Name())) return nil } diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go index 878ffe24ef..3db0938012 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go @@ -253,15 +253,6 @@ func (m *kafkaTopicManager) createTopic( return 0, err } - log.Info( - "kafka topic created", - zap.String("keyspace", m.changefeedID.Keyspace()), - zap.String("changefeed", m.changefeedID.Name()), - zap.String("topic", topicName), - zap.Int32("partitionNum", m.cfg.PartitionNum), - zap.Int16("replicationFactor", m.cfg.ReplicationFactor), - zap.Duration("duration", time.Since(start)), - ) m.tryUpdatePartitionsAndLogging(topicName, m.cfg.PartitionNum) return m.cfg.PartitionNum, nil @@ -294,6 +285,7 @@ func (m *kafkaTopicManager) CreateTopicAndWaitUntilVisible( return numPartition, nil } + start := time.Now() partitionNum, err := m.createTopic(ctx, topicName) if err != nil { if kafka.IsAdminAuthorizationFailed(err) { @@ -307,6 +299,16 @@ func (m *kafkaTopicManager) CreateTopicAndWaitUntilVisible( return 0, err } + log.Info( + "kafka topic created", + zap.String("keyspace", m.changefeedID.Keyspace()), + zap.String("changefeed", m.changefeedID.Name()), + zap.String("topic", topicName), + zap.Int32("partitionNum", partitionNum), + zap.Int16("replicationFactor", m.cfg.ReplicationFactor), + zap.Duration("duration", time.Since(start)), + ) + return partitionNum, nil } diff --git a/pkg/sink/kafka/options_test.go b/pkg/sink/kafka/options_test.go index 484ca52f97..025df9086b 100644 --- a/pkg/sink/kafka/options_test.go +++ b/pkg/sink/kafka/options_test.go @@ -409,9 +409,9 @@ func TestClientID(t *testing.T) { func TestTimeout(t *testing.T) { options := NewOptions() - require.Equal(t, defaultTimeout, options.DialTimeout) - require.Equal(t, defaultTimeout, options.ReadTimeout) - require.Equal(t, defaultTimeout, options.WriteTimeout) + require.Equal(t, 10*time.Second, options.DialTimeout) + require.Equal(t, 10*time.Second, options.ReadTimeout) + require.Equal(t, 10*time.Second, options.WriteTimeout) uri := "kafka://127.0.0.1:9092/kafka-test?dial-timeout=5s&read-timeout=1000ms" + "&write-timeout=2m" diff --git a/tests/integration_tests/kafka_compression/run.sh b/tests/integration_tests/kafka_compression/run.sh index e0f85df648..a57cd937b1 100755 --- a/tests/integration_tests/kafka_compression/run.sh +++ b/tests/integration_tests/kafka_compression/run.sh @@ -18,11 +18,6 @@ function test_compression() { run_kafka_consumer $WORK_DIR "kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=canal-json&version=${KAFKA_VERSION}&enable-tidb-extension=true" run_sql_file $CUR/data/$1_data.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT} - compression_algorithm=$(grep "Kafka producer uses $1 compression algorithm" "$WORK_DIR/cdc.log") - if [[ "$compression_algorithm" -ne 1 ]]; then - echo "can't found producer compression algorithm" - exit 1 - fi check_table_exists test.$1_finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200 check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml cdc_cli_changefeed pause -c $1 From 8ea28c399863bf07567ac6513c89dfc399e01ce0 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Thu, 30 Jul 2026 15:22:22 +0800 Subject: [PATCH 09/11] adjust logs --- downstreamadapter/sink/kafka/sink.go | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index 4851dbfe67..2133e1c6ba 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -455,6 +455,10 @@ func (s *sink) sendDDLEvent(event *commonEvent.DDLEvent) error { return err } if message == nil { + log.Info("kafka ddl event skipped", + zap.String("keyspace", s.changefeedID.Keyspace()), zap.String("changefeed", s.changefeedID.Name()), + zap.Uint64("startTs", e.GetStartTs()), zap.Uint64("commitTs", e.GetCommitTs()), + zap.String("query", e.Query)) continue } codecCommon.SetDDLMessageLogInfo(message, e) @@ -480,11 +484,11 @@ func (s *sink) sendDDLEvent(event *commonEvent.DDLEvent) error { if err != nil { return err } + log.Info("kafka ddl event sent", + zap.String("keyspace", s.changefeedID.Keyspace()), zap.String("changefeed", s.changefeedID.Name()), + zap.Uint64("startTs", e.GetStartTs()), zap.Uint64("commitTs", e.GetCommitTs()), + zap.String("query", e.GetDDLQuery())) } - log.Info("kafka ddl event sent", - zap.String("keyspace", s.changefeedID.Keyspace()), zap.String("changefeed", s.changefeedID.Name()), - zap.Any("startTs", event.GetStartTs()), zap.Any("commitTs", event.GetCommitTs()), zap.Any("DDL", event.GetDDLQuery()), - zap.String("schema", event.GetSchemaName()), zap.String("table", event.GetTableName())) return nil } From f774c9a8141e60321984748248fe1d3ac37fc887 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Thu, 30 Jul 2026 15:36:10 +0800 Subject: [PATCH 10/11] adjust logs --- pkg/sink/kafka/options.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go index 104e962067..361ca61a9c 100644 --- a/pkg/sink/kafka/options.go +++ b/pkg/sink/kafka/options.go @@ -663,7 +663,7 @@ func adjustExistingTopicOption( ) error { maxMessageBytes, found, err := getTopicMaxMessageBytes(admin, info.Name) if err != nil || !found { - log.Warn("kafka topic max message bytes unavailable, using configured value", + 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 @@ -685,7 +685,7 @@ func adjustNewTopicOptions( // 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 max message bytes unavailable, using configured value", + 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 From 185f1a52c350a47728e7fb1d9202bfe9f3967390 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Thu, 30 Jul 2026 15:45:26 +0800 Subject: [PATCH 11/11] adjust logs --- pkg/sink/kafka/options.go | 3 --- pkg/sink/kafka/sarama_factory.go | 13 +++++++++++++ 2 files changed, 13 insertions(+), 3 deletions(-) diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go index 361ca61a9c..da1dfb3640 100644 --- a/pkg/sink/kafka/options.go +++ b/pkg/sink/kafka/options.go @@ -199,9 +199,6 @@ func (o *options) setPartitionNum(changefeedID common.ChangeFeedID, realPartitio // user does not specify the `partition-num` in the sink-uri if o.PartitionNum == 0 { o.PartitionNum = realPartitionCount - log.Info("kafka partition count set from topic metadata", - zap.String("namespace", changefeedID.Keyspace()), zap.String("changefeed", changefeedID.Name()), - zap.Int32("partitionNum", realPartitionCount)) return nil } diff --git a/pkg/sink/kafka/sarama_factory.go b/pkg/sink/kafka/sarama_factory.go index 762e4ac930..8f73ca70b5 100644 --- a/pkg/sink/kafka/sarama_factory.go +++ b/pkg/sink/kafka/sarama_factory.go @@ -62,6 +62,19 @@ func NewSaramaFactory( if err = adjustOptions(changefeedID, admin, o, o.Topic); err != nil { return nil, err } + log.Info("kafka sink configuration resolved", + zap.String("namespace", changefeedID.Keyspace()), + zap.String("changefeed", changefeedID.Name()), + 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), + zap.Duration("dialTimeout", o.DialTimeout), + zap.Duration("readTimeout", o.ReadTimeout), + zap.Duration("writeTimeout", o.WriteTimeout)) return &saramaFactory{ changefeedID: changefeedID,