kafka: verify replication-factor when need to create the topic

[3AceShowHand]

What problem does this PR solve?

Issue Number: close #5720

Kafka sink Verify returned immediately after fetching metadata when the target topic already existed. This skipped NewEventEncoder, so encoder-owned checks—most notably Avro Schema Registry connectivity—were not run.

replication-factor is a topic creation option. The previous path could validate it even when the topic already existed or automatic topic creation was disabled, although TiCDC would not use it in either case.

What is changed and how it works?

• Preserve the initialization order from master and remove the early return for an existing topic. Encoder initialization now runs after Kafka verification for both existing and missing topics.
• Skip replication-factor validation and validate-only topic creation for existing topics.
• For a missing topic:
  • report that the topic does not exist when automatic topic creation is disabled;
  • when required-acks=all, compare the configured replication factor with the broker min.insync.replicas;
  • use Kafka's validate-only CreateTopic request during sink verification.
• Run the same replication-factor validation immediately before runtime topic creation, including dynamically routed topics.
• Reject a non-positive replication-factor while applying sink options.
• Keep broker configuration lookup local to the creation check; no existing-topic replication factor is fetched or validated.

Check List

Tests

• Unit test

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

Questions

Will it cause performance regression or break compatibility?

No. This changes verification and topic-creation validation only. It does not change a public API or protocol.

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

No.

Release note

None

[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

📝 Walkthrough

Walkthrough

Kafka sink verification now conditionally creates missing topics and always constructs the event encoder afterward. Kafka topic metadata includes the minimum partition replication factor, which is used when validating min.insync.replicas for existing topics.

Changes

Kafka sink validation

Layer / File(s)	Summary
Kafka sink verification flow downstreamadapter/sink/kafka/sink.go, downstreamadapter/sink/kafka/sink_test.go	Verify creates missing topics only when AutoCreate is enabled, continues to encoder construction for existing topics, and tests Schema Registry failures.
Replication-factor-aware option validation pkg/sink/kafka/admin.go, pkg/sink/kafka/admin_test.go, pkg/sink/kafka/options.go, pkg/sink/kafka/options_test.go	Topic metadata records the minimum partition replication factor, and required-acks=all validation uses existing-topic metadata with updated fixtures and assertions.

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

Sequence Diagram(s)

sequenceDiagram
  participant Verify
  participant KafkaAdmin
  participant EventEncoder
  Verify->>KafkaAdmin: fetch target topic metadata
  Verify->>KafkaAdmin: create missing topic when AutoCreate is enabled
  Verify->>EventEncoder: construct event encoder
  EventEncoder-->>Verify: return encoder or Schema Registry error

Loading

Suggested reviewers: asddongmen

Poem

I’m a rabbit counting replicas with care,
Checking each topic and encoder there.
Missing topics bloom when allowed to create,
Existing ones still validate their state.
Hop, hop—Kafka is ready and fair!

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 8.33% 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
Linked Issues check	✅ Passed	The changes cover the linked issue's needs: encoder failures surface, existing-topic replication uses metadata, and missing-topic dry-run creation remains.
Out of Scope Changes check	✅ Passed	The patch stays focused on Kafka sink verification and related Kafka metadata tests.
Title check	✅ Passed	The title is concise and matches the topic-creation/replication-factor part of the changes, though it omits the encoder fix.
Description check	✅ Passed	The description covers the required sections, 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

[3AceShowHand]

/test all

[3AceShowHand]

/retest

[3AceShowHand]

/test all

[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/kafka/sink.go`:
- Line 146: The Kafka sink must initialize the event encoder before any Kafka
interaction so Schema Registry errors are returned first. In
downstreamadapter/sink/kafka/sink.go lines 146-146, move codec.NewEventEncoder
after claim-check setup and before kafka.NewSaramaFactory/AdminClient or topic
metadata checks. In downstreamadapter/sink/kafka/sink_test.go lines 43-77,
update the test’s broker mock/assertions to fail if Kafka is contacted before
the Schema Registry error is returned.

🪄 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: 8f19d4d7-7a16-4bd0-9044-08cbdbd5fb59

📥 Commits

Reviewing files that changed from the base of the PR and between 9b60fa6 and 380b5e9.

📒 Files selected for processing (4)

• downstreamadapter/sink/kafka/sink.go
• downstreamadapter/sink/kafka/sink_test.go
• pkg/sink/kafka/admin_test.go
• pkg/sink/kafka/options.go

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

• pkg/sink/kafka/options.go

[3AceShowHand]

/test all

[ti-chi-bot]

[LGTM Timeline notifier]

Timeline:

• 2026-07-27 10:29:32.288423572 +0000 UTC m=+1832758.324518647: ☑️ agreed by wk989898.
• 2026-07-27 10:53:34.877584387 +0000 UTC m=+1834200.913679443: ☑️ agreed by asddongmen.

[3AceShowHand]

/test pull-cdc-kafka-integration-light

[ti-chi-bot]

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: asddongmen, hongyunyan, 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,hongyunyan,wk989898]

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

[3AceShowHand]

/cherry-pick release-8.5

[ti-chi-bot]

@3AceShowHand: new pull request created to branch release-8.5: #5813.
But this PR has conflicts, please resolve them!

Details

In response to this:

/cherry-pick release-8.5

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.