kafka: improve stability when creating many topics with Kafka v4

[3AceShowHand]

What problem does this PR solve?

Issue Number: close #6076

Kafka may acknowledge topic creation before every broker can return the new topic in metadata. This visibility window becomes easier to hit when a changefeed dynamically creates thousands of topics.

TiCDC previously suppressed UnknownTopicOrPartition from GetTopicsMeta even when the caller requested strict error handling. The topic manager could then read a missing metadata entry as zero partitions, cache the topic as ready, and start producing too early. The producer failure triggered repeated sink recovery and backoff, which could stall the changefeed checkpoint.

What is changed and how it works?

• Make GetTopicsMeta return topic-level errors, including UnknownTopicOrPartition, when ignoreTopicError is false. Discovery calls with ignoreTopicError=true continue to return metadata for valid topics.
• Retry temporary topic, broker, controller, and network metadata errors while waiting for a newly created topic to become visible. Authorization, authentication, configuration, and invalid-request errors fail immediately. Retries keep the existing six-attempt bound.
• Cache a topic and its partition count only after Kafka returns metadata for that topic.
• Mark the sink abnormal and close its input queues before releasing Kafka resources.
• Reject DML events after shutdown begins and honor cancellation before forwarding partitioned rows.

Together, these changes keep topic creation in the metadata-validation phase until Kafka confirms visibility and prevent the sink from accepting new DML events after shutdown begins.

Check List

Tests

• Unit test
  • GOTOOLCHAIN=go1.25.12 GOMAXPROCS=2 go test -p 1 -vet=off ./pkg/sink/kafka ./downstreamadapter/sink/topicmanager -count=1
  • Covers topic metadata error handling, retry classification, and cache timing.
  • This branch also adds unit tests for sink shutdown ordering and post-close event handling.

Questions

Will it cause performance regression or break compatibility?

Newly created topics may wait for up to the existing six metadata attempts during Kafka's visibility window; deterministic errors still return after the first attempt.

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

Release note

Fix Kafka changefeeds that can stall during large-scale dynamic topic creation when newly created topics are temporarily absent from broker metadata.

[coderabbitai]

Note

Reviews paused

It looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

• @coderabbitai resume to resume automatic reviews.
• @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

• ▶️ Resume reviews
• 🔍 Trigger review

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: ce753b33-0e51-4ef5-8d72-936db2ba35f4

📥 Commits

Reviewing files that changed from the base of the PR and between f6a8200 and 4a42703.

📒 Files selected for processing (11)

• downstreamadapter/sink/kafka/helper.go
• downstreamadapter/sink/kafka/sink.go
• downstreamadapter/sink/kafka/sink_test.go
• pkg/sink/kafka/admin.go
• pkg/sink/kafka/sarama_admin_mock.go
• pkg/sink/kafka/sarama_admin_test.go
• pkg/sink/kafka/sarama_async_producer.go
• pkg/sink/kafka/sarama_factory.go
• pkg/sink/kafka/sarama_sync_producer.go
• pkg/sink/kafka/sarama_sync_producer_mock.go
• pkg/sink/kafka/sarama_sync_producer_test.go

💤 Files with no reviewable changes (3)

• pkg/sink/kafka/sarama_sync_producer_mock.go
• pkg/sink/kafka/sarama_sync_producer.go
• pkg/sink/kafka/sarama_admin_mock.go

Included review availability: Your plan provides up to 1 included review per hour; 0 remain after this review.

---

📝 Walkthrough

Walkthrough

Kafka metadata handling now distinguishes retryable and unretryable errors. Topic visibility controls cache timing. The Kafka factory reuses one client, and sink shutdown follows an ordered lifecycle.

Changes

Kafka sink coordination

