Zhuoxi2000 opened a new pull request, #290: URL: https://github.com/apache/flink-connector-kafka/pull/290
## What is the purpose of the change Fixes [FLINK-36434](https://issues.apache.org/jira/browse/FLINK-36434): the `KafkaPartitionSplitReader` is constructed on the source-reader thread — or, when `KafkaSourceFetcherManager#commitOffsets` re-creates a fetcher after an idle shutdown, on the checkpoint thread — and its constructor eagerly created the `KafkaConsumer` (and registered consumer metrics) there. The consumer is then used almost exclusively on the split fetcher thread, but `KafkaConsumer` is not thread-safe; as described in the ticket, this broken threading model blocks Kafka client upgrades (sporadic `CorrelationIdMismatchException`). This PR makes the consumer lazy inside the reader, so it is created on the first consumer-touching call — which by construction happens on the split fetcher thread (`fetch`, `handleSplitsChanges`, `pauseOrResumeSplits`, the offset-commit task, and `close()` in the fetcher's shutdown path all run there). Design points, as discussed on the ticket: - The constructor only computes `consumerProps` (client-id derivation unchanged) and stores them; no consumer, no metric registration. - A private `ensureConsumer()` runs at the entry of every consumer-touching method; consumer metric registration moves there as well. - `wakeUp()` remains the only cross-thread entry point (the one call `KafkaConsumer` documents as thread-safe). `wakeUp()` and creation synchronize on a small lock: if the consumer does not exist yet, the wakeup is recorded and applied right after creation, so the first blocking call still observes it — matching the semantics of a wakeup against an eagerly-created consumer. The worst case is one spurious `WakeupException` on the first consumer call, which the existing `retryOnWakeup` handling already tolerates. - `close()` before first use is a no-op. `KafkaPartitionSplitReaderWrapper` (DynamicKafkaSource) subclasses this reader and only wraps `fetch()` output, so the fix covers the `DynamicKafkaSource` path as well. Deferring `splitReaderFactory.get()` itself onto the fetcher thread in flink-connector-base would fix this class of issue for all connectors; with the lazy consumer this connector no longer depends on that, so I will file it as a separate ticket (as discussed on FLINK-36434, it is orthogonal). ## Brief change log - `KafkaPartitionSplitReader` no longer creates the `KafkaConsumer` (nor registers consumer metrics) in its constructor; both happen in a new `ensureConsumer()` on the first consumer-touching call, i.e. on the split fetcher thread - `wakeUp()` before the consumer exists records a pending wakeup that is applied immediately after creation; `close()` before first use is a no-op - A protected `@VisibleForTesting createConsumer(Properties)` seam allows tests to observe consumer creation ## Verifying this change Please make sure both new and modified tests in this PR follow [the conventions for tests defined in our code quality guide](https://flink.apache.org/how-to-contribute/code-style-and-quality-common/#7-testing). This change added tests and can be verified as follows: - Added `KafkaPartitionSplitReaderConsumerThreadTest` (pure unit tests, no Kafka cluster needed): the consumer is not created at construction time and is created on the thread that first uses the reader; a `wakeUp()` arriving before creation is applied on creation exactly once; a `wakeUp()` after creation delegates directly; `close()` without prior use never creates a consumer - Existing `KafkaPartitionSplitReaderTest` covers the eager-path behaviors against a real cluster and passes unchanged in CI (not run locally: no Docker environment on this machine); `mvn test` for the module (compilation, checkstyle, spotless, and the new tests) passes locally ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): (no) - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no — `KafkaPartitionSplitReader` is `@Internal`) - The serializers: (no) - The runtime per-record code paths (performance sensitive): (no — `fetch()` gains a single volatile read on its fast path; consumer creation is one-time) - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no) - The S3 file system connector: (no) ## Documentation - Does this pull request introduce a new feature? (no) - If yes, how is the feature documented? (not applicable) --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (please specify the tool below) Claude Code -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
