Apache Camel Kafka消费者暂停后仍持续消费消息问题咨询

你的问题出在对Camel pausable组件的理解偏差上：pausable仅会暂停路由内的消息处理流程，但不会阻止Kafka Consumer底层的poll()调用。当canContinue()返回false时，路由停止处理已拉取的消息，但Kafka Consumer仍会持续从Broker拉取新消息，这些未被路由处理的消息会被丢弃；同时Kafka Consumer的本地偏移量（position）会随poll()自动前进，导致后续恢复时无法从暂停位置继续消费。

要实现「暂停时停止拉取消息、恢复时从原位置继续」的需求，需要直接控制Kafka Consumer的暂停与恢复，而非仅暂停路由处理。以下是具体修改方案：

1. 修正控制逻辑的线程安全性 #

首先将全局count替换为线程安全的AtomicInteger，避免多线程环境下的计数错误：

private final AtomicInteger count = new AtomicInteger(0);

2. 调整路由逻辑，控制Kafka Consumer的启停 #

修改路由，通过Camel的KafkaConsumer API直接暂停/恢复消费者，同时确保手动提交偏移量仅在消息处理完成后执行：

from("kafka:MykafkaTopic?brokers=localhost:9092&autoCommitEnable=false&allowManualCommit=true&maxPollRecords=1")
    .routeId("consumer")
    .process(exchange -> {
        // 获取Kafka Consumer实例
        KafkaConsumer<?, ?> consumer = exchange.getContext().getEndpoint("kafka:MykafkaTopic", KafkaEndpoint.class).getConsumer().getKafkaConsumer();
        boolean shouldContinue = canContinue();
        
        if (!shouldContinue) {
            // 暂停Kafka Consumer拉取消息
            consumer.pause(consumer.assignment());
            // 等待恢复信号（可根据业务逻辑调整等待逻辑，比如监听外部触发事件）
            while (!canContinue()) {
                Thread.sleep(1000);
            }
            // 恢复Kafka Consumer
            consumer.resume(consumer.assignment());
        }
    })
    .log("    on the topic ${headers[kafka.TOPIC]}")
    .log("    on the partition ${headers[kafka.PARTITION]}")
    .log("    with the offset ${headers[kafka.OFFSET]}")
    .to("direct:IncomingKafka")
    // 手动提交偏移量，确保消息处理完成后才提交
    .process(exchange -> {
        KafkaManualCommit manualCommit = exchange.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
        if (manualCommit != null) {
            manualCommit.commitSync();
        }
    })
    .end();

3. 修正canContinue()方法的线程安全实现 #

public boolean canContinue() {
    int currentCount = count.getAndIncrement();
    java.util.logging.Logger.getGlobal().info("count in canContinue::" + currentCount);
    
    if (currentCount < 4) {
        return true;
    }
    if (currentCount > 10) {
        return true;
    } else {
        return false;
    }
}

• 直接操作Kafka Consumer的pause()和resume()方法，能从底层阻止消息拉取，避免不必要的消息消费与丢弃。
• 必须确保手动提交偏移量仅在消息处理完成后执行，这样恢复消费时才能从最后提交的偏移量位置继续。
• 等待恢复的逻辑可根据业务需求调整，比如替换为监听外部信号（如数据库状态、MQ消息）而非简单的睡眠循环。

内容的提问来源于stack exchange，提问作者Dev2405