pulsar: support debezium protocol

[HGHNice]

What problem does this PR solve?

Issue Number: close #5056

The Pulsar sink currently only supports canal-json. Users who consume
TiCDC events via Pulsar with Debezium-compatible consumers (e.g. Flink CDC)
have no way to use the standard Debezium message format.

What is changed and how it works?

• Extend IsPulsarSupportedProtocols() in pkg/config/sink_protocol.go
  to include ProtocolDebezium.
• Update the error message in downstreamadapter/sink/pulsar/helper.go
  to reflect the expanded protocol list.
• The Debezium codec is already fully implemented and shared with the Kafka
  sink via the common codec builder, so no additional encoding logic is needed.
• Add unit test TestIsPulsarSupportedProtocols.

Check List

Tests

• Unit test

Questions

Will it cause performance regression or break compatibility?

No. Existing canal-json behavior is unchanged.

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

The Pulsar sink docs should note that debezium is now a valid protocol value.

Release note

The Pulsar sink now supports the `debezium` protocol. Set `protocol=debezium`
in the Pulsar sink URI to produce Debezium-format change events.


## Summary by CodeRabbit

## Release Notes

* **New Features**
  * Pulsar sink now supports debezium protocol in addition to canal-json protocol.

* **Tests**
  * Added validation tests for Pulsar sink protocol compatibility.

<!-- review_stack_entry_start -->

[![Review Change Stack](https://storage.googleapis.com/coderabbit_public_assets/review-stack-in-coderabbit-ui.svg)](https://app.coderabbit.ai/change-stack/pingcap/ticdc/pull/5054)

<!-- review_stack_entry_end -->

Summary by CodeRabbit

• New Features

  • Pulsar sink now supports Debezium protocol format in addition to Canal-JSON.

• Tests

  • Added test cases for protocol parsing and Pulsar-supported protocol validation.
  • Extended Debezium integration tests to validate both Kafka and Pulsar sink types.

[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 flowbehappy 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
• pkg/config/OWNERS

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]

Hi @HGHNice. Thanks for your PR.

I'm waiting for a pingcap member to verify that this patch is reasonable to test. If it is, they should reply with /ok-to-test on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should join the org to skip this step.

Once the patch is verified, the new status will be reflected by the ok-to-test label.

I understand the commands that are listed here.

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.

[ti-chi-bot]

Welcome @HGHNice!

It looks like this is your first PR to pingcap/ticdc 🎉.

I'm the bot to help you request reviewers, add labels and more, See available commands.

We want to make sure your contribution gets all the attention it needs!



Thank you, and welcome to pingcap/ticdc. 😃

[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: e75e270d-43b6-449b-bb2f-69bd090656e2

📥 Commits

Reviewing files that changed from the base of the PR and between 46882d4 and 4495deb.

📒 Files selected for processing (3)

• downstreamadapter/sink/pulsar/helper.go
• pkg/config/sink_protocol_test.go
• tests/integration_tests/debezium_basic/run.sh

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

• pkg/config/sink_protocol_test.go

---

📝 Walkthrough

Walkthrough

Pulsar sink now accepts debezium in addition to canal-json. Validation and tests updated, Pulsar helper error text and router initialization adjusted, and the debezium_basic integration script branches to run against kafka or pulsar sinks.

Changes

Pulsar Debezium Protocol Support

Layer / File(s)	Summary
Protocol validation and unit tests pkg/config/sink_protocol.go, pkg/config/sink_protocol_test.go	IsPulsarSupportedProtocols now returns true for ProtocolCanalJSON and ProtocolDebezium. Tests added/expanded for parsing debezium and simple, and a table-driven test verifies supported protocols.
Pulsar helper updates downstreamadapter/sink/pulsar/helper.go	The unsupported-protocol error message now lists [canal-json, debezium]. Event router construction passes protocol == config.ProtocolAvro instead of a hard-coded flag and removes the obsolete inline comment.
Integration test script branching tests/integration_tests/debezium_basic/run.sh	run.sh gates on SINK_TYPE (kafka or pulsar), conditionally starts Pulsar, builds the appropriate SINK_URI, creates the changefeed, and runs the matching consumer.

🎯 3 (Moderate) | ⏱️ ~20 minutes

"I’m a rabbit in the code glade, nibbling bugs away,
Canal and Debezium now hop in the same play.
Tests updated, scripts adapt, messages sing true—
A tiny hop forward, from me to you. 🐇"

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 28.57% 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
Title check	✅ Passed	The PR title 'pulsar: support debezium protocol' clearly and concisely describes the main change—extending Pulsar sink support to include Debezium protocol.
Description check	✅ Passed	The PR description is well-structured with all required sections: issue number, problem statement, detailed explanation of changes, test coverage, compatibility assessment, documentation recommendations, and comprehensive release notes.
Linked Issues check	✅ Passed	All coding requirements from issue #5056 are met: IsPulsarSupportedProtocols extended to include ProtocolDebezium, error messages updated, and unit tests added; Debezium codec reuse is confirmed.
Out of Scope Changes check	✅ Passed	All code changes directly support the stated objective of adding Debezium protocol support to the Pulsar sink; no extraneous modifications detected.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ 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 and usage tips.}

[pingcap-cla-assistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you all sign our Contributor License Agreement before we can accept your contribution.
1 out of 2 committers have signed the CLA.

✅ HGHNice
❌ heguangheng

---

heguangheng seems not to be a GitHub user. You need a GitHub account to be able to sign the CLA. If you have already a GitHub account, please add the email address used for this commit to your account.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

[gemini-code-assist]

Code Review

This pull request adds support for the Debezium protocol to the Pulsar sink by updating the protocol validation logic and error messages. Feedback includes a recommendation to replace a hardcoded boolean with a dynamic protocol check in the event router, a request for additional test coverage in existing protocol parsing and string conversion tests, and a minor grammatical correction in a user-facing error message.

[wk989898]

/ok-to-test

[wk989898]

Please add the Debezium integration tests for Pulsar.

[HGHNice]

Please add the Debezium integration tests for Pulsar.

Pulsar Debezium integration tests have been added. Thank！