Spring Kafka消费者跳过一些抵消

Question

我们有一个非常简单的Kafka Consumer（v 2.6.2）。它是使用者组中唯一的使用者，并且该组是唯一一个阅读主题的组（有6个分区，其中有大约300万个事件）。Broker也是2.6.x版本

由于我们需要实现一个“只有一次”的场景，我们深入研究了一下，如果我们真的只使用一次写入主题的每个事件。不幸的是，我们发现：消费者有时会跳过一个偏移量，有时甚至会跳过一组分区的偏移量。

消费者除了记录之外没有做更多的事情。错误处理程序配置了日志记录。

@KafkaListener(id = "${spring.kafka.consumer.group-id}", topics = "${topicname}")
public void onMessage(ConsumerRecord<String, BlaBlaEvent> record) throws Exception {
  // mostly logging
}

并记录：

consuming Offset=30 Partition=2
consuming Offset=31 Partition=2
consuming Offset=32 Partition=2
consuming Offset=67 Partition=2

(看到Offset=67了吗？)

这是我们的应用程序属性

spring.kafka.bootstrap-servers=${KAFKA_BOOTSTRAP_SERVERS_URL}
spring.kafka.properties.security.protocol: SSL
spring.kafka.properties.schema.registry.url=${SCHEMA_REGISTRY_URL}
spring.kafka.properties.specific.avro.reader=true

spring.kafka.ssl.trust-store-type=PKCS12
spring.kafka.ssl.key-store-type=PKCS12
spring.kafka.ssl.key-store-password=${KEYSTORE_PASSWORD}
spring.kafka.ssl.trust-store-password=${TRUSTSTORE_PASSWORD}
spring.kafka.ssl.trust-store-location=${TRUSTSTORE_PATH}
spring.kafka.ssl.key-store-location=${KEYSTORE_PATH}

spring.kafka.consumer.client-id=BlablaNotificationReader
spring.kafka.consumer.group-id=blabla-group
spring.kafka.consumer.auto-offset-reset=none
spring.kafka.consumer.max-poll-records=1
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.IntegerDeserializer
spring.kafka.consumer.value-deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer

logging.level.org.apache.kafka=DEBUG

• 消费者在OpenShift v3.11（库伯内特斯1.11）中运行。当时没有pod重启或发生任何事情。
• 在代理端和消费者端的日志中没有错误。只是无声跳过。
• 大约一周发生一次

有人知道如何找到这个问题的根源吗？

Answer 1

由于您正在寻找精确一次语义学，因此您可能有启用事务的生产者-如果是这样，这可能是由于分区具有事务记录。

Kafka消费者留档声明：

具有事务性消息的分区将包括指示事务结果的提交或中止标记。这些标记不会返回给应用程序，但在日志中有一个偏移量。因此，从具有事务性消息的主题中读取的应用程序将会看到使用的偏移量中的缺口。这些丢失的消息将成为事务标记，在这两个隔离级别中，它们都被消费者过滤掉了。此外，使用read_committed使用者的应用程序也可能会看到由于中止的事务而产生的间隙，因为这些消息不会由使用者返回，但会有有效的偏移量。

我看到实际错过记录的唯一方法是，如果使用者为由于某种原因未被处理的记录/记录提交偏移量或之后提交偏移量。

如果需要有关从代理轮询的记录和正在提交的偏移量的更多信息，可以将 org.springframework.kafka.listener.KafkaMessageListenerContainer 类的日志记录级别设置为 TRACE。这应该可以让您很好地了解收到的记录和提交的偏移量。如果在处理正在提交的偏移量之前，所有偏移量都已编辑：则您不太可能缺少任何偏移量。

为了检查＜code＞一次＜/code＞语义是否正确，也许可以尝试通过某些id将生产者的日志与消费者的日志关联起来，特别是在发生此事件时。

您可能还想检查发生这种情况时的生产者日志，因为这种差距可能是由于中止的事务造成的。