fozzie15 commented on code in PR #34258:
URL: https://github.com/apache/beam/pull/34258#discussion_r2004176033
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java:
##########
@@ -1901,13 +1902,32 @@ public void
processElement(OutputReceiver<KafkaSourceDescriptor> receiver) {
for (Map.Entry<String, List<PartitionInfo>> entry :
consumer.listTopics().entrySet()) {
if (pattern.matcher(entry.getKey()).matches()) {
+ List<PartitionInfo> partitionInfoList = entry.getValue();
+ if (partitionInfoList == null) {
+ LOG.warn(
+ "Could not find any partitions info for topic {}.
Please check Kafka configuration and make sure "
+ + "that provided topics exist.",
+ entry.getKey());
+ }
for (PartitionInfo p : entry.getValue()) {
partitions.add(new TopicPartition(p.topic(),
p.partition()));
}
}
}
} else {
for (String topic : topics) {
+ try {
+ Map<String, List<PartitionInfo>> topicsInConsumer =
consumer.listTopics();
+ checkState(
+ topicsInConsumer.containsKey(topic),
+ "Could not find any partitions info for topic "
+ + topic
+ + ". Please check Kafka configuration and make sure "
+ + "that provided topics exist.");
+ } catch (KafkaException e) {
+ LOG.warn(
+ "Unable to access consumer. Skipping fail fast checks
for topic {}.", topic);
+ }
Review Comment:
I modified this such that we use partitionsFor instead of listTopics. There
was an unexpected feature where partitionsFor would create a topic when looking
for partitions in an invalid topic, but I realized that is a feature, and can
be disabled with a consumer config. I updated the test to reflect this as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscr...@beam.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
