private void pollRecords(KafkaConsumer<String, byte[]> consumer, ConsumerCallback callback) {
            while (!closing) {
                try {
                    while (paused) {
                        Thread.sleep(PAUSE_INTERVAL);
                    }
                    ConsumerRecords<String, byte[]> records = consumer.poll(POLL_TIMEOUT);
                    for (ConsumerRecord<String, byte[]> record : records) {
                        long sendTimeNanos = Longs.fromByteArray(record.headers().lastHeader(HEADER_KEY_SEND_TIME_NANOS).value());
                        TopicPartition topicPartition = new TopicPartition(record.topic(), record.partition());
                        callback.messageReceived(topicPartition, record.value(), sendTimeNanos);
                    }
                } catch (InterruptException | InterruptedException e) {
                    // ignore, as we are closing
                } catch (Exception e) {
                    LOGGER.warn("exception occur while consuming message", e);
                }
            }
        }
Called when a message is received.

@param topicPartition the topic partition of the received message
@param payload        the received message payload
@param sendTimeNanos  the time in nanoseconds when the message was sent
tools/src/main/java/org/apache/kafka/tools/automq/perf/ConsumerService.java