kafka: improve topic validation, timeout checks, and logging

[3AceShowHand]

What problem does this PR solve?

Kafka sink has several correctness and diagnosability problems:

• Topic creation can treat an empty metadata response as proof that the requested topic is visible.
• Explicitly configured zero or negative Kafka timeouts are accepted.
• Kafka sink logs contain noisy normal-path messages, duplicated errors, inconsistent messages and fields, and per-message payload diagnostics. Some integration tests also validate implementation-specific log text instead of end-to-end behavior.

Issue Number: close #5819

What is changed and how it works?

• Require refreshed metadata to contain the requested topic before topic creation succeeds. The success log is emitted only after visibility is confirmed.
• Reject non-positive dial-timeout, read-timeout, and write-timeout values with CDC:ErrKafkaInvalidConfig. Omitted values keep the existing 10-second defaults.
• Improve Kafka sink logs:
  • remove normal channel-close, low-level configuration lookup, per-message payload, and log-before-return messages;
  • keep producer-level send failures with their original Kafka error and event context;
  • consolidate Kafka version detection failures into one fallback warning without changing version selection behavior;
  • standardize Kafka log messages, fields, and duration values;
  • bind DDL sent/skipped logs to each child DDL event;
  • emit one resolved Kafka sink configuration summary.
• Remove integration assertions that grep compression or Kafka payload log text. The compression test still validates end-to-end replication with sync_diff; generic log-redaction coverage remains in the integration suite.
• Add focused tests for empty topic metadata, non-positive timeout values, and Kafka version selection.

Check List

Tests

• Unit test
  • go test --tags=intest ./pkg/sink/kafka ./downstreamadapter/sink/topicmanager -count=1
• Other
  • scripts/check-log-style.sh
  • git diff --check

Questions

Will it cause performance regression or break compatibility?

No performance regression is expected. This intentionally tightens timeout validation: explicitly configured zero or negative Kafka timeouts are now rejected. Positive values and omitted values retain their existing behavior. Kafka version selection behavior is unchanged. Kafka sink log output changes intentionally.

Do you need to update user documentation, design documentation or monitoring documentation?

No.

Release note

None

[coderabbitai]

📝 Walkthrough

Walkthrough

Kafka topic visibility and timeout validation are tightened, Kafka version selection is split into detection and assignment handling, routine Kafka logs are reduced, and Kafka log-redaction and compression log checks are removed.

Changes

Kafka validation

Layer / File(s)	Summary
Topic visibility validation downstreamadapter/sink/topicmanager/kafka_topic_manager.go, downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go	waitUntilTopicVisible now requires matching metadata and returns a describe-topic error when absent; the test models an empty successful metadata response.
Kafka timeout validation pkg/sink/kafka/options.go, pkg/sink/kafka/options_test.go	Timeout defaults use defaultTimeout, and configured dial, read, and write timeouts must be greater than zero with invalid configurations tested.

Kafka version configuration

Layer / File(s)	Summary
Kafka version detection and selection pkg/sink/kafka/sarama_config.go, pkg/sink/kafka/sarama_config_test.go	Version detection, fallback handling, and explicit version selection are separated, with invalid assigned versions covered by table-driven tests.

Kafka logging and integration coverage

