kafka: avoid create changefeed failures if can't get a topic from broker

[wk989898]

What problem does this PR solve?

Issue Number: close #5563

What is changed and how it works?

If ticdc can't get the topic from kafka broker, still create a changefeed instead of throwing an error.

Check List

Tests

• Unit test
• Integration test
• Manual test (add detailed scripts or steps below)

Target	Startup mode	changefeed create result	Query after create	Runtime write test	Notes
master	Default, old arch	Failed, exit code 1	ErrChangeFeedNotExists	Not reached	CreateTopicAndWaitUntilVisible treated the topic as unavailable and attempted CreateTopic; Kafka returned Topic authorization failure
Current branch	Default, old arch	Failed, exit code 1	ErrChangeFeedNotExists	Not reached	Same visible behavior after rebuilding bin/cdc from acl-0720 @ 6225925f; this run did not use the modified downstreamadapter implementation
Current branch	cdc server --newarch	Success, exit code 0	state: normal	state: warning, ErrKafkaSendMessage after DDL	This run exercised the modified downstreamadapter implementation; create passed by skipping topic creation after authorization failure, but runtime write still requires Topic Write

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

• Bug Fixes
  • Improved handling of Kafka authorization failures during topic discovery and creation.
  • Services can now continue using the configured partition count when topic administration is denied.
  • Topic partition information remains available for subsequent operations.
• Tests
  • Added coverage for authorization failures during topic metadata retrieval and topic creation.

[coderabbitai]

📝 Walkthrough

Walkthrough

Kafka topic management now recognizes Kafka admin authorization failures during metadata lookup and topic creation, falls back to the configured partition count, and caches that value. Tests add authorization-failure mocks and coverage for the describe and create paths.

Changes

Kafka authorization fallback

Layer / File(s)	Summary
Kafka authorization error detection pkg/sink/kafka/admin.go	Adds IsAdminAuthorizationFailed for Sarama topic and cluster authorization errors.
Topic manager authorization fallback downstreamadapter/sink/topicmanager/kafka_topic_manager.go, downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go	Handles authorization failures during metadata retrieval and creation, stores configured partitions in the topic cache, and adds mocks and tests for these paths.

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

Sequence Diagram(s)

sequenceDiagram
  participant KafkaTopicManager
  participant KafkaClusterAdmin
  participant TopicsCache
  KafkaTopicManager->>KafkaClusterAdmin: GetTopicsMeta
  KafkaClusterAdmin-->>KafkaTopicManager: Authorization failure
  KafkaTopicManager->>TopicsCache: Store configured partition count
  KafkaTopicManager->>KafkaClusterAdmin: CreateTopic
  KafkaClusterAdmin-->>KafkaTopicManager: Authorization failure
  KafkaTopicManager->>TopicsCache: Store configured partition count

Loading

Suggested reviewers: 3aceshowhand

Poem

A rabbit hops through Kafka’s gate,
Finds ACLs that make it wait.
“No topic bloom? Then here’s the plan—
Use configured partitions where we can!”
The cache records, and hops away.

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Description check	⚠️ Warning	The PR follows the template structure, but the required release note is still placeholder text instead of a real note or None.	Replace the release note placeholder with an actual release note or None, and consider answering the compatibility and documentation questions.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Linked Issues check	✅ Passed	The change matches #5563 by falling back to configured values on Kafka authorization failures while preserving normal failures for invalid configs.
Out of Scope Changes check	✅ Passed	The code changes and tests are all directly related to handling Kafka admin authorization failures and topic caching.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.
Title check	✅ Passed	The title matches the main change: skipping Kafka topic-creation failures when the broker topic cannot be fetched.

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

[wk989898]

/test all

[coderabbitai]

Actionable comments posted: 1

🧹 Nitpick comments (1)

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

299-307: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low value

Avoid masking non-authorization and non-existence errors.

When GetTopicsMeta(..., false) returns an error, the current logic ignores it unless it's an authorization failure, allowing the flow to proceed to createTopic. While this is correct for sarama.ErrUnknownTopicOrPartition, falling through on other errors (like network timeouts) will cause an unnecessary createTopic attempt that will likely fail and obscure the original context.

Consider explicitly returning unexpected errors to prevent masking them.

♻️ Proposed refactor

 	topicDetails, err = m.admin.GetTopicsMeta([]string{topicName}, false)
 	if err != nil {
 		if kafka.IsAdminAuthorizationFailed(err) {
 			return m.useConfiguredPartitionNum(topicName, err), nil
 		}
+		if !errors.Is(err, sarama.ErrUnknownTopicOrPartition) {
+			return 0, errors.Trace(err)
+		}
 	} else if numPartition, ok := m.tryStoreTopicMeta(topicName, topicDetails); ok {
 		return numPartition, nil
 	}

🤖 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 299
- 307, Update the GetTopicsMeta error handling in the topic manager flow:
preserve the existing useConfiguredPartitionNum behavior for authorization
failures and allow sarama.ErrUnknownTopicOrPartition to continue toward topic
creation, but return all other errors immediately instead of falling through to
createTopic.

🤖 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_test.go`:
- Around line 261-263: In TestCreateTopicWithCreateDenied, replace the
mockAdminClientWithDeniedDescribe instantiation with
mockAdminClientWithDeniedCreate so GetTopicsMeta succeeds and the test reaches
the createTopic authorization-denied path, preserving the createTopicCalled
assertion.

---

Nitpick comments:
In `@downstreamadapter/sink/topicmanager/kafka_topic_manager.go`:
- Around line 299-307: Update the GetTopicsMeta error handling in the topic
manager flow: preserve the existing useConfiguredPartitionNum behavior for
authorization failures and allow sarama.ErrUnknownTopicOrPartition to continue
toward topic creation, but return all other errors immediately instead of
falling through to createTopic.

🪄 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

Run ID: f5d4a68c-a7eb-4e7e-b2fe-b82fef6bf742

📥 Commits

Reviewing files that changed from the base of the PR and between eceaf65 and 3437e95.

📒 Files selected for processing (3)

• downstreamadapter/sink/topicmanager/kafka_topic_manager.go
• downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go
• pkg/sink/kafka/admin.go

[wk989898]

/test all

[ti-chi-bot]

[APPROVALNOTIFIER] This PR is APPROVED

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

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,lidezhu]

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-20 10:38:39.244227348 +0000 UTC m=+1228505.280322404: ☑️ agreed by lidezhu.
• 2026-07-20 10:50:53.383826129 +0000 UTC m=+1229239.419921175: ☑️ agreed by asddongmen.

[wk989898]

/retest

[ti-chi-bot]

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