springboot消费者不会消费任何消息

Question

我有一个springboot消费者应用程序。当我第一次运行它时，它消耗了来自Kafka主题的信息。但当我再次运行它时，它停止了消耗。在日志中，我看到以下消息。

2022-08-02 00:14:08.130  INFO 96561 --- [ntainer#0-0-C-1] o.a.k.c.consumer.internals.Fetcher       : [Consumer clientId=consumer-my-transformation-local-1, groupId=my-transformation-local] Fetch position FetchPosition{offset=17400, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[localhost:9092 (id: 1 rack: null)], epoch=0}} is out of range for partition CraveVideoTestLog-0, resetting offset
2022-08-02 00:14:08.135  INFO 96561 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.SubscriptionState    : [Consumer clientId=consumer-my-transformation-local-1, groupId=my-transformation-local] Resetting offset for partition CraveVideoTestLog-0 to position FetchPosition{offset=56464, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[localhost:9092 (id: 1 rack: null)], epoch=0}}.

我知道消费者无法获得偏移量。在这种情况下，消费者将引用自动偏移重置属性。如您所见，我已将其设置为最早，希望消费者从头开始阅读。但它没有。

应用程序. yml

spring:
  application:
    name: my-transformation
  kafka:
    bootstrap-servers: ${BOOTSTRAP_SERVERS:localhost:9092}
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
      retries: 2
      compression-type: snappy
    consumer:
      group-id: ${GROUP_ID:my-transformation}
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

在我的Java课上

    @SendTo("output-topic-here")
    @KafkaListener(topics = "${my-transformation.input.topic}", errorHandler = "myErrorHandler")
    public Message<?> process(Message<String> in, ConsumerRecord<String, String> record) throws Exception {
        log.info("Going to process event for key: {} offset: {} partition: {}", record.key(), record.offset(), record.partition());
    }

我尝试了一些东西。

• 我将自动偏移重置值设置为none。不出所料，它抛出了一个异常，抱怨偏移量
• 我创建了一个新的消费者群体。我可以看到，新的消费者群体是从融合的UI创建的。但是，同样，消息没有被消耗
• 我读了很多帖子。以下是一些与我面临的问题非常接近的问题。链接1链接2

我觉得我错过了一些很傻的东西，但我不知道是什么。

Answer 1

最后，我发现了我的消费者行为的原因。主题的保留策略设置为删除，保留时间设置为默认（1周）。

消费者停止消费记录的原因是因为主题的保留时间已经过去。因此，主题中的整个数据都被删除了，因此消费者没有什么可读的。当我再次将新记录推送到主题中时，消费者开始消费。

Answer 2

属性将按以下方式工作。

用例1：消费者启动并拥有auto.offset。reset＝最新，并且主题分区当前具有从某个范围到另一个范围的偏移量数据。消费者群体之前已经为该主题提交了一些补偿。消费者将从何处阅读？

Ans：这个消费者组和主题分区的偏移量已经提交，所以忽略auto.offset.reset属性

用例2:一个消费者启动并拥有auto.offset.reset=none，主题分区当前拥有从某个范围到另一个范围的偏移数据。消费者团体之前已经promise了一些话题的补偿。消费者将从哪里阅读？

答：auto.offset.reset=无意味着如果消费者正在恢复的偏移量已从Kafka中删除，则消费者将崩溃。

用例3：消费者具有auto.offset。reset＝最新，并且主题分区当前具有从某个范围到另一个范围的偏移量数据。消费者群体以前从未为该主题提交过补偿。消费者将从何处阅读？

回答：最新意味着数据检索将从偏移当前结束的位置开始。

用例4:消费者拥有auto.offset.reset=earliest，并且主题分区当前拥有从某个范围到另一个范围的偏移量的数据。消费者组织以前从未promise过这个话题的补偿。消费者将从哪里阅读？

答：最早意味着数据检索将从分区的开头开始

用例5：消费者启动并具有auto.offset.reset=最早，并且主题分区当前具有从某个范围到其他范围的偏移量的数据。消费者组之前已经为主题提交了一些偏移量。消费者将从哪里读取？

Ans：这个消费者组和主题分区的偏移量已经提交，所以忽略auto.offset.reset属性