I am trying to write a quick sample of streaming word count using Beam APIs and FlinkBeamRunner. The problem that I am getting into is that apply("Write to Kafka", Write.to(UnboundedFlinkSink.of(kafkaSink))) Does not work in this way - it assumes bounded stream and mine is unbounded.
I have not found any unbounded equivalent for Write, So I tried to implement a custom ParDo function: /** * Write content to Kafka. * */ static class WriteToKafkaFn extends DoFn<Tuple2<String, Integer>, Tuple2<String, Integer>> { private FlinkKafkaProducer09<Tuple2<String, Integer>> kafkaSink; private boolean opened = false; public WriteToKafkaFn(FlinkKafkaProducer09<Tuple2<String, Integer>> kafkaSink){ this.kafkaSink = kafkaSink; } @ProcessElement public void processElement(ProcessContext c) { if(!opened){ kafkaSink.open(new Configuration()); opened = true; } Tuple2<String, Integer> record = c.element(); try { kafkaSink.invoke(record); }catch(Throwable t){ System.out.println("Error writing record " + record + " to Kafka"); t.printStackTrace(); } } } The problem with this approach is that ParDo is not initialized with Streaming context, that FlinkKafkaConsumer relies upon, so open fails. Any suggestions?