KafkaConsumer API: internal cache and visibility of it

[adamkotwasinski]

Is there any good way to tell if KafkaConsumer has anything stored in internal queue "cache" for messages? e.g. by trying to consume(...) again with timeout = 0 ?

I'm trying to figure out if it's possible to invoke consumer.consume(timeout) and receive all the records (that arrived in FetchResponses) that were received in that call, but without any extra time wasted (this is somewhat similar to Java's ConsumerRecords return type, that's an iterable over multiple records).

FWIW, this appears to be working, but it would be great to get a confirmation whether it is depending on undefined behaviour:

RdKafka::Message *msg = consumer->consume(3000); // Initial seeding.
if (RdKafka::ERR_NO_ERROR == msg->err()) {
  // got the first message, do something
  while (true) {
    RdKafka::Message *extra = consumer->consume(0); // "trying to consume the buffer"
    if (RdKafka::ERR_NO_ERROR == extra->err()) {
      // do some processing
    } else if (RdKafka::ERR__TIMED_OUT == extra->err()) {
      // buffer is empty
      break;
    } else {
      // other failures
    }
    delete extra;
  }
}

[edenhill]

Messages are prefetched from the Kafka broker and put in an internal queue, all that consume() does is pop a message from that internal queue - so there is very little overhead in calling consume() and your approach seems fine to me.

https://github.com/edenhill/librdkafka/wiki/FAQ#how-are-partitions-fetched