Consuming Kafka fails due to "Timeout expired while fetching topic metadata"

Question

I'm trying to consume events using Apache Flink. The code is very basic, trying to connect the topic split words by space and print it to the console. Kafka version is 0.9.

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09;
import org.apache.flink.util.Collector;
import java.util.Properties;

public class KafkaStreaming {

public static void main(String[] args) throws Exception {
    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    Properties props = new Properties();
    props.setProperty("bootstrap.servers", "kafka servers:9092...");
    props.setProperty("zookeeper.connect", "kafka servers:2181...");
    props.setProperty("group.id", "flinkPOC");
    FlinkKafkaConsumer09<String> consumer = new FlinkKafkaConsumer09<>("topic", new SimpleStringSchema(), props);

    DataStream<String> dataStream = env.addSource(consumer);

    DataStream<String> wordDataStream = dataStream.flatMap(new Splitter());
    wordDataStream.print();
    env.execute("Word Split");

}

public static class Splitter implements FlatMapFunction<String, String> {

    public void flatMap(String sentence, Collector<String> out) throws Exception {

        for (String word : sentence.split(" ")) {
            out.collect(word);
        }
    }

}
}

The app does not print anything to the screen (although I produced events to Kafka). I tried to skip the Splitter FlatMap function, but still nothing happens. SSL is not required from Kafka. When I submitted the job to the cluster I found in the logs timeout exception:

2019-08-20 14:36:17,654 INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph        - Source: Custom Source -> Flat Map -> Sink: Print to Std. Out (1/1) (02258a2cafab83afbc0f5650c088da2b) switched from RUNNING to FAILED.
org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata

I am really not sure what am I doing wrong:(

Chris Matta Chris Matta · Accepted Answer · 2019-08-20T12:44:45

This is a common error that simply tells you that your client has failed to properly connect to the Kafka cluster. Do you know if the cluster has authentication enabled? Can you connect to the cluster with the kafka-topics script using the same zookeeper property?

I would also use the kafka-console-producer and kafka-console-consumer scripts that come with Kafka from the Flink host to ensure basic functionality.

Consuming Kafka fails due to "Timeout expired while fetching topic metadata"

2 Answers