Delay between message insertion into Kafka and message consumption from KafkaConsumer client

[verrocchimarco]

Description

We're using a KafkaConsumer object (not Legacy consumer) which is consuming from a single topic. The behaviour I'm about to describe happens regardless of whether the KafkaConsumer is consuming all partitions, a subset of them or a single partition.

We're registering an average delay of 950ms between the moment a new record is uploaded into the Kafka topic and the moment our consumer actually manages to consume it.

This is an example of the events we are registering

Here we have the first highlighted trace indicating that we have received the END_OF_PARTITION errorcode from Kafka for partition 5. The HighOffset is 3193270. Later on, at the second highlighted trace, we receive the first record added to partition 5. We compute the epoch difference between the RdkafkaMessage timestamp and "now". Considering our KafkaConsumer was ready to consume any other message (even the other partitions had reached the END_OF_PARTITION), one would expect the difference to be very very small (maybe 200 ms), instead we can see here it gets to 859 ms. Sometimes it even gets as big as 1500ms.

My team and I have been trying to understand why we're getting such a considerable delay for over 2 weeks and we don't understand what could be the cause.

Checklist

• librdkafka version: 1.8.2
• Apache Kafka version: 6.2.1
• librdkafka client configuration:

  
  KAFKA conf cfg [gzip,snappy,sasl,regex,lz4,sasl_plain,plugins]=[builtin.features]
  KAFKA conf cfg [rdkafka]=[client.id]
  KAFKA conf cfg [librdkafka]=[client.software.name]
  KAFKA conf cfg [10.232.64.2:9092,10.232.64.50:9092,10.232.64.51:9092,10.232.64.52:9092,10.232.64.61:9092]=[metadata.broker.list]
  KAFKA conf cfg [1000000]=[message.max.bytes]
  KAFKA conf cfg [65535]=[message.copy.max.bytes]
  KAFKA conf cfg [100000000]=[receive.message.max.bytes]
  KAFKA conf cfg [1000000]=[max.in.flight.requests.per.connection]
  KAFKA conf cfg [10]=[metadata.request.timeout.ms]
  KAFKA conf cfg [300000]=[topic.metadata.refresh.interval.ms]
  KAFKA conf cfg [900000]=[metadata.max.age.ms]
  KAFKA conf cfg [250]=[topic.metadata.refresh.fast.interval.ms]
  KAFKA conf cfg [10]=[topic.metadata.refresh.fast.cnt]
  KAFKA conf cfg [true]=[topic.metadata.refresh.sparse]
  KAFKA conf cfg [30000]=[topic.metadata.propagation.max.ms]
  KAFKA conf cfg []=[debug]
  KAFKA conf cfg [60000]=[socket.timeout.ms]
  KAFKA conf cfg [1000]=[socket.blocking.max.ms]
  KAFKA conf cfg [0]=[socket.send.buffer.bytes]
  KAFKA conf cfg [0]=[socket.receive.buffer.bytes]
  KAFKA conf cfg [false]=[socket.keepalive.enable]
  KAFKA conf cfg [false]=[socket.nagle.disable]
  KAFKA conf cfg [1]=[socket.max.fails]
  KAFKA conf cfg [1000]=[broker.address.ttl]
  KAFKA conf cfg [any]=[broker.address.family]
  KAFKA conf cfg [0]=[connections.max.idle.ms]
  KAFKA conf cfg [true]=[enable.sparse.connections]
  KAFKA conf cfg [0]=[reconnect.backoff.jitter.ms]
  KAFKA conf cfg [100]=[reconnect.backoff.ms]
  KAFKA conf cfg [10000]=[reconnect.backoff.max.ms]
  KAFKA conf cfg [0]=[statistics.interval.ms]
  KAFKA conf cfg [0]=[enabled_events]
  KAFKA conf cfg [0x413260]=[log_cb]
  KAFKA conf cfg [6]=[log_level]
  KAFKA conf cfg [false]=[log.queue]
  KAFKA conf cfg [true]=[log.thread.name]
  KAFKA conf cfg [true]=[enable.random.seed]
  KAFKA conf cfg [true]=[log.connection.close]
  KAFKA conf cfg [0x428000]=[socket_cb]
  KAFKA conf cfg [0x44a920]=[open_cb]
  KAFKA conf cfg [0x1ca0410]=[default_topic_conf]
  KAFKA conf cfg [0]=[internal.termination.signal]
  KAFKA conf cfg [true]=[api.version.request]
  KAFKA conf cfg [10000]=[api.version.request.timeout.ms]
  KAFKA conf cfg [0]=[api.version.fallback.ms]
  KAFKA conf cfg [0.10.0]=[broker.version.fallback]
  KAFKA conf cfg [plaintext]=[security.protocol]
  KAFKA conf cfg [Root]=[ssl.ca.certificate.stores]
  KAFKA conf cfg [dynamic]=[ssl.engine.id]
  KAFKA conf cfg [true]=[enable.ssl.certificate.verification]
  KAFKA conf cfg [none]=[ssl.endpoint.identification.algorithm]
  KAFKA conf cfg [GSSAPI]=[sasl.mechanisms]
  KAFKA conf cfg [kafka]=[sasl.kerberos.service.name]
  KAFKA conf cfg [kafkaclient]=[sasl.kerberos.principal]
  KAFKA conf cfg [kinit -R -t "%{sasl.kerberos.keytab}" -k %{sasl.kerberos.principal} || kinit -t "%{sasl.kerberos.keytab}" -k %{sasl.kerberos.principal}]=[sasl.kerberos.kinit.cmd]
  KAFKA conf cfg [60000]=[sasl.kerberos.min.time.before.relogin]
  KAFKA conf cfg [false]=[enable.sasl.oauthbearer.unsecure.jwt]
  KAFKA conf cfg [0]=[test.mock.num.brokers]
  KAFKA conf cfg [drv_kafka_1804289383]=[group.id]
  KAFKA conf cfg [range,roundrobin]=[partition.assignment.strategy]
  KAFKA conf cfg [45000]=[session.timeout.ms]
  KAFKA conf cfg [3000]=[heartbeat.interval.ms]
  KAFKA conf cfg [consumer]=[group.protocol.type]
  KAFKA conf cfg [600000]=[coordinator.query.interval.ms]
  KAFKA conf cfg [300000]=[max.poll.interval.ms]
  KAFKA conf cfg [false]=[enable.auto.commit]
  KAFKA conf cfg [5000]=[auto.commit.interval.ms]
  KAFKA conf cfg [true]=[enable.auto.offset.store]
  KAFKA conf cfg [100000]=[queued.min.messages]
  KAFKA conf cfg [1048576]=[queued.max.messages.kbytes]
  KAFKA conf cfg [500]=[fetch.wait.max.ms]
  KAFKA conf cfg [1048576]=[fetch.message.max.bytes]
  KAFKA conf cfg [52428800]=[fetch.max.bytes]
  KAFKA conf cfg [1]=[fetch.min.bytes]
  KAFKA conf cfg [500]=[fetch.error.backoff.ms]
  KAFKA conf cfg [broker]=[offset.store.method]
  KAFKA conf cfg [read_committed]=[isolation.level]
  KAFKA conf cfg [true]=[enable.partition.eof]
  KAFKA conf cfg [false]=[check.crcs]
  KAFKA conf cfg [false]=[allow.auto.create.topics]
  KAFKA conf cfg []=[client.rack]
  KAFKA conf cfg [60000]=[transaction.timeout.ms]
  KAFKA conf cfg [false]=[enable.idempotence]
  KAFKA conf cfg [false]=[enable.gapless.guarantee]
  KAFKA conf cfg [100000]=[queue.buffering.max.messages]
  KAFKA conf cfg [1048576]=[queue.buffering.max.kbytes]
  KAFKA conf cfg [5]=[queue.buffering.max.ms]
  KAFKA conf cfg [2147483647]=[message.send.max.retries]
  KAFKA conf cfg [100]=[retry.backoff.ms]
  KAFKA conf cfg [1]=[queue.buffering.backpressure.threshold]
  KAFKA conf cfg [none]=[compression.codec]
  KAFKA conf cfg [10000]=[batch.num.messages]
  KAFKA conf cfg [1000000]=[batch.size]
  KAFKA conf cfg [false]=[delivery.report.only.error]
  KAFKA conf cfg [10]=[sticky.partitioning.linger.ms]
  KAFKA conf topic cfg [-1]=[request.required.acks]
  KAFKA conf topic cfg [30000]=[request.timeout.ms]
  KAFKA conf topic cfg [300000]=[message.timeout.ms]
  KAFKA conf topic cfg [fifo]=[queuing.strategy]
  KAFKA conf topic cfg [false]=[produce.offset.report]
  KAFKA conf topic cfg [consistent_random]=[partitioner]
  KAFKA conf topic cfg [inherit]=[compression.codec]
  KAFKA conf topic cfg [-1]=[compression.level]
  KAFKA conf topic cfg [true]=[auto.commit.enable]
  KAFKA conf topic cfg [60000]=[auto.commit.interval.ms]
  KAFKA conf topic cfg [largest]=[auto.offset.reset]
  KAFKA conf topic cfg [.]=[offset.store.path]
  KAFKA conf topic cfg [-1]=[offset.store.sync.interval.ms]
  KAFKA conf topic cfg [broker]=[offset.store.method]
  KAFKA conf topic cfg [0]=[consume.callback.max.messages]

 - **Operating system:** RedHat 8

[le-cardinal]

hello, i would also be very interested by some beginning of explanations or just some ideas about latency observed by @verrocchimarco