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 2026/05/01 15:38:45 UTC
(beam) branch master updated: Fix unhandled exception in KafkaIO SDF (#37449) (#37553)
This is an automated email from the ASF dual-hosted git repository.
johnjcasey 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 a4cb67621c6 Fix unhandled exception in KafkaIO SDF (#37449) (#37553)
a4cb67621c6 is described below
commit a4cb67621c6a24387c6699564638ee38effd5119
Author: M Junaid Shaukat <15...@users.noreply.github.com>
AuthorDate: Fri May 1 20:38:38 2026 +0500
Fix unhandled exception in KafkaIO SDF (#37449) (#37553)
Replace checkState() with LOG.warn() in GenerateKafkaSourceDescriptor
and WatchForKafkaTopicPartitions to prevent IllegalStateException from
causing infinite retries when Kafka returns null or empty partition info
for a topic. Topics with missing partitions are now gracefully skipped
with a warning log instead of throwing.
---
.../java/org/apache/beam/sdk/io/kafka/KafkaIO.java | 12 ++----
.../sdk/io/kafka/WatchForKafkaTopicPartitions.java | 15 +++++---
.../io/kafka/WatchForKafkaTopicPartitionsTest.java | 43 ++++++++++++++++++++++
3 files changed, 57 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 518319a38e3..f9bfdd87aaa 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
@@ -2130,16 +2130,12 @@ public class KafkaIO {
} else {
for (String topic : topics) {
List<PartitionInfo> partitionInfoList = consumer.partitionsFor(topic);
- if (logTopicVerification == null || !logTopicVerification) {
- checkState(
- partitionInfoList != null && !partitionInfoList.isEmpty(),
- "Could not find any partitions info for topic %s. Please check Kafka configuration and make sure that provided topics exist.",
- topic);
- } else {
+ if (partitionInfoList == null || partitionInfoList.isEmpty()) {
LOG.warn(
- "Could not find any partitions info for topic {}. Please check Kafka configuration "
- + "and make sure that the provided topics exist.",
+ "Could not find any partitions info for topic {}. Please check Kafka "
+ + "configuration and make sure that the provided topics exist.",
topic);
+ continue;
}
for (PartitionInfo p : partitionInfoList) {
diff --git a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/WatchForKafkaTopicPartitions.java b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/WatchForKafkaTopicPartitions.java
index 490faafb22f..3184b18267b 100644
--- a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/WatchForKafkaTopicPartitions.java
+++ b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/WatchForKafkaTopicPartitions.java
@@ -18,7 +18,6 @@
package org.apache.beam.sdk.io.kafka;
import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects.firstNonNull;
-import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
import java.util.ArrayList;
import java.util.List;
@@ -45,6 +44,8 @@ import org.apache.kafka.common.TopicPartition;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.joda.time.Duration;
import org.joda.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* A {@link PTransform} for continuously querying Kafka for new partitions, and emitting those
@@ -57,6 +58,7 @@ import org.joda.time.Instant;
*/
class WatchForKafkaTopicPartitions extends PTransform<PBegin, PCollection<KafkaSourceDescriptor>> {
+ private static final Logger LOG = LoggerFactory.getLogger(WatchForKafkaTopicPartitions.class);
private static final Duration DEFAULT_CHECK_DURATION = Duration.standardHours(1);
private static final String COUNTER_NAMESPACE = "watch_kafka_topic_partition";
@@ -191,10 +193,13 @@ class WatchForKafkaTopicPartitions extends PTransform<PBegin, PCollection<KafkaS
if (topics != null && !topics.isEmpty()) {
for (String topic : topics) {
List<PartitionInfo> partitionInfoList = kafkaConsumer.partitionsFor(topic);
- checkState(
- partitionInfoList != null && !partitionInfoList.isEmpty(),
- "Could not find any partitions info for topic %s. Please check Kafka configuration and make sure that provided topics exist.",
- topic);
+ if (partitionInfoList == null || partitionInfoList.isEmpty()) {
+ LOG.warn(
+ "Could not find any partitions info for topic {}. Please check Kafka "
+ + "configuration and make sure that the provided topics exist.",
+ topic);
+ continue;
+ }
for (PartitionInfo partition : partitionInfoList) {
current.add(new TopicPartition(topic, partition.partition()));
}
diff --git a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/WatchForKafkaTopicPartitionsTest.java b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/WatchForKafkaTopicPartitionsTest.java
index 595d040bf40..30ace6cd86d 100644
--- a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/WatchForKafkaTopicPartitionsTest.java
+++ b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/WatchForKafkaTopicPartitionsTest.java
@@ -18,10 +18,12 @@
package org.apache.beam.sdk.io.kafka;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import java.util.Collections;
import java.util.Set;
import java.util.regex.Pattern;
import org.apache.beam.sdk.io.kafka.KafkaMocks.PartitionGrowthMockConsumer;
@@ -108,6 +110,47 @@ public class WatchForKafkaTopicPartitionsTest {
(input) -> mockConsumer, null, givenTopics, null));
}
+ @Test
+ public void testGetAllTopicPartitionsWithNullPartitionInfo() throws Exception {
+ Set<String> givenTopics = ImmutableSet.of("topic1");
+
+ Consumer<byte[], byte[]> mockConsumer = Mockito.mock(Consumer.class);
+ when(mockConsumer.partitionsFor("topic1")).thenReturn(null);
+ assertTrue(
+ WatchForKafkaTopicPartitions.getAllTopicPartitions(
+ (input) -> mockConsumer, null, givenTopics, null)
+ .isEmpty());
+ }
+
+ @Test
+ public void testGetAllTopicPartitionsWithEmptyPartitionInfo() throws Exception {
+ Set<String> givenTopics = ImmutableSet.of("topic1");
+
+ Consumer<byte[], byte[]> mockConsumer = Mockito.mock(Consumer.class);
+ when(mockConsumer.partitionsFor("topic1")).thenReturn(Collections.emptyList());
+ assertTrue(
+ WatchForKafkaTopicPartitions.getAllTopicPartitions(
+ (input) -> mockConsumer, null, givenTopics, null)
+ .isEmpty());
+ }
+
+ @Test
+ public void testGetAllTopicPartitionsSkipsMissingTopics() throws Exception {
+ Set<String> givenTopics = ImmutableSet.of("topic1", "topic2");
+
+ Consumer<byte[], byte[]> mockConsumer = Mockito.mock(Consumer.class);
+ when(mockConsumer.partitionsFor("topic1")).thenReturn(null);
+ when(mockConsumer.partitionsFor("topic2"))
+ .thenReturn(
+ ImmutableList.of(
+ new PartitionInfo("topic2", 0, null, null, null),
+ new PartitionInfo("topic2", 1, null, null, null)));
+ assertEquals(
+ ImmutableList.of(new TopicPartition("topic2", 0), new TopicPartition("topic2", 1)),
+ WatchForKafkaTopicPartitions.getAllTopicPartitions(
+ (input) -> mockConsumer, null, givenTopics, null));
+ }
+
@Test
public void testGetAllTopicPartitionsWithGivenPattern() throws Exception {
Consumer<byte[], byte[]> mockConsumer = Mockito.mock(Consumer.class);