Kafka Streams消费者跳过偏移量问题排查咨询

问题解答 #

关于ProcessorContext.offset()是否会打印每个偏移量

在正常流程下，每条被传递到process方法的消息，ProcessorContext.offset()都会返回该消息对应的偏移量。由于你的日志是process方法的第一行，只要消息进入process方法，就会打印对应的偏移量；如果出现偏移量跳过的情况，说明这些跳过的偏移量对应的消息根本没有被传递到process方法。

1. 可查看的Kafka Streams内部日志

以下几类日志可帮助确认消息是否被消费：

• org.apache.kafka.streams.processor.internals.StreamThread：记录Stream线程的核心运行状态，包括从topic拉取的偏移量范围、处理进度、偏移量提交记录等，能直接验证是否拉取到了那些“被跳过”的偏移量。
• org.apache.kafka.clients.consumer.KafkaConsumer：消费者拉取消息的详细日志，包含每次拉取的topic分区、偏移量区间，可对比生产者日志中的偏移量，确认Broker是否将消息推送给了Streams应用。
• org.apache.kafka.streams.processor.internals.RecordDeserializer：记录消息反序列化过程中的错误，若某条消息反序列化失败，这里会有明确报错，这类消息通常不会进入process方法。
• 状态存储相关日志（如org.apache.kafka.streams.state.internals.RocksDBStore）：如果状态存储操作出现异常（如写入失败、加载错误），相关日志会记录问题，这类异常可能导致消息处理中断或跳过。

2. 消息被跳过的可能原因

结合已知信息，排除重平衡、容器重启、事务因素后，可能的原因包括：

• 消息反序列化失败：如果foo topic中的某条消息格式不符合反序列化器要求，默认情况下Streams会跳过该消息（除非配置了自定义default.deserialization.exception.handler），导致后续消息的偏移量看起来跳过了中间值。
• 状态存储异常：你的Streams应用有3个状态存储，且变更日志副本因子为1，之前的Broker长时间宕机可能导致变更日志丢失或损坏。Streams恢复后，状态存储的恢复可能与消费偏移量不一致，进而导致部分消息被跳过；或者处理消息时，状态存储写入出现不可恢复的错误，导致Streams跳过该消息及后续部分消息。
• 拓扑过滤逻辑：如果Streams拓扑包含filter/filterNot等过滤操作，被过滤掉的消息不会进入process方法，自然不会打印偏移量，看起来像是“跳过”，实际是业务逻辑过滤的结果。
• 原topic消息丢失：如果foo topic的副本因子也是1，之前的Broker宕机可能导致部分已发送的消息未被持久化（比如生产者设置acks=1时，Broker确认接收但未完成持久化就宕机），导致Streams根本无法拉取到这些消息。
• 未捕获的异常导致跳过：如果process方法或上游处理逻辑中抛出未捕获的异常，默认情况下Streams会跳过该消息，继续处理下一条，从而出现偏移量不连续的情况。

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