Hi Miki,
First, I would like to thank you for the fast response.
I recheck Kafka and it is up and running fine.
I’m still getting the same error (Timeout expired while fetching topic
metadata).
Maybe my Flink version is wrong (Kafka version is 0.9)?
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-core</artifactId>
<version>1.5.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka-0.11_2.11</artifactId>
<version>1.5.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.11</artifactId>
<version>1.5.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>1.5.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.10</artifactId>
<version>1.1.4</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_2.11</artifactId>
<version>1.7.0</version>
</dependency>
Best regards
Eyal Peer / Data Platform Developer
[cid:[email protected]]
From: miki haiat <[email protected]>
Sent: Thursday, August 22, 2019 11:03 AM
To: Eyal Pe'er <[email protected]>
Cc: [email protected]
Subject: Re: timeout error while connecting to Kafka
Can you double check that the kafka instance is up ?
The code looks fine.
Best,
Miki
On Thu, Aug 22, 2019 at 10:45 AM Eyal Pe'er
<[email protected]<mailto:[email protected]>> wrote:
Hi,
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("group.id<http://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
or any kind of authentication is not required from Kafka.
This is the error that I found in the logs:
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
The Kafka’s topic has only one partition, so the topic metadata supposed to be
very basic.
I ran Kafka and the Flink locally in order to eliminate network related issues,
but the issue persists. So my assumption is that I’m doing something wrong…
Did you encounter such issue? Does someone have different code for consuming
Kafka events ?
Best regards
Eyal Peer / Data Platform Developer
[cid:[email protected]]
