kafka(ticdc): ddl sink close the underline sink if send ddl or checkpoint failed and refactor the kafka ddl sink (#12112)

[ti-chi-bot]

This is an automated cherry-pick of #12112

What problem does this PR solve?

Issue Number: close #12666, close #12096

What is changed and how it works?

• owner/ddl_sink should close if meet any error, and rebuilt by the retry.
• kafka factory implementations should wrap error immediately, and caller should just return error, no need to trace the error.
• kafka ddl sink use encoder directly, instead of encoderBuilder
• syncProducer no need to close in a goroutine, since it only deliver one message at a time, won't be blocked if the downstream kafka cluster unavailable.

Check List

Tests

• Unit test
• Integration test
• Manual test (add detailed scripts or steps below)
• No code

Questions

Will it cause performance regression or break compatibility?

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

Release note

`None`

[ti-chi-bot]

@3AceShowHand This PR has conflicts, I have hold it.
Please resolve them or ask others to resolve them, then comment /unhold to remove the hold label.

[ti-chi-bot]

@ti-chi-bot: ## If you want to know how to resolve it, please read the guide in TiDB Dev Guide.

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 ti-community-infra/tichi repository.

[gemini-code-assist]

Code Review

This pull request refactors the DDL sink to close the underlying sink upon failure and wraps Kafka send errors at the producer level. However, there are critical issues that must be addressed: unresolved git conflict markers are present in multiple files, including mq_ddl_sink.go and pulsar_ddl_sink.go. Additionally, a redundant import alias cerror was introduced in pkg/sink/kafka/v2/factory.go which should be replaced with the existing errors import.

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.

[codecov]

Codecov Report

❌ Patch coverage is 62.96296% with 10 lines in your changes missing coverage. Please review.
⚠️ Please upload report for BASE (release-8.5@a69a385). Learn more about missing BASE report.

Additional details and impacted files

Components	Coverage Δ
cdc	57.6082% <0.0000%> (?)
dm	50.6011% <0.0000%> (?)
engine	50.7275% <0.0000%> (?)

Flag	Coverage Δ
cdc	57.6082% <62.9629%> (?)
unit	54.0094% <62.9629%> (?)

Flags with carried forward coverage won't be shown. Click here to find out more.

@@               Coverage Diff                @@
##             release-8.5     #12677   +/-   ##
================================================
  Coverage               ?   54.0094%           
================================================
  Files                  ?       1027           
  Lines                  ?     139496           
  Branches               ?          0           
================================================
  Hits                   ?      75341           
  Misses                 ?      58563           
  Partials               ?       5592           

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[ti-chi-bot]

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: 3AceShowHand

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

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

[3AceShowHand]

/unhold

[3AceShowHand]

/retest

[lidezhu]

/hold

[3AceShowHand]

/hold

[3AceShowHand]

/retest

[lidezhu]

/unhold