如何获取Kafka Broker中记录详情？排查消费异常问题

Got it, let's tackle this Kafka message tracking problem you're facing. I've dealt with similar scenarios before, so here are some practical methods to get detailed status of individual or batch records (like consumption status, timestamps) and troubleshoot those "sent but not consumed" cases:

Kafka stores all messages on disk in segment files, and you can directly inspect these files to verify if messages were actually written to the broker, along with their metadata. Use the kafka-run-class.sh tool with the DumpLogSegments utility:

# Replace the file path with your actual topic segment file
kafka-run-class.sh kafka.tools.DumpLogSegments --files /var/lib/kafka/data/your-topic-0/00000000000000000000.log --print-data-log

This command will output detailed info for each message, including:

• offset: The unique position of the message in the partition
• timestamp: When the message was written to the broker
• key and value: The actual content of the message (if not encrypted)

To check if a specific message was consumed, compare its offset with the consumer group's committed offset for that partition (we'll cover how to get that next).

While kafka-consumer-groups.sh doesn't show individual message status, you can use it to get the committed offsets for your consumer group, then cross-reference with message offsets from the log inspection:

kafka-consumer-groups.sh --bootstrap-server your-kafka-broker:9092 --describe --group your-consumer-group --topic your-topic

Look for these fields in the output:

• CURRENT-OFFSET: The latest offset the consumer group has committed
• LOG-END-OFFSET: The latest offset available in the partition
• LAG: The number of unconsumed messages (LOG-END-OFFSET - CURRENT-OFFSET)

If a message's offset is less than CURRENT-OFFSET, it should have been consumed. If it's between CURRENT-OFFSET and LOG-END-OFFSET, it's waiting to be consumed. If your producer confirmed the message was sent but the offset isn't in the log, that means the broker never persisted it (check producer logs or broker errors).

For more granular tracking (including when messages are consumed), enable Kafka's audit logging. Edit your server.properties file to add these configurations:

# Enable audit logger
kafka.audit.logger.name=kafka.audit.logger
kafka.audit.logger.level=INFO
kafka.audit.logger.handlers=kafka.audit.logger.handler

# Configure the audit log file location
kafka.audit.logger.handler.class=org.apache.kafka.common.logging.handlers.RollingFileHandler
kafka.audit.logger.handler.file=/var/log/kafka/kafka-audit.log
kafka.audit.logger.handler.max.filesize=100MB
kafka.audit.logger.handler.max.backup.index=10

Restart the broker, and the audit logs will record events like:

• Producer sending a message (with offset, timestamp, topic, partition)
• Consumer fetching/committing offsets (with consumer group, message offsets)

This is super useful for tracing exactly when a message was processed (or not processed) by consumers.

If you need ongoing monitoring, write a simple standalone consumer that reads all messages from the topic (starting from the earliest offset) and tracks which offsets have been committed by your target consumer group. You can use the Kafka AdminClient API to fetch the committed offsets, then compare them with the offsets your validation consumer reads.

For example, in Java:

// Fetch committed offsets for the target group
AdminClient adminClient = AdminClient.create(configs);
Map<TopicPartition, OffsetAndMetadata> committedOffsets = adminClient.listConsumerGroupOffsets("your-consumer-group").partitionsToOffsetAndMetadata().get();

// Then, in your validation consumer, check if each message's offset is <= the committed offset for its partition
consumer.subscribe(Collections.singletonList("your-topic"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
    for (ConsumerRecord<String, String> record : records) {
        OffsetAndMetadata committed = committedOffsets.get(new TopicPartition(record.topic(), record.partition()));
        boolean isConsumed = committed != null && record.offset() <= committed.offset();
        System.out.printf("Message offset %d: consumed=%b%n", record.offset(), isConsumed);
    }
}

This gives you real-time visibility into which messages have been consumed by your group.

Sometimes the issue isn't about tracking messages, but about why they're not being consumed. Here are quick checks:

• Consumer auto.offset.reset: If it's set to latest, a new or restarted consumer will skip existing messages. Check if it should be earliest instead.
• Consumer errors: Look for exceptions in consumer logs (like deserialization errors, which cause the consumer to skip messages or stop processing).
• Partition assignment issues: Ensure the consumer group has enough members to cover all partitions, and that there are no rebalance loops preventing message processing.
• Broker-side issues: Check broker logs for underreplicated partitions or disk errors that might be preventing messages from being served.

内容的提问来源于stack exchange，提问作者spike