我有一个带有单个 kafka 消费者的 Spring Boot 应用程序，用于从某个主题获取消息。 但有时在处理消息时会发生错误。
我想像往常一样继续接收以下消息，同时能够不丢失该消息并接收它，例如，在修复后与消费者一起重新启动服务时。
可以这样做吗？
我知道我需要禁用自动提交并手动提交成功的消息，但是，在这种情况下，如果我不为此异常情况抛出任何异常并手动提交下一条成功的消息，那么我将丢失上一个不成功的，对吧？
【问题讨论】：
标签： spring-boot apache-kafka kafka-consumer-api
是的，您必须手动提交它们。您重试特定消息 2-3 次。如果重试后失败，那么您可以将这些消息移动到另一个主题，并在修复导致它失败的任何内容时使用这些消息。这不会阻塞您的队列，您也不会丢失消息。
【讨论】：
我想像往常一样继续收到以下消息，并在 同时能够不丢失该消息并接收它，因为 例如，下次服务与消费者一起重新启动时 修复后。
可以这样做吗？
您不需要手动提交，相反，您可以选择实现一种机制来进行重试，方法是在另一个队列中发布事件并延迟使用该事件。 =====> Amazon SQS 有delay Queue 但不幸的是kafka中没有这样的东西，你必须自己编写实现。
参考文章：
如果您正在重试消息处理，则消息的顺序可能会根据您的实现而改变。请记住这一点。
请记住，如果消息处理时间超过 max.poll.interval，kafka 确实会考虑消费者死亡。阅读this
【讨论】：
如果我正确理解了您的问题，您的假设是异常发生是由于您的代码中的问题，而不是在阅读来自该主题的消息时。在这种情况下，任何重试或其他措施都无法解决您的问题。
我们通常做的是捕获异常并将其发送到另一个 Kafka 主题。理想情况下，您还将添加一些有关异常发生的原因或代码部分的详细信息。修复应用程序中的错误后，您可以使用来自其他主题的消息。
我知道我需要禁用自动提交并手动提交成功的消息，但是，在这种情况下，如果我不为此异常情况抛出任何异常并手动提交下一条成功的消息，那么我将丢失上一个不成功的，对吧？
是的，你的理解是正确的。更准确地说，您不会“丢失”消息，但是一旦您的 ConsumerGroup 提交了更高的偏移量，它就不会在没有任何手动修改的情况下再次尝试读取较低的偏移量。
如果您只希望在极少数情况下会抛出异常，但您只是忽略它，您始终可以在纯 Kafka 中使用 consumer.seek() 方法
public void seek(TopicPartition partition, long offset)
从主题分区的特定偏移量开始读取。
【讨论】：