[Bug]: Unhandled exception in KafkaIO SDF

[sjvanrossum]

What happened?

#34659 introduced a "fail fast" feature that appears to have broken KafkaIO's SDF.

KafkaIO.java L1924-1939 (GenerateKafkaSourceDescriptor):

                List<PartitionInfo> partitionInfoList = consumer.partitionsFor(topic);
                if (logTopicVerification == null || !logTopicVerification) {
                  checkState(
                      partitionInfoList != null && !partitionInfoList.isEmpty(),
                      "Could not find any partitions info for topic "
                          + topic
                          + ". Please check Kafka configuration and make sure "
                          + "that provided topics exist.");
                } else {
                  LOG.warn(
                      "Could not find any partitions info for topic {}. Please check Kafka configuration "
                          + "and make sure that the provided topics exist.",
                      topic);
                }

                for (PartitionInfo p : partitionInfoList) {

WatchForKafkaTopicPartitions.java L193-200:

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

This will throw IllegalStateException from those DoFns and result in retrying instead of stopping. I think logging the condition should suffice for GenerateKafkaSourceDescriptor and WatchForKafkaTopicPartitions.

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[pratx08]

Hi @sjvanrossum, could you please assign this issue to me.
I'll plan to fix this by replacing the checkState assertions and will log a warning in GenerateKafkaSourceDescriptor and WatchForKafkaTopicPartitions.

[alisha-1000]

+1, @sjvanrossum this makes sense.

Throwing IllegalStateException from these DoFns results in retries rather than a clean stop, which seems unintended for missing topic metadata. Replacing checkState with logging / graceful handling in GenerateKafkaSourceDescriptor and WatchForKafkaTopicPartitions looks like the right fix.

Happy to help review or test the PR.

[junaiddshaukat]

.take-issue

[MansiSingh17]

Hi, I’m Mansi Singh, prospective GSoC 2026 contributor interested in Kafka Streams Runner.
I’d like to work on this issue if no one else is assigned. Thanks!

[yashpawarrr]

Heyy !! Can we replace the "checkState" that throws IllegalStateException when no partitions are found with a warning log and an early return. This allows KafkaIO SDF to continue polling for partitions instead of failing and retrying the DoFn.

Extremely sorry if the issue was already fixed ...

[MansiSingh17]

@sjvanrossum, Hi, I’m Mansi Singh, prospective GSoC 2026 contributor interested in Kafka Streams Runner.
I’d like to work on this issue if no one else is assigned. Thanks!