You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by jo...@apache.org on 2025/08/04 17:35:07 UTC
(beam) branch master updated: Fix: Dedup input list of Topics/TopicPartitions in KafkaIO (#35758)
This is an automated email from the ASF dual-hosted git repository.
johncasey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 89ae2763844 Fix: Dedup input list of Topics/TopicPartitions in KafkaIO (#35758)
89ae2763844 is described below
commit 89ae2763844882d52853538138e5b08e398e6445
Author: Dip Patel <37...@users.noreply.github.com>
AuthorDate: Mon Aug 4 13:35:02 2025 -0400
Fix: Dedup input list of Topics/TopicPartitions in KafkaIO (#35758)
---
.../java/org/apache/beam/sdk/io/kafka/KafkaIO.java | 29 +++++++++-
...KafkaIOReadImplementationCompatibilityTest.java | 3 +-
.../org/apache/beam/sdk/io/kafka/KafkaIOTest.java | 63 ++++++++++++++++++----
3 files changed, 82 insertions(+), 13 deletions(-)
diff --git a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
index 338ee59c380..bbe20b1ed63 100644
--- a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
+++ b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
@@ -1583,6 +1583,8 @@ public class KafkaIO {
final KafkaIOReadImplementationCompatibilityResult compatibility =
KafkaIOReadImplementationCompatibility.getCompatibility(this);
+ Read<K, V> kafkaRead = deduplicateTopics(this);
+
// For a number of cases, we prefer using the UnboundedSource Kafka over the new SDF-based
// Kafka source, for example,
// * Experiments 'beam_fn_api_use_deprecated_read' and use_deprecated_read will result in
@@ -1599,9 +1601,9 @@ public class KafkaIO {
|| compatibility.supportsOnly(KafkaIOReadImplementation.LEGACY)
|| (compatibility.supports(KafkaIOReadImplementation.LEGACY)
&& runnerPrefersLegacyRead(input.getPipeline().getOptions()))) {
- return input.apply(new ReadFromKafkaViaUnbounded<>(this, keyCoder, valueCoder));
+ return input.apply(new ReadFromKafkaViaUnbounded<>(kafkaRead, keyCoder, valueCoder));
}
- return input.apply(new ReadFromKafkaViaSDF<>(this, keyCoder, valueCoder));
+ return input.apply(new ReadFromKafkaViaSDF<>(kafkaRead, keyCoder, valueCoder));
}
private void checkRedistributeConfiguration() {
@@ -1648,6 +1650,29 @@ public class KafkaIO {
}
}
+ private Read<K, V> deduplicateTopics(Read<K, V> kafkaRead) {
+ final List<String> topics = getTopics();
+ if (topics != null && !topics.isEmpty()) {
+ final List<String> distinctTopics = topics.stream().distinct().collect(Collectors.toList());
+ if (topics.size() == distinctTopics.size()) {
+ return kafkaRead;
+ }
+ return kafkaRead.toBuilder().setTopics(distinctTopics).build();
+ }
+
+ final List<TopicPartition> topicPartitions = getTopicPartitions();
+ if (topicPartitions != null && !topicPartitions.isEmpty()) {
+ final List<TopicPartition> distinctTopicPartitions =
+ topicPartitions.stream().distinct().collect(Collectors.toList());
+ if (topicPartitions.size() == distinctTopicPartitions.size()) {
+ return kafkaRead;
+ }
+ return kafkaRead.toBuilder().setTopicPartitions(distinctTopicPartitions).build();
+ }
+
+ return kafkaRead;
+ }
+
// This class is designed to mimic the Flink pipeline options, so we can check for the
// checkpointingInterval property, but without needing to depend on the Flink runner
// Do not use this
diff --git a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOReadImplementationCompatibilityTest.java b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOReadImplementationCompatibilityTest.java
index 8eda52bcec9..26682946afc 100644
--- a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOReadImplementationCompatibilityTest.java
+++ b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOReadImplementationCompatibilityTest.java
@@ -116,7 +116,8 @@ public class KafkaIOReadImplementationCompatibilityTest {
false, /*redistribute*/
false, /*allowDuplicates*/
0, /*numKeys*/
- null /*offsetDeduplication*/)));
+ null, /*offsetDeduplication*/
+ null /*topics*/)));
return p.run();
}
diff --git a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java
index 6caa1868c99..fcc7b16e467 100644
--- a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java
+++ b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java
@@ -392,7 +392,8 @@ public class KafkaIOTest {
false, /*redistribute*/
false, /*allowDuplicates*/
0, /*numKeys*/
- null /*offsetDeduplication*/);
+ null, /*offsetDeduplication*/
+ null /*topics*/);
}
static KafkaIO.Read<Integer, Long> mkKafkaReadTransformWithOffsetDedup(
@@ -404,7 +405,23 @@ public class KafkaIOTest {
true, /*redistribute*/
false, /*allowDuplicates*/
100, /*numKeys*/
- true /*offsetDeduplication*/);
+ true, /*offsetDeduplication*/
+ null /*topics*/);
+ }
+
+ static KafkaIO.Read<Integer, Long> mkKafkaReadTransformWithTopics(
+ int numElements,
+ @Nullable SerializableFunction<KV<Integer, Long>, Instant> timestampFn,
+ List<String> topics) {
+ return mkKafkaReadTransform(
+ numElements,
+ numElements,
+ timestampFn,
+ false, /*redistribute*/
+ false, /*allowDuplicates*/
+ 0, /*numKeys*/
+ null, /*offsetDeduplication*/
+ topics /*topics*/);
}
/**
@@ -418,15 +435,21 @@ public class KafkaIOTest {
@Nullable Boolean redistribute,
@Nullable Boolean withAllowDuplicates,
@Nullable Integer numKeys,
- @Nullable Boolean offsetDeduplication) {
+ @Nullable Boolean offsetDeduplication,
+ @Nullable List<String> topics) {
KafkaIO.Read<Integer, Long> reader =
KafkaIO.<Integer, Long>read()
.withBootstrapServers(mkKafkaServers)
- .withTopics(mkKafkaTopics)
+ .withTopics(topics != null ? topics : mkKafkaTopics)
.withConsumerFactoryFn(
new ConsumerFactoryFn(
- mkKafkaTopics, 10, numElements, OffsetResetStrategy.EARLIEST)) // 20 partitions
+ topics != null
+ ? topics.stream().distinct().collect(Collectors.toList())
+ : mkKafkaTopics,
+ 10,
+ numElements,
+ OffsetResetStrategy.EARLIEST)) // 20 partitions
.withKeyDeserializer(IntegerDeserializer.class)
.withValueDeserializer(LongDeserializer.class);
if (maxNumRecords != null) {
@@ -648,6 +671,21 @@ public class KafkaIOTest {
p.run();
}
+ @Test
+ public void testUnboundedSourceWithDuplicateTopics() {
+ int numElements = 1000;
+ List<String> topics = ImmutableList.of("topic_a", "topic_b", "topic_a");
+
+ PCollection<Long> input =
+ p.apply(
+ mkKafkaReadTransformWithTopics(numElements, new ValueAsTimestampFn(), topics)
+ .withoutMetadata())
+ .apply(Values.create());
+
+ addCountingAsserts(input, numElements);
+ p.run();
+ }
+
@Test
public void testRiskyConfigurationWarnsProperly() {
int numElements = 1000;
@@ -682,7 +720,8 @@ public class KafkaIOTest {
true, /*redistribute*/
true, /*allowDuplicates*/
0, /*numKeys*/
- null /*offsetDeduplication*/)
+ null, /*offsetDeduplication*/
+ null /*topics*/)
.commitOffsetsInFinalize()
.withConsumerConfigUpdates(
ImmutableMap.of(ConsumerConfig.GROUP_ID_CONFIG, "group_id"))
@@ -709,7 +748,8 @@ public class KafkaIOTest {
true, /*redistribute*/
false, /*allowDuplicates*/
0, /*numKeys*/
- null /*offsetDeduplication*/)
+ null, /*offsetDeduplication*/
+ null /*topics*/)
.commitOffsetsInFinalize()
.withConsumerConfigUpdates(
ImmutableMap.of(ConsumerConfig.GROUP_ID_CONFIG, "group_id"))
@@ -737,7 +777,8 @@ public class KafkaIOTest {
false, /*redistribute*/
false, /*allowDuplicates*/
0, /*numKeys*/
- null /*offsetDeduplication*/)
+ null, /*offsetDeduplication*/
+ null /*topics*/)
.withRedistributeNumKeys(100)
.commitOffsetsInFinalize()
.withConsumerConfigUpdates(
@@ -2109,7 +2150,8 @@ public class KafkaIOTest {
false, /*redistribute*/
false, /*allowDuplicates*/
0, /*numKeys*/
- null /*offsetDeduplication*/)
+ null, /*offsetDeduplication*/
+ null /*topics*/)
.withStartReadTime(new Instant(startTime))
.withoutMetadata())
.apply(Values.create());
@@ -2154,7 +2196,8 @@ public class KafkaIOTest {
false, /*redistribute*/
false, /*allowDuplicates*/
0, /*numKeys*/
- null /*offsetDeduplication*/)
+ null, /*offsetDeduplication*/
+ null /*topics*/)
.withStartReadTime(new Instant(startTime))
.withoutMetadata())
.apply(Values.create());