Skip to content

kafka: improve topic validation, timeout checks, and logging #5818

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ti-chi-bot merged 11 commits into from
Jul 30, 2026
Merged

kafka: improve topic validation, timeout checks, and logging #5818

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
e417fd3
adjust the code
3AceShowHand Jul 29, 2026
08747bb
add default timeout
3AceShowHand Jul 29, 2026
03b4643
remove useless log
3AceShowHand Jul 30, 2026
0140249
remove useless logs
3AceShowHand Jul 30, 2026
33f9c5c
adjust kafka version detection log
3AceShowHand Jul 30, 2026
80f00d8
unify the log format
3AceShowHand Jul 30, 2026
549f420
unify the log fields
3AceShowHand Jul 30, 2026
a5f93e4
fix code
3AceShowHand Jul 30, 2026
8ea28c3
adjust logs
3AceShowHand Jul 30, 2026
f774c9a
adjust logs
3AceShowHand Jul 30, 2026
185f1a5
adjust logs
3AceShowHand Jul 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 9 additions & 36 deletions downstreamadapter/sink/kafka/sink.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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())))
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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 {
Expand All @@ -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 {
Expand Down Expand Up @@ -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]
Expand Down Expand Up @@ -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 {
Expand All @@ -447,16 +431,11 @@ 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,
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
Expand All @@ -476,9 +455,10 @@ 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))
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)
Expand All @@ -504,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 sink send DDL event",
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()))
return nil
}

Expand Down Expand Up @@ -541,9 +521,6 @@ 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",
zap.String("keyspace", s.changefeedID.Keyspace()),
zap.String("changefeed", s.changefeedID.Name()))
return nil
}

Expand Down Expand Up @@ -596,10 +573,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)
Expand Down
74 changes: 32 additions & 42 deletions downstreamadapter/sink/topicmanager/kafka_topic_manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -137,23 +133,16 @@ 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),
zap.Int32("oldPartitionNumber", oldPartitions.(int32)),
zap.Int32("newPartitionNumber", partitions),
zap.Int32("oldPartitionNum", oldPartitions.(int32)),
zap.Int32("newPartitionNum", partitions),
)
}
} 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),
)
}
}

Expand All @@ -172,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)),
Expand Down Expand Up @@ -201,33 +190,32 @@ 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
}
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.Duration("duration", time.Since(start)))
_, ok := meta[topicName]
if !ok {
return errors.ErrKafkaAdminAPI.GenWithStackByArgs("describe-topic", topicName)
}
return nil
}, retry.WithBackoffBaseDelay(500),
retry.WithBackoffMaxDelay(1000),
retry.WithMaxTries(6),
)

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
}

Expand All @@ -253,27 +241,18 @@ 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),
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)),
)
return 0, err
}

log.Info(
"Kafka admin client create the topic success",
zap.String("keyspace", m.changefeedID.Keyspace()),
zap.String("changefeed", m.changefeedID.Name()),
zap.String("topic", topicName),
zap.Int32("partitionNumber", 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
Expand Down Expand Up @@ -306,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) {
Expand All @@ -319,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
}

Expand All @@ -338,11 +328,11 @@ 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),
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
Expand Down
4 changes: 1 addition & 3 deletions downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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: {
Expand Down
21 changes: 3 additions & 18 deletions pkg/sink/kafka/admin.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
}

Expand All @@ -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
}

Expand All @@ -136,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),
Expand Down Expand Up @@ -190,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))
Expand All @@ -199,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))
Expand Down
2 changes: 1 addition & 1 deletion pkg/sink/kafka/claimcheck/claim_check.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)),
Expand Down
Loading
Loading