Layer / File(s)	Summary
Apply metadata policies and visibility retries pkg/sink/kafka/admin.go, pkg/sink/kafka/sarama_admin_test.go, downstreamadapter/sink/topicmanager/...	Unknown-topic errors follow normal handling. IsUnretryableTopicMetadataError classifies non-retryable errors. Topic visibility retries other metadata errors and updates the cache only after success. Tests cover these paths.
Reuse the shared Sarama client pkg/sink/kafka/sarama_factory.go, pkg/sink/kafka/sarama_*producer.go, pkg/sink/kafka/sarama_*test.go, pkg/sink/kafka/*mock.go	The factory shares one Sarama client with the admin and producers. Producer wrappers no longer close the shared client. Related mocks and shutdown tests were updated.
Order sink cleanup and terminal state downstreamadapter/sink/kafka/helper.go, downstreamadapter/sink/kafka/sink.go, downstreamadapter/sink/kafka/sink_test.go	Cleanup closes shared components before producer facades. The sink becomes abnormal before shutdown, rejects later DML events, and checks cancellation before row delivery.

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

Merge Risk: ⚪ Minimal · up to 4a427

This change is merge-ready after normal checks and review; no actionable merge-blocking risk remains.

Sequence Diagram(s)

sequenceDiagram
  participant TopicManager
  participant KafkaAdmin
  participant TopicCache
  TopicManager->>KafkaAdmin: Create topic
  loop Until metadata is visible
    TopicManager->>KafkaAdmin: GetTopicsMeta
    KafkaAdmin-->>TopicManager: Metadata or classified error
  end
  TopicManager->>TopicCache: Store partition count

Loading

sequenceDiagram
  participant Sink
  participant TopicManager
  participant saramaAdminClient
  participant Producers
  Sink->>Sink: Mark abnormal
  Sink->>TopicManager: Close
  Sink->>saramaAdminClient: Close admin and shared client
  Sink->>Producers: Close producer facades

Loading

Suggested reviewers: wk989898

Poem

A rabbit checks the topic trail,

Retries what may yet prevail.
Shared clients close in ordered flight,
Channels rest when state turns quiet.
The cache waits for visible light.

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 15.79% which is insufficient. The required threshold is 80.00%. Docstring coverage is scoped to functions touched by this diff. Analyzed 19 functions across 10 files.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Linked Issues check	✅ Passed	The changes address [#6076] by retrying temporary Kafka metadata errors, stopping promptly on unretryable errors, propagating topic-level failures, and updating the cache only after topic visibility i…
Out of Scope Changes check	✅ Passed	The shared-client lifecycle, shutdown ordering, closed-sink checks, and related test updates support Kafka stability, error handling, and resource cleanup. No unrelated code changes are evident.
Title check	✅ Passed	The title clearly and concisely describes the main Kafka stability improvement for large-scale topic creation.
Description check	✅ Passed	The description covers the issue, problem, implementation, tests, compatibility impact, and release note. The documentation question is left unanswered, but this is non-critical.

Full details: Linked Issues check

Explanation

The changes address [#6076] by retrying temporary Kafka metadata errors, stopping promptly on unretryable errors, propagating topic-level failures, and updating the cache only after topic visibility is confirmed.

✨ 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

[Copilot]

Pull request overview

This PR improves Kafka topic-creation robustness in TiCDC’s Kafka sink by treating newly-created topics as potentially not yet visible in metadata, retrying transient Kafka metadata errors, and only caching a topic after Kafka confirms it is visible.

Changes:

• Update GetTopicsMeta to return topic-level errors (including UnknownTopicOrPartition) when topic errors are not ignored, while preserving the previous “best-effort discovery” behavior when errors are ignored.
• Introduce IsRetryableTopicMetadataError to classify transient Kafka metadata errors as retryable and use it to drive bounded retries when waiting for topic visibility.
• Ensure the topic manager’s cache is only updated after the topic becomes visible (and add/adjust unit tests to validate caching + retry behavior).

Validation:

• Not run here (code review only; no CI execution available in this environment).

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated no comments.

File	Description
pkg/sink/kafka/admin.go	Makes topic metadata error handling stricter when not ignoring topic errors; adds retryable-metadata error classifier.
pkg/sink/kafka/sarama_admin_test.go	Adds unit coverage for the stricter unknown-topic error behavior and for retryable metadata error classification.
downstreamadapter/sink/topicmanager/kafka_topic_manager.go	Adds retryable-error classification to topic visibility wait; caches topic only after visibility.
downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go	Adds tests ensuring cache isn’t written before visibility and retry behavior honors cancellation / stops on non-retryable errors.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[3AceShowHand]

/test all

[3AceShowHand]

/retest

[3AceShowHand]

/unhold

[3AceShowHand]

/hold

[3AceShowHand]

/retest

[3AceShowHand]

/retest

[3AceShowHand]

/unhold

[3AceShowHand]

/retest

[3AceShowHand]

/test all

[3AceShowHand]

/test pull-cdc-pulsar-integration-heavy

[3AceShowHand]

/test pull-cdc-pulsar-integration-heavy

[3AceShowHand]

/test all

[3AceShowHand]

/retest

[ti-chi-bot]

In response to a cherrypick label: new pull request created to branch release-nextgen-202603: #6111.
But this PR has conflicts, please resolve them!

[ti-chi-bot]

@3AceShowHand: The following test failed, say /retest to rerun all failed tests or /retest-required to rerun all mandatory failed tests:

Test name	Commit	Details	Required	Rerun command
pull-cdc-mysql-integration-light-next-gen-legacy-safepoint	b9d3b04	link	unknown	/test pull-cdc-mysql-integration-light-next-gen-legacy-safepoint

Full PR test history. Your PR dashboard.

Details

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. I understand the commands that are listed here.