Kafka connector issue

Posted by Michael Latta on
URL: http://deprecated-apache-flink-user-mailing-list-archive.369.s1.nabble.com/Kafka-connector-issue-tp22367.html

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