[Bug] Kafka sync producer does not close underlying sarama client

[tenfyzhong]

Description

TiCDC's Kafka sync producer path leaks the underlying sarama.Client when the sink is closed and recreated.

pkg/sink/kafka/sarama_factory.go creates a dedicated sarama.Client for the sync producer via sarama.NewClient(...), then passes it to sarama.NewSyncProducerFromClient(client). However, pkg/sink/kafka/sarama_sync_producer.go only calls p.producer.Close() in (*saramaSyncProducer).Close() and never closes p.client.

According to Sarama's contract, NewSyncProducerFromClient does not take ownership of the client, and the caller must still close the underlying client explicitly. Internally Sarama wraps the client with nopCloserClient, so closing the sync producer does not close the original client.

As a result, repeated sink close/recreate cycles can accumulate:

• sarama background metadata updater goroutines
• broker metadata/cache state
• broker TCP connections owned by the leaked client

This should be tracked separately from #4437, which fixes the Kafka admin client cleanup path. That PR does not address the sync producer path.

Context

Current code path:

• saramaFactory.SyncProducer() creates a new sarama.Client
• sarama.NewSyncProducerFromClient(client) creates the sync producer
• sink.Close() calls s.ddlProducer.Close()
• (*saramaSyncProducer).Close() only closes the producer, not the client

Relevant files:

• pkg/sink/kafka/sarama_factory.go
• pkg/sink/kafka/sarama_sync_producer.go
• downstreamadapter/sink/kafka/sink.go

Relevant Sarama behavior:

• NewSyncProducerFromClient requires the caller to close the underlying client
• nopCloserClient.Close() is a no-op

Steps to Reproduce

1. Create a Kafka sink so that TiCDC initializes a sync producer.
2. Trigger repeated sink close/recreate cycles, for example via repeated dispatcher/changefeed rebalance or restart.
3. Observe that (*saramaSyncProducer).Close() is called repeatedly.
4. Observe that the underlying sarama.Client from the sync producer is never closed.
5. Over time, broker connections and sarama background resources accumulate.

Expected Behavior

Closing the Kafka sync producer should also close the underlying sarama.Client created by TiCDC for that producer.

Actual Behavior

Only the sync producer is closed. The underlying sarama.Client stays alive.

Proposed Fix

• In (*saramaSyncProducer).Close(), explicitly close p.client in addition to p.producer
• Handle producer-creation failure paths carefully so a partially created client is also released
• Add unit tests that verify both the producer and the underlying client are closed

Additional Information

The current implementation even contains a misleading comment in sarama_sync_producer.go saying producer.Close() also closes the client, but that is not true for NewSyncProducerFromClient.