Repository navigation
[KafkaIO] Add with_gcp_adc and num_partitions options to Kafka SchemaTransforms and YAML - #40181
Conversation
…hemaTransforms and Beam YAML - Expose `with_gcp_adc` (`withGcpAdc`) in KafkaIO read/write SchemaTransform, Python xlang IO and YAML - Add `num_partitions` in KafkaIO read SchemaTransform, Python xlang IO and YAML. This avoids the need to Kafka instance connection at pipeline submission time for Dataflow runner v1 (Streaming runner) if using SchemaTransform
|
Assigning reviewers: R: @jrmccluskey for label python. Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
| + " construction time.") | ||
| @SchemaFieldNumber("18") | ||
| @Nullable | ||
| public abstract Integer getNumPartitions(); |
There was a problem hiding this comment.
In most cases the partition set for a topic will match the range of integers from 0 to N, but users may arbitrarily retire individual partitions.
Examples:
- Create a partition per entity with a non-sequential integer id (e.g., a partition per player in a game).
- Retire a partition after the last usable offset is used to produce a record.
Replacing the number of partitions property with a topic partition list property would spare adding it later with more boilerplate to validate the different combinations of configuration properties.
There was a problem hiding this comment.
It's true there is a risk number of partitions evolving won't get honored, but in fact it's the same limitation for a typical setup
-
on Dataflow streaming runner, it uses UnboundedSource, which resolves partitions at once, at initial split that happens at pipeline expansion time
-
on Dataflow portable runner, it uses SDF implementation. Initial split happens at pipeline execution time, but after that the partitions to consume is still fixed
There is an .withDynamicRead(...) option that support dynamic partitions, led to SDK implementation and requires Dataflow runner v2.
One of the motivation of exposing this option is to be able to use KafkaIO via SchemaTransform to connect to a GMK on Dataflow Streaming runner, therefore .withDynamicRead(...) isn't available here
| + " authenticating with a Google Managed Kafka cluster.") | ||
| @SchemaFieldNumber("17") | ||
| @Nullable | ||
| public abstract Boolean getWithGcpAdc(); |
There was a problem hiding this comment.
Nit: This doesn't match the name used to configure the transform. Application default credentials is annoyingly long to spell so I'm not opposed to ADC, but I'd consider keeping it the same for consistency and clarity to users who might be unfamiliar with the acronym.
There was a problem hiding this comment.
Understand. Since this is mostly a ease-of-use change (users can current set GCP configs themselves by adding all needed configurations to config) a simple name is preferred
|
@sjvanrossum friendly pin PTAL thanks! |
|
Merging for now |
Expose
with_gcp_adc(withGcpAdc) in KafkaIO read/write SchemaTransform, Python xlang IO and YAML. This option is already present in Java KafkaIO but not exposed to xlang. It make Python and YAML user easier to use Google managed Kafka (GMK).Add
num_partitionsin KafkaIO read SchemaTransform, Python xlang IO and YAML. This avoids the need to Kafka instance connection at pipeline submission time for Dataflow runner v1 (Streaming runner) if using SchemaTransformPlease add a meaningful description for your change here
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, commentfixes #<ISSUE NUMBER>instead.CHANGES.mdwith noteworthy changes.See the Contributor Guide for more tips on how to make review process smoother.
To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md
GitHub Actions Tests Status (on master branch)
See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.