Async bulk / batch poll

[raphaelauv]

Would it be possible for a bulk poll ?

exemple :

parallelConsumer.polls(records -> records.forEach( record ->
                  log.info("Concurrently processing a record: {}", record)
                  )
);

thank to this interface

void polls(Consumer<ConsumerRecords<K, V>> usersVoidConsumptionFunction);

---

I currently have a consumer that for each result of poll make a single async bulk load ( pipelining operation ) to redis ( with redisson lib )

so my flow is

while(true) :

    msgs = consumer.poll()
    bulk_redis_operation = transform(msgs)
    future = bulk_redis_operation.executeAsync()
    future.await()
    consumer.commitSync()

Thank you

[astubbs]

Yes, absolutely! See issue #18 and POC #19

Please vote on the issue :)

[astubbs]

Do you have thoughts on how you might like to control the quantity of records in the batch? Would a set max quantity be enough?

[raphaelauv]

By default it could use the value of (for a set max) of the current consumer

ConsumerConfig.MAX_POLL_RECORDS_CONFIG 

?

[astubbs]

Hmm interesting, so your use case you think you'd like to bulk process 500 messages at a time?

[raphaelauv]

500 is default value , but my application custom that value

    final int maxPollRecord = parseInt(argsMap.get("maxPollRecord"));
    logger.info("Max poll record=" + maxPollRecord);
    consumerConfig.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecord);
    Consumer<String, T> consumer = new KafkaConsumer<>(consumerConfig);