kafka: create and wait for topic during verification

[3AceShowHand]

What problem does this PR solve?

Issue Number: close #5896

What is changed and how it works?

When a Kafka consumer starts immediately after a changefeed is created, it can auto-create the topic with the broker defaults before TiCDC initializes the Kafka sink. Kafka sink verification previously sent a validate-only CreateTopics request, so the topic was not actually created. This can leave the topic with a partition count different from the changefeed configuration, causing message delivery failures and increasing replication lag.

What is changed and how it works?

• Make Kafka sink verification create a missing topic and wait until it is visible in Kafka metadata before returning.
• Keep verification lightweight by using a one-shot topic initialization path that does not spawn unncessary resource.
• Start the runtime topic manager's metadata refresh only after the initial topic is ready.
• Make TiCDC's Kafka admin CreateTopic operation always create the topic. The Sarama adapter now owns the fixed validateOnly=false argument.

Check List

Tests

• Unit test

  go test --tags=intest ./downstreamadapter/sink/topicmanager \
    ./downstreamadapter/sink/kafka ./pkg/sink/kafka -count=1

Questions

Will it cause performance regression or break compatibility?

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

Release note

Please refer to [Release Notes Language Style Guide](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/release-notes-style-guide.html) to write a quality release note.

If you don't think this PR needs a release note then fill it with `None`.

Summary by CodeRabbit

Summary by CodeRabbit

• New Features

  • Kafka topics now use a streamlined creation and verification process.
  • Topic visibility is confirmed before use, improving setup reliability.
  • Metadata refresh begins only after a topic is created or detected.

• Bug Fixes

  • Improved handling of existing topics and topic-creation errors.
  • Topic configuration validation, including replication requirements, remains enforced.
  • Simplified topic creation behavior provides more consistent Kafka administration.

[3AceShowHand]

/test all

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 88ec0926-2e53-4cee-b32a-046223f31009

📥 Commits

Reviewing files that changed from the base of the PR and between 631f259 and 7ab9a92.

📒 Files selected for processing (1)

• downstreamadapter/sink/topicmanager/kafka_topic_manager.go

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

• downstreamadapter/sink/topicmanager/kafka_topic_manager.go

---

📝 Walkthrough

Walkthrough

The Kafka sink now delegates topic verification and creation to topicmanager.EnsureTopic. Topic creation uses a simplified admin API without validateOnly. Topic manager refresh starts after topic visibility is confirmed.

Changes

Kafka topic management

Layer / File(s)	Summary
Simplified topic creation contract pkg/sink/kafka/cluster_admin_client.go, pkg/sink/kafka/admin.go, pkg/sink/kafka/cluster_admin_client_mock.go, pkg/sink/kafka/*_test.go	CreateTopic now accepts only TopicDetail. Existing-topic errors remain ignored, and other errors remain wrapped.
Centralized topic ensuring downstreamadapter/sink/kafka/sink.go, downstreamadapter/sink/topicmanager/kafka_topic_manager.go	Verify calls EnsureTopic. EnsureTopic creates the topic and waits for visibility. Background metadata refresh starts after successful topic handling.
Topic manager lifecycle coverage downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go	Tests cover topic visibility, refresh initialization, replication validation, authorization scenarios, and the simplified admin API.

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

Sequence Diagram(s)

sequenceDiagram
  participant Verify
  participant EnsureTopic
  participant ClusterAdminClient
  Verify->>EnsureTopic: ensure topic with topic configuration
  EnsureTopic->>ClusterAdminClient: create topic
  ClusterAdminClient-->>EnsureTopic: return creation result
  EnsureTopic->>ClusterAdminClient: check topic visibility
  ClusterAdminClient-->>EnsureTopic: return topic metadata
  EnsureTopic-->>Verify: return result

Loading

Possibly related PRs

• pingcap/ticdc#5818: Updates Kafka topic visibility and creation handling.
• pingcap/ticdc#5715: Modifies Kafka topic verification and creation flow.
• pingcap/ticdc#5696: Modifies Kafka topic-manager creation and visibility handling.

Suggested reviewers: wk989898, asddongmen, lidezhu

Poem

A rabbit checks the topic gate,
Creates it, then waits in state.
Refresh starts when visibility is clear,
One detail guides each create call here.
Hop, hop—Kafka flows tonight!

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 0.00% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly summarizes the primary change: Kafka verification creates the topic and waits for visibility.
Description check	✅ Passed	The description identifies issue #5896, explains the change, lists unit tests, and includes the required sections, although some fields remain unanswered.
Linked Issues check	✅ Passed	The changes directly address issue #5896 by creating the Kafka topic with configured settings before verification completes.
Out of Scope Changes check	✅ Passed	The modified production code, APIs, mocks, and tests are directly related to Kafka topic creation and verification.

✨ 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.}

[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 `@downstreamadapter/sink/topicmanager/kafka_topic_manager.go`:
- Around line 74-76: The topic manager must not report success from EnsureTopic
until the created topic is confirmed visible. Update the flow around
CreateTopicAndWaitUntilVisible in
downstreamadapter/sink/topicmanager/kafka_topic_manager.go:74-76 to return an
error when visibility is not achieved, including denied-create cases. In
downstreamadapter/sink/kafka/sink.go:126-129, preserve propagation of this
strict readiness error from EnsureTopic without converting it to success.

🪄 Autofix

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: 1c5431db-aad9-4356-98b0-a6f0f99011f1

📥 Commits

Reviewing files that changed from the base of the PR and between 03d1fe2 and 631f259.

📒 Files selected for processing (8)

• downstreamadapter/sink/kafka/sink.go
• downstreamadapter/sink/topicmanager/kafka_topic_manager.go
• downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go
• pkg/sink/kafka/admin.go
• pkg/sink/kafka/admin_test.go
• pkg/sink/kafka/cluster_admin_client.go
• pkg/sink/kafka/cluster_admin_client_mock.go
• pkg/sink/kafka/options_test.go

[3AceShowHand]

/test all

[ti-chi-bot]

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: lidezhu, 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 [lidezhu,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-08-06 09:17:30.969220184 +0000 UTC m=+2692437.005315260: ☑️ agreed by lidezhu.
• 2026-08-07 02:51:53.790320571 +0000 UTC m=+2755699.826415617: ☑️ agreed by wk989898.

[ti-chi-bot]

In response to a cherrypick label: new pull request created to branch release-8.5: #5906.