kafka: validate sink config against existing topics

[3AceShowHand]

What problem does this PR solve?

Kafka sink verification creates the claim-check instance before checking topic metadata, but it initializes the event encoder only after the missing-topic branch. If the topic already exists, verification returns before encoder initialization and can miss errors such as an unavailable Avro Schema Registry.

The required-acks=all check also uses the configured replication-factor instead of the existing topic's actual replication factor.

Issue Number: close #5720

What is changed and how it works?

• Initialize the event encoder before connecting to Kafka or checking whether the topic exists.
• Keep claim-check creation and cleanup in the verification path and pass the instance to the encoder.
• Inline topic verification in Verify; no separate verifyKafkaSink helper is needed.
• Include the smallest partition replication factor in topic metadata.
• Validate min.insync.replicas against the actual replication factor for existing topics.
• Keep the dry-run topic creation check when the topic does not exist.

Check List

Tests

• Unit test

  go test --tags=intest ./downstreamadapter/sink/kafka -run '^(TestVerifyInvalidEncoderConfig|TestVerifyEncoderInitialization)$' -count=1
  go test --tags=intest ./pkg/sink/kafka -run '^(TestGetTopicsMetaUsesSmallestReplicationFactor|TestAdjustConfigMinInsyncReplicas)$' -count=1
  

Questions

Will it cause performance regression or break compatibility?

No compatibility changes. Existing-topic verification now initializes the encoder and checks its dependencies. For Avro, this includes the expected Schema Registry connectivity check during verification.

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

No. This fixes verification behavior without changing user-facing configuration or monitoring.

Release note

Fix Kafka sink verification for encoder dependencies and existing topic replication factors.

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

[coderabbitai]

Warning

Review limit reached

@3AceShowHand, you've reached your PR review limit, so we couldn't start this review.

Next review available in: 7 minutes

Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available.
You're only billed for reviews past your plan's rate limits ($0.25/file).

How can I continue?

After more reviews become available, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews.

How do review limits work?

CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability.

For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window.

Please refer docs for additional details.

Review details

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: cf89c2a1-eeb6-4b13-80a9-d6f5a12185cd

📥 Commits

Reviewing files that changed from the base of the PR and between 9b60fa6 and 403c637.

📒 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

📝 Walkthrough

Walkthrough

Kafka sink verification now validates existing topics through encoder construction, creates missing topics conditionally, and uses topic replication metadata when validating min.insync.replicas. Tests cover claim-check validation and minimum replication-factor calculation.

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 the admin client and configures the encoder; verifyKafkaSink conditionally creates missing topics and validates existing topics through encoder construction, including unsupported claim-check storage errors.
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 smallest partition replication factor, and min.insync.replicas validation uses that value for existing topics with updated test fixtures and assertions.

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

Sequence Diagram(s)

sequenceDiagram
  participant Verify
  participant AdminClient
  participant verifyKafkaSink
  participant EventEncoder
  Verify->>AdminClient: create admin client
  Verify->>verifyKafkaSink: pass topic and encoder configuration
  verifyKafkaSink->>AdminClient: fetch topic metadata
  verifyKafkaSink->>AdminClient: create missing topic when AutoCreate is enabled
  verifyKafkaSink->>EventEncoder: construct encoder and validate claim-check storage

Loading

Possibly related PRs

• pingcap/ticdc#5420: Both update Kafka encoder sizing configuration with message and batch limits.
• pingcap/ticdc#5617: Both refactor Kafka sink topic verification and encoder construction.
• pingcap/ticdc#5718: Both modify claim-check and event-encoder wiring in Kafka sink verification.

Suggested labels: lgtm, approved

Suggested reviewers: wk989898, asddongmen

Poem

I’m a rabbit who checks every topic in line,
With replicas counted, the numbers align.
Missing ones sprout when creation is true,
Claim-check paths get validation too.
Hop, hop—the Kafka flow is bright!

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 25.00% 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	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Title check	✅ Passed	The title clearly summarizes the main change: validating Kafka sink config against existing topics.
Description check	✅ Passed	The description includes the required issue line, change summary, 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

[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	403c637	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.