Add topicPattern property to KafkaIO.Read to match topics using a regex #26948

sjvanrossum · 2023-05-31T13:36:18Z

This addresses #19217 and #21338, matching Apache Flink's KafkaSource property topicPattern.

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Mention the appropriate issue in your description (for example: 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, comment fixes #<ISSUE NUMBER> instead.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

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.

sjvanrossum · 2023-05-31T13:43:33Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java

@@ -1578,15 +1604,30 @@ static class GenerateKafkaSourceDescriptor extends DoFn<byte[], KafkaSourceDescr

      @VisibleForTesting final @Nullable List<String> topics;

+      private final @Nullable Pattern topicPattern;


Not annotated with @VisibleForTesting since the property is not accessed directly in tests, but neither are the properties which do specify @VisibleForTesting.

github-actions · 2023-05-31T14:05:46Z

Assigning reviewers. If you would like to opt out of this review, comment assign to next reviewer:

R: @bvolpato for label java.
R: @pabloem for label io.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

bvolpato · 2023-06-01T00:12:56Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java

+              }
+            } else {
+              for (String topic : topics) {
+                for (PartitionInfo p : consumer.partitionsFor(topic)) {


Any chance this is null at this point? In the split below you have null checks, but not here.

Is this referring to topics? Both topicPartitions and topics are initialized as empty lists in the builder by default and replaced using .withTopics() and .withTopicPartitions(). The previous Preconditions.checkStateNotNull(topics) expression in the for loop should still not be null under any circumstance. Special care should be taken to carry that property forward when we add support for this property in KafkaIO's ExternalTransformRegistrar though, since it doesn't guarantee the same object state the builder guarantees.
In regards to topicPattern, if both topicPartitions and topics are empty, then topicPattern must be non-null, since the PTransform's expansion checks that at least one of those properties is set and the .withX() builder methods check that none are previously set.

As far as Kafka's topic metadata goes, .partitionsFor() will throw an exception if an unauthorized topic is requested and .listTopics() will only list all authorized topics. Both methods return initialized objects and ensure that potential null responses from the server are translated to empty collections (as far back as org.apache.kafka:kafka-clients:0.11.0.3) or throw an exception in the case of an authorization failure. I'd say that the existing check on partitionInfoList seems superfluous and could potentially be considered for deletion:

List<PartitionInfo> partitionInfoList = consumer.partitionsFor(topic); checkState( partitionInfoList != null, "Could not find any partitions info. Please check Kafka configuration and make sure " + "that provided topics exist."); for (PartitionInfo p : partitionInfoList) { partitions.add(new TopicPartition(p.topic(), p.partition())); }

github-actions · 2023-06-08T12:13:48Z

Reminder, please take a look at this pr: @bvolpato @pabloem

github-actions · 2023-06-12T12:13:57Z

Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment assign to next reviewer:

R: @robertwb for label java.
R: @chamikaramj for label io.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

bvolpato · 2023-06-13T03:08:29Z

@pabloem @johnjcasey can you please TAL / merge? Thanks!

github-actions · 2023-06-20T12:13:56Z

Reminder, please take a look at this pr: @robertwb @chamikaramj

johnjcasey · 2023-06-20T14:08:05Z

sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java

@@ -723,6 +724,31 @@ public void testUnboundedSourceWithExplicitPartitions() {
    p.run();
  }

+  @Test


can you add a test, or update this test, such that you don't match all the topics?

Added one for partial matches and one for no matches.

sjvanrossum · 2023-06-23T11:38:48Z

@johnjcasey Unit tests are failing early on a missing dependency.
The Avro plugin version in use for Beam seems to have been yanked and all versions of the published artifact com.commercehub.gradle.plugin:gradle-avro-plugin are obsolete.

It has been superseded by com.github.davidmc24.gradle.plugin.avro it seems, see changelog.
I'm not sure what the compatibility story is between <1.0.0 and >=1.0.0, but upgrading seems prudent.

github-actions · 2023-06-27T12:15:09Z

Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment assign to next reviewer:

R: @Abacn for label java.
R: @Abacn for label io.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

johnjcasey · 2023-06-27T13:58:42Z

run RAT PreCommit

johnjcasey · 2023-06-28T11:41:37Z

Tests appear to be building fine to me

chamikaramj · 2023-06-28T13:47:45Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java

+     * <p>See {@link KafkaUnboundedSource#split(int, PipelineOptions)} for description of how the
+     * partitions are distributed among the splits.
+     */
+    public Read<K, V> withTopicPattern(Pattern topicPattern) {


Can change this API to use a portable type (for example, a string regex) so that this can be supported via cross-language wrappers ?

Read.External.Configuration has fields for keyDeserializer and valueDeserializer which are resolved from String to Class during external transform construction, does it make sense to provide a mapping there instead?

Ideally we don't want to be using external read configuration as a pattern. It makes it so the configurations for an IO diverge for python v. java users

Got it. Would an additional method overload with String topicPattern help at all?
Otherwise I'll change the method signature.

I think we can have both

Is there a significant advantage to using "Pattern" over a string regex ?

If it's a perf issue, we could just build the Patter object once within "withTopicPattern" and use that ?

Mostly to make sure that pattern flags can be specified by users, but nearly all (except LITERAL and CANON_EQ) can be specified in the expression as well.

Thanks. If there is no significant advantage for specifying a Pattern object I would just support specifying a String regex to keep the API simple.

chamikaramj

Thanks. LGTM.

…ex (apache#26948) * Add topicPattern property to KafkaIO.Read to match topics using a regex * Add partially matched and unmatched topic pattern test * Change method signature of withTopicPattern to use String

github-actions bot added io java kafka labels May 31, 2023

sjvanrossum commented May 31, 2023

View reviewed changes

github-actions bot added the Next Action: Reviewers label May 31, 2023

bvolpato reviewed Jun 1, 2023

View reviewed changes

bvolpato approved these changes Jun 1, 2023

View reviewed changes

github-actions bot added the slow-review label Jun 8, 2023

github-actions bot removed the slow-review label Jun 12, 2023

github-actions bot added the slow-review label Jun 20, 2023

johnjcasey requested changes Jun 20, 2023

View reviewed changes

sjvanrossum added 2 commits June 22, 2023 13:52

Add topicPattern property to KafkaIO.Read to match topics using a regex

b689988

Add partially matched and unmatched topic pattern test

72d6b8e

sjvanrossum force-pushed the master branch from 0e11d82 to 72d6b8e Compare June 22, 2023 13:53

github-actions bot removed the slow-review label Jun 27, 2023

johnjcasey approved these changes Jun 28, 2023

View reviewed changes

chamikaramj reviewed Jun 28, 2023

View reviewed changes

Change method signature of withTopicPattern to use String

e89c4f8

chamikaramj approved these changes Jun 30, 2023

View reviewed changes

chamikaramj merged commit c36f0f1 into apache:master Jun 30, 2023
9 of 10 checks passed

This was referenced Jul 27, 2023

KafkaIO should support topic patterns #19217

Closed

Adding Regex based subscriptions to KafkaIo Dynamic Read. #21338

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add topicPattern property to KafkaIO.Read to match topics using a regex #26948

Add topicPattern property to KafkaIO.Read to match topics using a regex #26948

sjvanrossum commented May 31, 2023 •

edited

Loading

sjvanrossum May 31, 2023

github-actions bot commented May 31, 2023

bvolpato Jun 1, 2023

sjvanrossum Jun 1, 2023 •

edited

Loading

github-actions bot commented Jun 8, 2023

github-actions bot commented Jun 12, 2023

bvolpato commented Jun 13, 2023

github-actions bot commented Jun 20, 2023

johnjcasey Jun 20, 2023

sjvanrossum Jun 22, 2023

sjvanrossum commented Jun 23, 2023

github-actions bot commented Jun 27, 2023

johnjcasey commented Jun 27, 2023

johnjcasey commented Jun 28, 2023

chamikaramj Jun 28, 2023

sjvanrossum Jun 29, 2023

johnjcasey Jun 29, 2023

sjvanrossum Jun 29, 2023

johnjcasey Jun 29, 2023

chamikaramj Jun 29, 2023

sjvanrossum Jun 29, 2023

chamikaramj Jun 29, 2023

sjvanrossum Jun 30, 2023

chamikaramj left a comment

		@@ -1578,15 +1604,30 @@ static class GenerateKafkaSourceDescriptor extends DoFn<byte[], KafkaSourceDescr

		@VisibleForTesting final @Nullable List<String> topics;

		private final @Nullable Pattern topicPattern;

Add topicPattern property to KafkaIO.Read to match topics using a regex #26948

Add topicPattern property to KafkaIO.Read to match topics using a regex #26948

Conversation

sjvanrossum commented May 31, 2023 • edited Loading

GitHub Actions Tests Status (on master branch)

Choose a reason for hiding this comment

github-actions bot commented May 31, 2023

Choose a reason for hiding this comment

sjvanrossum Jun 1, 2023 • edited Loading

Choose a reason for hiding this comment

github-actions bot commented Jun 8, 2023

github-actions bot commented Jun 12, 2023

bvolpato commented Jun 13, 2023

github-actions bot commented Jun 20, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sjvanrossum commented Jun 23, 2023

github-actions bot commented Jun 27, 2023

johnjcasey commented Jun 27, 2023

johnjcasey commented Jun 28, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chamikaramj left a comment

Choose a reason for hiding this comment

sjvanrossum commented May 31, 2023 •

edited

Loading

sjvanrossum Jun 1, 2023 •

edited

Loading