暂停和恢复KafkaConsumer

Question

我要做的是暂停< code>KafkaConsumer，如果在使用消息的过程中出现错误。

这是我写的

@KafkaListener(...)
public void consume(
 @Header(KafkaHeaders.CONSUMER) KafkaConsumer<String,String> consumer,
 @Payload String message) {
  
    try {
        //consumer message
    } catch(Exception e) {
        saveConsumer(consumer);
        consumer.pause();
    }
}

然后我写了一个REST服务来恢复消费者

@RestController
@RequestMapping("/consumer")
class ConsumerRestController {
    @PostMapping("/resume")
    public void resume() {
        KafkaConsumer<String,String> consumer = getConsumer();
        if(consumer != null) {
            consumer.resume(consumer.paused());
        }
    }
}

现在，我有两个问题。第一个问题：当我打电话给消费者时。来自＜code＞@KafkaListener＜/code＞注释方法的pause（）会发生什么？消费者立即暂停，或者我可以接收到同一主题分区的其他偏移量上的其他消息。例如，我有偏移量为3的“message1”和偏移量为4的“mesmage2”，“message 1”会导致异常，“mesage2”会发生什么？它到底被消耗了吗？

第二个问题：从REST服务中恢复消费者会给出一个ConprestModificationException，因为Kafka消费者不是线程安全的。那么，我为什么要这样做呢？

Answer 1

不要直接让消费者停顿;请改为暂停容器。

@KafkaListener(id = "foo", ...)

@Autowired KafkaListenerEndpointRegistry;

...

registry.getListenerContainer("foo").pause();

暂停将在下一次轮询之前生效；如果您想立即暂停（并且不处理上次轮询的剩余记录），请在暂停后抛出一个异常（假设您正在使用现在默认的SeekToCurrentErrorHandler。