Springboot Kafka @Listener消费者暂停/恢复不起作用

Question

我有一个Spring靴Kafka消费者

为了避免重新平衡，我尝试在KafkaContainer上调用pause（）和resume（），但消费者总是在运行

@KafkaListener(id = "c1", topics = "${app.topics.topic1}", containerFactory = "listenerContainerFactory1")
public void poll(ConsumerRecord<String, String> record, Acknowledgment ack) {
    log.info("Received Message by consumer of topic1: " + value);
    String result = process(record.value());
    producer.sendMessage(result + " topic2");
    log.info("Message sent from " + topicIn + " to " + topicOut);
    ack.acknowledge();
    log.info("Offset committed by consumer 1");
}

private String process(String value) {
    try {
        pauseConsumer();
        // Perform time intensive network IO operations
        resumeConsumer();
    } catch (InterruptedException e) {
        log.error(e.getMessage());
    }
    return value;
}

private void pauseConsumer() throws InterruptedException {
    if (registry.getListenerContainer("c1").isRunning()) {
        log.info("Attempting to pause consumer");
        Objects.requireNonNull(registry.getListenerContainer("c1")).pause();
        Thread.sleep(5000);
        log.info("kafkalistener container state - " + registry.getListenerContainer("c1").isRunning());
    }
}

private void resumeConsumer() throws InterruptedException {
    if (registry.getListenerContainer("c1").isContainerPaused() || registry.getListenerContainer("c1").isPauseRequested()) {
        log.info("Attempting to resume consumer");
        Objects.requireNonNull(registry.getListenerContainer("c1")).resume();
        Thread.sleep(5000);
        log.info("kafkalistener container state - " + registry.getListenerContainer("c1").isRunning());
    }
}

我错过了什么吗？有人能指导我如何正确地达到要求的行为吗？

Answer 1

您正在侦听器线程上运行 process（） 方法，因此暂停/恢复不会产生任何影响;仅当侦听器线程退出侦听器方法（并且在处理了上一次轮询收到的所有记录之后）才会暂停。

将于本月晚些时候发布的下一个版本(2.9)有一个新属性< code>pauseImmediate，它使暂停在处理完当前记录后生效。

Answer 2

你可以这样试试。这对我有用

public class kafkaConsumer {
    public void run(String topicName) {
        try {
            Consumer<String, String> consumer = new KafkaConsumer<>(config);
            consumer.subscribe(Collections.singleton(topicName));
                while (true) {
                    try {
                        ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(80000));
                        for (TopicPartition partition : consumerRecords.partitions()) {
                            List<ConsumerRecord<String, String>> partitionRecords = consumerRecords.records(partition);
                            for (ConsumerRecord<String, String> record : partitionRecords) {

                                kafkaEvent = record.value();

                                consumer.pause(consumer.assignment());

                /** Implement Your Business Logic Here **/

                Once your processing done

                consumer.resume(consumer.assignment());

                                try {
                                    consumer.commitSync();
                                } catch (CommitFailedException e) {
                                }
                            }
                        }
                    } catch (Exception e) {
                        continue;
                    }
                }

        } catch (Exception e) {
        }
    }