[FLINK-8497] KafkaConsumer throws NPE if topic doesn't exist

• Type: Bug
• Status: Closed
• Priority: Minor
• Resolution: Duplicate
• Affects Version/s: 1.4.0
• Component/s: Connectors / Kafka
• Labels: pull-request-available
• Assignee: Aleksei Lesnov
• Reporter: chris snow
• Created: 23/Jan/18 16:08
• Updated: 05/Feb/20 13:48
• Resolved: 05/Feb/20 13:48

Description

If I accidentally set the kafka consumer with a topic that doesn't exist:

FlinkKafkaConsumer011 kafkaConsumer = new FlinkKafkaConsumer011(
   "does_not_exist",
    new JSONKeyValueDeserializationSchema(false),
    properties
    );
DataStream<String> input = env.addSource(kafkaConsumer);

Flink throws NPE

Caused by: java.lang.NullPointerException
at org.apache.flink.streaming.connectors.kafka.internal.Kafka09PartitionDiscoverer.getAllPartitionsForTopics(Kafka09PartitionDiscoverer.java:75)
at org.apache.flink.streaming.connectors.kafka.internals.AbstractPartitionDiscoverer.discoverPartitions(AbstractPartitionDiscoverer.java:128)
at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.open(FlinkKafkaConsumerBase.java:415)
at org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:36)
at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:102)
at org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:393)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:254)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:718)
at java.lang.Thread.run(Thread.java:748)

Maybe Flink could through an IllegalStateException("Topic not found")?

Issue Links

• duplicates: FLINK-11321 Clarify the NPE on fetching an nonexistent Kafka topic
• links to: GitHub Pull Request #5929
• links to: GitHub Pull Request #5929