未提交的消息不再被消费答案

【问题标题】：Uncommitted message not getting consumed again未提交的消息不再被消费
【发布时间】：2018-05-23 14:02:28
【问题描述】：

我正在使用 Kafka 侦听器来接收消费者消息。我已启用手动确认。我已经使用 Acknowledgement.acknowledge() 方法来提交偏移量以防万一成功。由于某些API调用失败而导致异常的情况，我没有提交。

我有点困惑，因为除非我重新启动我的应用程序，否则未提交的消息不会再次被使用。我是 kafka 的新手，所以我确信我错过了一些非常基本的东西，但为此搜索是徒劳的。有人可以解释或指出我可以知道这里有什么问题的正确来源吗？

 @KafkaListener(topics = "temp_topic", groupId = "temp-consumer", containerFactory = "tempListenerContainerFactory")
public void receive(ConsumerRecord<?, ?> consumerRecord,
                                   Acknowledgment acknowledgment) {
    try {
        log.info("Consuming message : {} ", consumerRecord.value().toString());
        String message = consumerRecord.value().toString();
         objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
         Payload payload = objectMapper.readValue(message,Payload.class);
         tempService.processEvent(payload);
         log.info("Message: {} successfully consumed",message);
         acknowledgment.acknowledge();
    } catch (Exception e) {
        log.error("Error while listening to kafka event: ", e);
    }
}

提前谢谢你。

【问题讨论】：

标签： java apache-kafka

【解决方案1】：

确认不会改变您的消费者实际阅读的时间点。
它仅向某些存储（zookeeper 或带有偏移量的 kafka 主题）提交偏移量，而不对消费者跟踪的偏移量进行任何更改 - 例如，您可能不会提交当前偏移量并通过对某个下一个偏移量进行 ack 来覆盖它。

有一些方法可以尝试处理失败的消息：

您可以为特定类型的故障（例如超时、反序列化错误）做好准备 - 在侦听器中捕获它们并采取正确的措施（例如等待/重试或应用程序失败 - 取决于问题的性质）。
您可以跳过失败的消息 - 在侦听器中捕获错误，将它们放入某个存储以存储失败的消息（请参阅死信队列），然后对它们进行确认。

【讨论】：

【解决方案2】：

我阅读了一些内容并找到了一种处理失败消息的方法。我们制作了 3 个主题

主题
副话题
死话题

想法是使用来自主主题的消息，如果由于任何原因失败，将其推送到副主题，它将在 tn 次/strong> 时间间隔。

如果在侧队列中的所有重试后失败，则将其推送到死主题，在那里它将简单地通知（使用电子邮件或任何其他服务）有关失败的信息。

我不确定这是否是标准，但这是迄今为止我所阅读的所有内容中最有意义的。

如果可以，请随时添加或更正。

【讨论】：