kafka: share one claimcheck instance across encoders

[3AceShowHand]

What problem does this PR solve?

When Kafka claim-check handling is enabled, every event encoder creates its own ClaimCheck and external storage client. Because an encoder group contains multiple encoders, the number of storage clients grows with encoder concurrency even though they all use the same configuration and storage URI.

Issue Number: close #5719

What is changed and how it works?

• Create one ClaimCheck at the Kafka sink component level.
• Inject the same instance into the encoder group and the standalone encoder.
• Close the underlying claim-check storage when the Kafka sink component is closed.
• Remove encoder-owned claim-check metric cleanup because ownership now belongs to the Kafka sink component.
• Update codec and claim-check tests to use the shared instance and verify concurrent writes and storage cleanup.

Pulsar does not enable claim-check handling and continues to pass no ClaimCheck to its encoders.

Check List

Tests

• Unit test

  go test --tags=intest ./pkg/sink/kafka/claimcheck ./pkg/sink/codec/open ./pkg/sink/codec/canal ./pkg/sink/codec/simple ./downstreamadapter/sink/kafka ./downstreamadapter/sink/pulsar -count=1
  go test -race --tags=intest ./pkg/sink/kafka/claimcheck -count=1
  

Questions

Will it cause performance regression or break compatibility?

It reduces external storage client and connection usage when Kafka claim-check is enabled. It changes exported Go encoder constructor signatures to accept a sink-owned ClaimCheck, but does not change sink configuration, encoded message formats, or runtime protocol compatibility.

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

No user-facing documentation or monitoring changes are required because configuration and behavior remain unchanged.

Release note

None

Summary by CodeRabbit

• New Features

  • Improved handling of oversized Kafka messages through shared claim-check storage across supported encoders.
  • Added safer resource lifecycle management for large-message storage.

• Bug Fixes

  • Improved concurrent large-message writes and cleanup behavior.
  • Prevented resource leaks when sink or encoder initialization fails.

• Tests

  • Expanded coverage for concurrent writes, storage closure, and large-message encoding.

[coderabbitai]

📝 Walkthrough

Walkthrough

Claim-check ownership moves to sink setup, codec constructors accept shared instances, and encoder-level cleanup methods are removed. Kafka and codec tests update constructor calls, while claim-check closure and concurrent writes gain coverage.

Changes

Claim-check lifecycle

Layer / File(s)	Summary
Sink ownership and cleanup downstreamadapter/sink/kafka/helper.go, downstreamadapter/sink/kafka/sink.go	Kafka sink setup creates, shares, and closes ClaimCheck instances, including failure-path cleanup and statistics closure.
Codec contracts and wiring pkg/sink/codec/*	Encoder constructors receive external claim-check instances, and the Clean() interface and implementations are removed.
Constructor call-site updates downstreamadapter/sink/*, pkg/sink/codec/**/*_test.go	Kafka, Pulsar, Canal JSON, Open, and Simple encoder construction is updated to pass nil or initialized claim-check instances.
Claim-check storage validation pkg/sink/kafka/claimcheck/*	ClaimCheck.Close() replaces metric cleanup, and tests cover nil safety, storage closure, and concurrent writes.

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

Possibly related PRs

• pingcap/ticdc#5475: Overlaps in the encoder factory path and protocol dispatch.
• pingcap/ticdc#5617: Overlaps in Kafka sink verification and encoder cleanup.
• pingcap/ticdc#5715: Overlaps in Kafka verification and claim-check validation.

Suggested labels: release-note, size/XXL, do-not-merge/needs-linked-issue

Suggested reviewers: wk989898, asddongmen

Poem

A rabbit wired a check with care,
Passing it through codecs everywhere.
Close the store when work is done,
Tests race together, one by one.
Clean paths hop beneath the moon.

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 6.59% 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 matches the main change: sharing one claimcheck instance across Kafka encoders.
Description check	✅ Passed	The description follows the required template and includes the issue link, 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

[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/pulsar/helper.go`:
- Around line 133-138: The Pulsar component setup around NewEncoderGroup and
NewEventEncoder must preserve claim-check support instead of always passing nil.
Create and retain a claim-check handler in the Pulsar component when
encoderConfig enables it, pass the same handler to both constructors, and close
it on setup failure and component shutdown; alternatively, explicitly reject
claim-check configuration for Pulsar.

🪄 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: 8c5d2079-de1e-4c1f-b227-42287faee569

📥 Commits

Reviewing files that changed from the base of the PR and between d480b05 and c603b16.

📒 Files selected for processing (19)

• downstreamadapter/sink/kafka/helper.go
• downstreamadapter/sink/kafka/sink.go
• downstreamadapter/sink/kafka/sink_test.go
• downstreamadapter/sink/pulsar/helper.go
• pkg/sink/codec/avro/arvo.go
• pkg/sink/codec/bootstraper.go
• pkg/sink/codec/builder.go
• pkg/sink/codec/canal/canal_json_encoder.go
• pkg/sink/codec/canal/canal_json_encoder_test.go
• pkg/sink/codec/canal/canal_json_test.go
• pkg/sink/codec/common/encoder.go
• pkg/sink/codec/debezium/encoder.go
• pkg/sink/codec/encoder_group.go
• pkg/sink/codec/open/encoder.go
• pkg/sink/codec/open/encoder_test.go
• pkg/sink/codec/simple/encoder.go
• pkg/sink/codec/simple/encoder_test.go
• pkg/sink/kafka/claimcheck/claim_check.go
• pkg/sink/kafka/claimcheck/claim_check_test.go

💤 Files with no reviewable changes (4)

• pkg/sink/codec/common/encoder.go
• pkg/sink/codec/avro/arvo.go
• pkg/sink/codec/bootstraper.go
• pkg/sink/codec/debezium/encoder.go

[3AceShowHand]

/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-24 10:44:25.296679396 +0000 UTC m=+1574451.332774462: ☑️ agreed by lidezhu.
• 2026-07-24 13:57:49.616492859 +0000 UTC m=+1586055.652587915: ☑️ agreed by asddongmen.

[3AceShowHand]

/unhold