sink: Make ddl and dml encoders share schema manager (#6100)

[ti-chi-bot]

This is an automated cherry-pick of #6100

What problem does this PR solve?

Issue Number: close #6097

What is changed and how it works?

One shared SchemaManager for DDL and DML

The Kafka sink creates one SchemaManager and injects it into all DML encoders and the DDL/checkpoint encoder:

Kafka sink
└── shared SchemaManager
├── DML encoder 1
├── DML encoder 2
├── ...
└── DDL/checkpoint encoder

This avoids creating separate Registry connections and caches for every encoder.

Check List

Tests

• Unit test
• Integration test

Questions

Will it cause performance regression or break compatibility?

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

Release note

Share the schema manager between DDL and DML encoders to reduce the pressure on the schema manager.

Summary by CodeRabbit

New Features

• Added Avro and Debezium Avro support for Confluent and AWS Glue schema registries.
• Enabled consistent schema handling across Kafka and Pulsar integrations.
• Added reliable schema metadata and wire-format handling.

Bug Fixes

• Improved schema caching for routed tables and concurrent processing.
• Reduced duplicate schema registrations during concurrent activity.
• Schema registry initialization failures are now reported clearly.

[ti-chi-bot]

@wk989898 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]

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

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

[coderabbitai]

Warning

Review limit reached

Next included review available in 59 minutes.

Check out review usage here.

View limit details

Limit details: You’ve used the included review currently available.

You've used all free OSS reviews for now. Wait for the free limit to reset to keep reviewing this public repository.

Learn how review limits work.

Review configuration:

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Team

Run ID: 3d143f50-876a-4f23-b389-c3779ab0e03f

📥 Commits

Reviewing files that changed from the base of the PR and between 578cb65 and 483cc9a.

📒 Files selected for processing (27)

• 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/avro/avro_test.go
• pkg/sink/codec/avro/codec_cache.go
• pkg/sink/codec/avro/decoder.go
• pkg/sink/codec/avro/decoder_test.go
• pkg/sink/codec/avro/encoder.go
• pkg/sink/codec/avro/encoder_test.go
• pkg/sink/codec/avro/schema_manager.go
• pkg/sink/codec/builder.go
• pkg/sink/codec/debezium/avro.go
• pkg/sink/codec/debezium/avro_decoder.go
• pkg/sink/codec/debezium/avro_test.go
• pkg/sink/codec/debezium/encoder.go
• pkg/sink/codec/encoder_group.go
• pkg/sink/codec/encoder_group_test.go
• pkg/sink/codec/schemamanager/confluent_schema_registry.go
• pkg/sink/codec/schemamanager/confluent_schema_registry_test.go
• pkg/sink/codec/schemamanager/glue_client.go
• pkg/sink/codec/schemamanager/glue_schema_registry.go
• pkg/sink/codec/schemamanager/glue_schema_registry_test.go
• pkg/sink/codec/schemamanager/mock_schema_registry.go
• pkg/sink/codec/schemamanager/schema_cache.go
• pkg/sink/codec/schemamanager/schema_manager.go

---

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