kafka: improve stability when creating many topics with Kafka v4 (#6081)

[ti-chi-bot]

This is an automated cherry-pick of #6081

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.

[ti-chi-bot]

@3AceShowHand This PR has conflicts, I have hold it.
Please resolve them or ask others to resolve them, then comment /unhold to remove the hold label.

[ti-chi-bot]

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by:
Once this PR has been reviewed and has the lgtm label, please assign flowbehappy for approval. For more information see the Code Review Process.
Please ensure that each of them provides their approval before proceeding.

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

Details

Needs approval from an approver in each of these files:

• OWNERS

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]

@ti-chi-bot: ## If you want to know how to resolve it, please read the guide in TiDB Dev Guide.

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 ti-community-infra/tichi repository.

[coderabbitai]

Important

Review skipped

Auto reviews are disabled on base/target branches other than the default branch.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 32575277-120a-4cb2-a15a-bc42c1d156dd

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

---

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