I have seen this in the past and running into it again.
I have a kafka consumer that is not getting all the records from the topic.
Kafka conforms there are 300k messages in each partition, and flink only sees a
total of 8000 records in the source.
Kafka is 2.0, flink is 1.4.2 connector is FlinkKafkaConsumer011
Properties props = new Properties();
props.setProperty("bootstrap.servers", servers);
props.setProperty("group.id", UUID.randomUUID().toString());
props.setProperty("flink.partition-discovery.interval-millis", "10000");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DataStream<String> ds = consumers.get(a.eventType);
if (ds == null) {
FlinkKafkaConsumer011<String> cons = new FlinkKafkaConsumer011<String>(
topic, new SimpleStringSchema(), props);
cons.setStartFromEarliest();
ds = env.addSource(cons).name(et.name).rebalance();
consumers.put(a.eventType, ds);
}
I am about to rip out the kafka consumer and build a source using the client
library which has been 100% reliable in working with kafka. Any pointers
welcome.
Michael