Layer / File(s)	Summary
Routine logging reduction downstreamadapter/sink/kafka/sink.go, downstreamadapter/sink/topicmanager/kafka_topic_manager.go, pkg/sink/kafka/*	Routine lifecycle, processing, metadata, configuration, producer, and cancellation logs are removed or renamed without changing the described processing and return behavior.
Integration coverage updates tests/integration_tests/log_redaction/run.sh, tests/integration_tests/run_light_it_in_ci.sh, tests/integration_tests/kafka_compression/run.sh	Kafka-specific redaction checks and compression log-marker validation are removed.

Estimated code review effort: 3 (Moderate) | ~30 minutes

Possibly related PRs

• pingcap/ticdc#5368: Related Kafka topic-manager test mocking changes.
• pingcap/ticdc#5696: Related Kafka topic metadata handling.
• pingcap/ticdc#5786: Related topic creation and visibility workflow changes.

Suggested reviewers: hongyunyan, wk989898

Poem

I’m a rabbit guarding Kafka’s gate,
Empty maps must kindly wait.
Zero clocks and negatives flee,
Topics appear, then carrots cheer.

🚥 Pre-merge checks | ✅ 3 | ❌ 2

❌ Failed checks (2 warnings)

Check name	Status	Explanation	Resolution
Out of Scope Changes check	⚠️ Warning	Several logging-only changes and test removals in unrelated Kafka files go beyond topic visibility and timeout validation.	Remove or split the unrelated logging and integration test changes into a separate PR focused on the visibility and timeout fixes.
Docstring Coverage	⚠️ Warning	Docstring coverage is 14.29% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (3 passed)

Check name	Status	Explanation
Linked Issues check	✅ Passed	The PR implements the linked requirements to retry until topic metadata contains the topic and reject non-positive timeout values.
Title check	✅ Passed	The title clearly summarizes the main changes: Kafka topic validation, timeout checks, and logging cleanup.
Description check	✅ Passed	The description matches the template well, includes the issue number, tests, questions, and release note.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[3AceShowHand]

/test all

[3AceShowHand]

/test all

[ti-chi-bot]

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: asddongmen, wk989898

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Details

Needs approval from an approver in each of these files:

• OWNERS [asddongmen,wk989898]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[ti-chi-bot]

[LGTM Timeline notifier]

Timeline:

• 2026-07-30 02:23:05.640877616 +0000 UTC m=+2062771.676972672: ☑️ agreed by wk989898.
• 2026-07-30 02:24:55.904044917 +0000 UTC m=+2062881.940139972: ☑️ agreed by asddongmen.

[coderabbitai]

Actionable comments posted: 1

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@pkg/sink/kafka/options_test.go`:
- Around line 412-414: Update the three timeout assertions in the options test
to compare DialTimeout, ReadTimeout, and WriteTimeout against the independent
expected value 10*time.Second instead of defaultTimeout. Keep the test focused
and deterministic, and do not modify the production default constant.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 0a5fb09f-f336-4d97-bbb6-477ecf75455b

📥 Commits

Reviewing files that changed from the base of the PR and between 69ab69a and 03b4643.

📒 Files selected for processing (4)

• downstreamadapter/sink/topicmanager/kafka_topic_manager.go
• downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go
• pkg/sink/kafka/options.go
• pkg/sink/kafka/options_test.go

[3AceShowHand]

/hold

[coderabbitai]

Actionable comments posted: 1

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@pkg/sink/kafka/sarama_config.go`:
- Around line 224-230: Update the version-warning condition in the Sarama
configuration logic around assignedVersion so it uses the detected broker
version rather than maxKafkaVersion: only compare versions when detection did
not reach its detection ceiling, and warn when assignedVersion is greater than
detectedVersion. Preserve the existing warning message and structured fields.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: d34319e9-28b4-471a-ae82-aef15fe0bd2f

📥 Commits

Reviewing files that changed from the base of the PR and between 03b4643 and 33f9c5c.

📒 Files selected for processing (8)

• downstreamadapter/sink/kafka/sink.go
• downstreamadapter/sink/topicmanager/kafka_topic_manager.go
• pkg/sink/kafka/admin.go
• pkg/sink/kafka/sarama_async_producer.go
• pkg/sink/kafka/sarama_config.go
• pkg/sink/kafka/sarama_config_test.go
• tests/integration_tests/log_redaction/run.sh
• tests/integration_tests/run_light_it_in_ci.sh

💤 Files with no reviewable changes (4)

• pkg/sink/kafka/sarama_async_producer.go
• pkg/sink/kafka/admin.go
• tests/integration_tests/log_redaction/run.sh
• downstreamadapter/sink/kafka/sink.go

🚧 Files skipped from review as they are similar to previous changes (1)

• downstreamadapter/sink/topicmanager/kafka_topic_manager.go

[3AceShowHand]

/test all

[coderabbitai]

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

downstreamadapter/sink/topicmanager/kafka_topic_manager.go (1)

288-297: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

Cache the topic only after visibility succeeds.

createTopic stores the partition count before waitUntilTopicVisible returns. If metadata never appears and the wait fails, a later GetPartitionNum call returns this cached value immediately (Lines 98-100), bypassing the new visibility check and potentially using a topic that is still unavailable. Move the cache update into the successful post-wait path, or remove it when waiting fails.

Proposed fix

 func (m *kafkaTopicManager) createTopic(
 	_ context.Context,
 	topicName string,
 ) (int32, error) {
 	// ...
-	m.tryUpdatePartitionsAndLogging(topicName, m.cfg.PartitionNum)
-
 	return m.cfg.PartitionNum, nil
 }

 	err = m.waitUntilTopicVisible(ctx, topicName)
 	if err != nil {
 		return 0, err
 	}
+	m.tryUpdatePartitionsAndLogging(topicName, partitionNum)

 	log.Info(

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@downstreamadapter/sink/topicmanager/kafka_topic_manager.go` around lines 288
- 297, Update the topic creation flow around createTopic and
waitUntilTopicVisible so the partition-count cache is populated only after
waitUntilTopicVisible succeeds. If visibility waiting fails, remove any cache
entry created by createTopic before returning the error, ensuring later
GetPartitionNum calls cannot bypass the visibility check.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Outside diff comments:
In `@downstreamadapter/sink/topicmanager/kafka_topic_manager.go`:
- Around line 288-297: Update the topic creation flow around createTopic and
waitUntilTopicVisible so the partition-count cache is populated only after
waitUntilTopicVisible succeeds. If visibility waiting fails, remove any cache
entry created by createTopic before returning the error, ensuring later
GetPartitionNum calls cannot bypass the visibility check.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: f4e623a2-08d4-40a7-96be-e69fb3485144

📥 Commits

Reviewing files that changed from the base of the PR and between 33f9c5c and a5f93e4.

📒 Files selected for processing (11)

• downstreamadapter/sink/kafka/sink.go
• downstreamadapter/sink/topicmanager/kafka_topic_manager.go
• pkg/sink/kafka/admin.go
• pkg/sink/kafka/claimcheck/claim_check.go
• pkg/sink/kafka/options.go
• pkg/sink/kafka/options_test.go
• pkg/sink/kafka/sarama_async_producer.go
• pkg/sink/kafka/sarama_config.go
• pkg/sink/kafka/sarama_factory.go
• pkg/sink/kafka/sarama_sync_producer.go
• tests/integration_tests/kafka_compression/run.sh

💤 Files with no reviewable changes (1)

• tests/integration_tests/kafka_compression/run.sh

🚧 Files skipped from review as they are similar to previous changes (5)

• pkg/sink/kafka/sarama_async_producer.go
• pkg/sink/kafka/sarama_config.go
• pkg/sink/kafka/admin.go
• downstreamadapter/sink/kafka/sink.go
• pkg/sink/kafka/options.go

[3AceShowHand]

/unhold

[3AceShowHand]

/test all