我正在使用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);
    }
}

先谢谢你。