Tracking Kafka enhancement and migrate to franz-go

[3AceShowHand]

This issue tracks the Kafka Sink migration to franz-go. Related work should be added as sub-issues, and implementation progress should be recorded here.

Planning baseline

• Plan all work from master.
• The current franz-go POC branch is reference material and will not be merged directly.
• master currently uses Sarama for Kafka Admin, sync producer, async producer, authentication, and metrics.

Milestone 0: complete the existing Kafka Sink tests

Goal

Establish a complete Kafka Sink behavior baseline before adding franz-go. The current Sarama implementation on master must pass this baseline, but the tests and mocks must not depend on Sarama implementation details.

Out of scope

• Adding the franz-go implementation.
• Changing the default Kafka client.
• Adding the per-Changefeed franz-go/Sarama selector.

These belong to Milestone 1 after the test baseline is complete.

Required work

1. Make Kafka Sink tests client-independent

• Audit the existing unit, integration, and test-infra Kafka coverage and record the missing behavior and failure cases.
• Define common behavior tests around TiCDC-owned interfaces in pkg/sink/kafka/factory.go:23 and pkg/sink/kafka/cluster_admin_client.go:30.
• Remove Sarama types, configuration objects, errors, mocks, and call-order assumptions from Kafka Sink behavior tests.
• Generate mocks only for TiCDC-owned interfaces such as Factory, ClusterAdminClient, SyncProducer, AsyncProducer, and MetricsCollector.
• Refactor downstreamadapter/sink/kafka/sink_test.go:90 so Verify tests do not use sarama.MockBroker. Add a client-independent factory/Admin test seam if required.
• Keep client-library adapter tests separate from the common behavior suite. Adapter tests cannot define Kafka Sink behavior or replace the common acceptance tests.

2. Complete unit coverage for Kafka Sink behavior

• Admin: broker config, topic config, metadata, missing topic/response, create and validate topic, authorization, timeout, error mapping, and close behavior.
• Sync producer: single-partition and all-partition sends, partial failure, timeout, send after close, concurrent close, and error annotation.
• Async producer: success callback, asynchronous error, backpressure, context cancellation, send after close, concurrent close, and callback exactly once.
• Options: required acks, retry, compression, Kafka version, message-size settings, TLS, SASL, invalid values, defaults, and URI/config precedence without asserting a Sarama configuration object.
• Factory and lifecycle: construction failure, partial resource cleanup, Admin/producer/metrics shutdown order, context cancellation, and no goroutine or metric-series leaks.
• Verify and Sink creation: invalid configuration, missing Topic, Topic validation, encoder/Schema Registry errors, and cleanup after partial construction in downstreamadapter/sink/kafka/sink.go:74 and downstreamadapter/sink/kafka/helper.go:56.
• Record package coverage before and after the work. Critical correctness, error, and lifecycle branches must have behavior assertions.

3. Complete the existing Kafka regression baseline

• Cover initial replication, incremental replication, supported protocols, dispatchers, large messages, Topic management, Schema Registry, multiple Changefeeds, scaling, and HA.
• Ensure each regression case validates produced data through the consumer and checks end-to-end data consistency.
• Remove client-library names and assumptions from testcase names, parameters, logs, and success conditions.
• Make each testcase reusable by a later client selector without copying the test logic.

4. Complete failure-recovery coverage

• Rename cdc_sarama_no_broken_pipe to a client-independent broken-pipe/reconnect case.
• Cover broker restart, controller failover, network interruption, metadata refresh, request timeout, retry, idle connection, and broken pipe.
• Determine recovery through controller election, broker registration, Changefeed checkpoint progress, and final data consistency. Do not use fixed sleeps as success conditions.
• Verify that every failure case recovers automatically, catches up, and loses no data.