Apache Kafka "max.poll.records" configuration not working

user_15981445 0
Viewed 257

I am working on developing a Kafka Consumer, for which I need to poll some limited records from the Kafka channel at a time. I have set the property max.poll.records = 10, but when I start my consumer, I see that it reads all the unread messages from the Kafka at once. I am not sure why the property is not working, even though I can see the property displayed in the config logs when the Java Kafka Consumer starts. Below are some of the configs I have:

allow.auto.create.topics = true
    auto.commit.interval.ms = 5000
    auto.offset.reset = earliest
    enable.auto.commit = false
    fetch.max.bytes = 52428800
    fetch.max.wait.ms = 500
    fetch.min.bytes = 1
    heartbeat.interval.ms = 3000
    max.partition.fetch.bytes = 1048576
    max.poll.interval.ms = 300000
    max.poll.records = 10

Any help is appreciated.

kafka-consumer-api apache-kafka java
2 Answers

max.poll.records:

This controls the maximum number of records that a single call to poll() will return. This is useful to help control the amount of data your application will receive in your processing loop.

Default Value The default value of max.poll.records is 500

This setting does not impact the underlying fetching behavior. The consumer will cache the records from each fetch request and returns them incrementally from each poll.

Please refer https://lightrun.com/answers/confluentinc-kafka-connect-jdbc-how-to-use-batchsize

user_195746210

I see that it reads all the unread messages from the Kafka at once.

So? Since you have enable.auto.commit = false, you're in full control over the number of records that actually get processed, then which offset gets committed.

In other words, poll as many records as you want, but only process 10.

I cannot reproduce your issue

  public static void main(String[] args) throws Exception {
    final String topic = "foobar";
    Properties props = new Properties();
    final String bootstrap = "localhost:9092";
    props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, BytesSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
    log.info("Producing data...");
    try (Producer<Void, Integer> producer = new KafkaProducer<>(props)) {
      IntStream.range(0, 1000).forEachOrdered(i -> producer.send(new ProducerRecord<>(topic, i)));
      log.info("Flushing producer buffer...");
      producer.flush();
      Thread.sleep(Duration.ofSeconds(2).toMillis());
    }

    props = new Properties();
    props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer-" + java.util.UUID.randomUUID());
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, OffsetResetStrategy.EARLIEST.name().toLowerCase());
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, BytesDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
    props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 10); // default=500
    try (Consumer<Void, Integer> consumer = new KafkaConsumer<>(props)) {
      consumer.subscribe(Collections.singletonList(topic));
      final ConsumerRecords<Void, Integer> records = consumer.poll(Duration.ofSeconds(2));
      log.info("Consumed {}", records.count()); // Prints: 'Consumed 10'
      // for (ConsumerRecord<Void, Integer> r : records) { ... }
      // consumer.commitSync();
    }

  }
user_23086830
Related