Kafka恢复消费者无法收到第一条消息答案

【问题标题】：Kafka resume consumer can't receive the first messageKafka恢复消费者无法收到第一条消息
【发布时间】：2019-08-28 16:39:51
【问题描述】：

我需要暂停 Kafka 消费者从主题中消费消息，直到消息到达它的等待时间。为此，我在 Kafka 中使用了暂停/恢复方法。但是当我恢复时，暂停前消耗的第一条消息将不会再次收到。但是，自从我手动确认后，主题的偏移量仍然没有更新（滞后是一个）。

@StreamListener(ChannelName.MESSAGE_INPUT_RETRY_CHANNEL)
public void onMessageRetryReceive(org.springframework.messaging.Message<Message> message, @Header(KafkaHeaders.CONSUMER)KafkaConsumer<?,?> consumer){

    long waitTime = //Calculate the wait time of the message
    Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
    if(waitTime > 0){
        consumer.pause(Collections.singleton(new TopicPartition("message-retry-topic",0)));
    }else{
        messageProducer.sendMessage(message.getPayload());
        acknowledgment.acknowledge();
    }
}


@Bean
public ApplicationListener<ListenerContainerIdleEvent> idleListener() {
    return event -> {
        boolean isReady = //Logi to check if ready to resume
        if(isReady){
            event.getConsumer().resume(event.getConsumer().paused());
        }
    };
}

这与KafkaConsumer resume partition cannot continue to receive uncommitted messages 中提到的问题有关。但我不确定 seeks 方法如何有助于检索第一条消费消息。我正在使用弹簧云流。我需要一些建议

【问题讨论】：

标签： apache-kafka spring-cloud-stream

【解决方案1】：

您不调用acknowledgment.acknowledge(); 的事实并不意味着您的KafkaConsumer 实例不会在内存中保留上次使用的位置。

我们肯定需要为分区上的后续消费者提交偏移量。当前运行的消费者不需要提交此类信息，因为它具有自己的内存状态。

为了能够重新使用相同的记录，您需要执行seek() 操作。有关更多信息，请参阅文档：https://docs.spring.io/spring-kafka/docs/current/reference/html/#seek

【讨论】：