kafka: avoid duplicate topic create-and-wait operations for concurrent callers

[3AceShowHand]

Is your feature request related to a problem?

kafkaTopicManager.GetPartitionNum checks the local topic cache and calls CreateTopicAndWaitUntilVisible on a cache miss. The cache is concurrency-safe, while the cache-miss workflow has no per-topic in-flight coordination.

A Kafka sink has independent execution paths for DML, checkpoint messages, and block events. The following pairs can occasionally access the same uncached dynamic topic concurrently:

• DML and checkpoint
• DDL and an earlier checkpoint operation that is still in progress
• DML from an unrelated dispatcher and DDL when multiple tables share a dynamic topic

Each caller can independently run metadata queries, CreateTopic, and topic-visibility retries. Kafka handles concurrent creation through TopicAlreadyExists, and the Topic Ready barrier preserves correctness. The duplicate workflows still add avoidable Admin requests and metadata retries when topic visibility is slow.

Describe the feature you'd like

Ensure concurrent callers for the same topic share one in-flight CreateTopicAndWaitUntilVisible operation within a kafkaTopicManager instance.

Expected behavior:

• The first caller starts the create-and-wait workflow.
• Concurrent callers for the same topic wait for and reuse the same result.
• Calls for different topics continue independently.
• A waiting caller can return when its own context is canceled.
• Cancellation by one waiting caller does not terminate the shared workflow for other callers.
• A failed workflow is removed from the in-flight registry so a later call can retry.
• A successful workflow publishes the partition count to the topic cache before releasing waiters.

Add concurrency tests covering shared success, shared failure, caller cancellation, retry after failure, and parallel operations for different topics.

Describe alternatives you've considered

• Continue relying on Kafka TopicAlreadyExists handling and independent Topic Ready checks. This preserves correctness and retains duplicate Admin work.
• Protect the complete cache-miss workflow with one mutex. This also serializes initialization of unrelated topics and reduces parallelism in high-topic-count changefeeds.

Teachability, Documentation, Adoption, Migration Strategy

This is an internal coordination change with no user-facing configuration or migration. Existing topic creation, partition validation, and Topic Ready semantics remain unchanged.