kafka: make the verify lightweight

[3AceShowHand]

What problem does this PR solve?

Issue Number: close #5618

Kafka sink Verify reused the normal sink construction path. That path creates runtime components such as the encoder group and topic manager, and can perform startup-only topic creation work. Verification should stay lightweight: validate user configuration, construct only cheap local helpers, and use Kafka admin APIs only for read-only or validate-only checks.

What is changed and how it works?

• Reworks Kafka sink Verify to run the shared parse/config checks directly:
  • GetProtocol
  • GetTopic
  • options.Apply
  • GetEncoderConfig(...).Validate()
• Keeps only the lightweight local construction checks needed by Verify:
  • eventrouter.NewEventRouter
  • columnselector.New
  • one codec.NewEventEncoder call to reject protocols that Kafka sink does not support
• Changes Kafka topic verification to use admin metadata plus validate-only creation:
  • existing topic: pass after GetTopicsMeta
  • missing topic with auto-create-topic=false: return a config error
  • missing topic with auto-create-topic=true: call CreateTopic(..., validateOnly=true) to validate parameters and permissions without creating the topic
• Leaves the real sink startup path unchanged, so actual topic creation still happens in New / topic manager startup.
• Removes the now-unneeded test-only Kafka factory creator indirection.

Check List

Tests

• Unit test

Commands run locally:

go test ./downstreamadapter/sink/kafka -run TestVerifyRejectsKafkaUnsupportedProtocolBeforeConnecting -count=1
go test --tags=intest ./downstreamadapter/sink/kafka -count=1
git diff --check

Questions

Will it cause performance regression or break compatibility?

No performance regression is expected. This makes Verify lighter and removes startup-only side effects from verification. Sink startup behavior and real topic creation remain unchanged.

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

No. This changes internal verification behavior only; no user-facing config, protocol, metric, or documentation surface changes.

Release note

None

Summary by CodeRabbit

• Bug Fixes
  • Kafka sink verification now validates protocol support, configuration, encoder setup, and topic readiness before connecting.
  • Kafka topics are created when missing and auto-creation is enabled.
  • Kafka consumer partition discovery now rejects topics with metadata errors.
  • Unsupported protocols are reported clearly before connection attempts.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 977ba792-78c6-4864-a5d6-112cecc54df9

📥 Commits

Reviewing files that changed from the base of the PR and between 9dc0839 and 709fe63.

📒 Files selected for processing (3)

• cmd/kafka-consumer/consumer.go
• downstreamadapter/sink/kafka/helper.go
• pkg/sink/kafka/factory.go

💤 Files with no reviewable changes (1)

• pkg/sink/kafka/factory.go

---

📝 Walkthrough

Walkthrough

Kafka sink verification now performs explicit configuration, encoder, Kafka admin, and topic readiness checks, while component construction uses a direct Sarama factory and consumer metadata handling validates Kafka error codes.

Changes

Kafka sink verification and Kafka integration

Layer / File(s)	Summary
Explicit verification and protocol rejection downstreamadapter/sink/kafka/sink.go, downstreamadapter/sink/kafka/sink_test.go	Verify validates protocol and topic settings, initializes encoding components, checks Kafka topic metadata, conditionally creates missing topics, and tests rejection of CSV protocol configuration.
Direct Kafka factory construction downstreamadapter/sink/kafka/helper.go, pkg/sink/kafka/factory.go	Kafka sink construction directly creates the Sarama factory, and the unused exported FactoryCreator type is removed.
Kafka metadata error validation cmd/kafka-consumer/consumer.go	Partition discovery now requires topic metadata to report ErrNoError before using the topic details.

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

Sequence Diagram(s)

sequenceDiagram
  participant Verify
  participant EventRouter
  participant ColumnSelector
  participant SaramaFactory
  participant KafkaAdmin
  Verify->>EventRouter: initialize routing for Avro-like protocols
  Verify->>ColumnSelector: initialize column selection
  Verify->>Verify: create and clean event encoder
  Verify->>SaramaFactory: create Kafka admin client
  SaramaFactory->>KafkaAdmin: obtain topic metadata
  KafkaAdmin-->>Verify: return topic status
  Verify->>KafkaAdmin: create missing topic when enabled

Loading

Possibly related PRs

• pingcap/ticdc#5475: Both changes touch Kafka sink initialization and Avro-like protocol handling.

Suggested reviewers: wk989898, lidezhu, tenfyzhong

Poem

I’m a rabbit with a Kafka tune,
Checking topics by the moon.
Bad protocols hop away,
Good encoders cleanly stay.
New topics bloom when rules say “go”—
Verify keeps the stream in flow.

🚥 Pre-merge checks | ✅ 3 | ❌ 2

❌ Failed checks (2 warnings)

Check name	Status	Explanation	Resolution
Out of Scope Changes check	⚠️ Warning	The consumer metadata error-code change in cmd/kafka-consumer/consumer.go is unrelated to the verify simplification scope.	Split that consumer fix into a separate PR or add linked-issue justification if it must stay here.
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 (3 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly summarizes the main change: making Kafka verify lightweight.
Description check	✅ Passed	The PR description matches the template with issue number, problem, change summary, tests, questions, and release note.
Linked Issues check	✅ Passed	The changes implement the requested Kafka verify simplification in #5618.

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

[gemini-code-assist]

Code Review

This pull request refactors the Verify function in the Kafka sink to perform comprehensive validation of configurations (such as protocol, topic, options, encoder, event router, and column selector) without initializing a full sink component, and adds a corresponding unit test. The review feedback suggests two important improvements to the validation logic: first, ensuring that the user-specified partition count does not exceed the actual partition count of an existing topic; and second, defaulting the partition count to 3 during dry-run topic creation validation if it is unspecified (<= 0) to prevent Kafka broker validation failures.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[3AceShowHand]

/test all

[ti-chi-bot]

[LGTM Timeline notifier]

Timeline:

• 2026-07-10 09:08:40.292192128 +0000 UTC m=+359106.328287174: ☑️ agreed by wk989898.
• 2026-07-10 09:20:59.753470762 +0000 UTC m=+359845.789565839: ☑️ agreed by hongyunyan.

[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]

/retest

[3AceShowHand]

/retest

[3AceShowHand]

/test all

[ti-chi-bot]

@3AceShowHand: The following tests 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	709fe63	link	false	/test pull-cdc-mysql-integration-light-next-gen
pull-unit-test-next-gen	709fe63	link	unknown	/test pull-unit-test-next-gen
pull-unit-test	709fe63	link	unknown	/test pull-unit-test
pull-cdc-mysql-integration-light	709fe63	link	true	/test pull-cdc-mysql-integration-light

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.