On Aug 22, 2019, at 15:45, Eyal Pe'er <eyal.peer@startapp.com> 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.f= unctions.FlatMapFunction;
import org.apache.flink.api.common.s= erialization.SimpleStringSchema;

import org.apache.flink.streaming.ap= i.datastream.DataStream;
import org.apache.flink.streaming.ap= i.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.co= nnectors.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 =3D StreamExecutionEnvironment.getExecutionEnvironment();

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

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

    DataStream<String> wordDataStream =3D = 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(" = ")) {
          &nb= sp; 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 -> FlatMap -> 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=E2=80=99s 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=E2=80=99m doing something wrong=E2=80=A6
Did you = encounter such issue? Does someone have different code for consuming = Kafka events ?

Best regards
Eyal Peer / Data = Platform Developer
<image001.png>