PDGGK opened a new pull request, #41107: URL: https://github.com/apache/beam/pull/41107
Fixes #41103. Python's `ReadFromKafka` always sends `redistribute_num_keys`, and its default is 0. The external builder copies that value, so the default of 32768 that `Read.withRedistribute()` applies since #36124 never reaches a cross-language read. With 0 keys, the Redistribute creates one key per record. The issue describes what that does on Flink: timer state grows until the job runs out of memory. `setupExternalBuilder` now follows `withRedistribute()`: when `redistribute` is on, 0 or an unset value means the default. A positive value is kept, and `redistribute=False` still gets 0. A Python user who passes 0 explicitly also gets the default now, as a Java user does when `withRedistribute()` is called. The issue suggests changing the Python default to `None`. That does not work on its own, because the external path never calls `withRedistribute()` and `redistributeNumKeys` is required. With the old builder and an unset value, the build fails with `Missing required properties: redistributeNumKeys`. So the fix is in the Java builder, and it also covers an unset value. Tests: `testExternalRedistributeUsesDefaultNumKeysWhenNotSet` builds the read the way the external builders do, and checks 0 and an unset value against `KafkaIO.read().withRedistribute()`. Without the change it fails with `expected:<32768> but was:<0>`. `testExternalRedistributeKeepsConfiguredNumKeys` covers a positive value and `redistribute=False`. `:sdks:java:io:kafka:test`, `spotlessCheck` and `checkstyle` pass. I did not run a Flink job, so the memory growth itself is not reproduced here. The tests check the key count that the external path ends up with. -- 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]
