spuru9 opened a new pull request, #292: URL: https://github.com/apache/flink-connector-kafka/pull/292
## What is the purpose of the change `KafkaPartitionSplitReader` polls the consumer with a hardcoded 10 second timeout. On idle partitions the split reader only reacts to split changes after the ongoing poll returns, so this can add up to 10 seconds of delay to checkpoints, which is especially painful for jobs using transactional Kafka producers. This makes the timeout configurable via a new `poll.timeout.ms` source property, defaulting to the previous 10 seconds so existing jobs are unaffected. ## Brief change log - Add `KafkaSourceOptions.POLL_TIMEOUT_MS` (`poll.timeout.ms`), default 10000 ms. - `KafkaPartitionSplitReader` parses the property once in the constructor and uses it for `consumer.poll(...)` instead of the `POLL_TIMEOUT` constant. Parsing happens before the `KafkaConsumer` is created so an invalid value cannot leak a consumer, and non-positive values are rejected because they would turn the fetch loop into a busy loop. - Document the option in the DataStream Kafka docs (English and Chinese). The dynamic Kafka docs already link to that list. The property flows through the existing `Properties` bag, so it works for `KafkaSource.builder().setProperty(...)`, `DynamicKafkaSource` per-cluster properties, and SQL via `properties.poll.timeout.ms` without additional plumbing. ## Verifying this change Added to `KafkaPartitionSplitReaderTest`: - `testConfiguredPollTimeoutIsUsedForPolling`: assigns a split on an empty topic with a 500 ms poll timeout and asserts the fetch returns well below the 10 second default. Verified the assertion is meaningful by temporarily restoring the hardcoded timeout, which makes the test fail. - `testNonPositivePollTimeoutIsRejected`: parameterized over `0` and `-1`. - `testDefaultPollTimeoutIsTenSeconds`: guards the unchanged default. ## 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 - The serializers: no - The runtime per-record code paths (performance sensitive): yes, the poll timeout in `fetch()` is now read from a field; behavior is unchanged at the default value - Anything that affects deployment or recovery: no - The S3 file system connector: no ## Documentation - Does this pull request introduce a new feature? yes, a new source configuration option - If yes, how is the feature documented? docs -- 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]
