如何在下游微服务故障时停止Kafka Consumer读取消息并重试？

问题分析 #

你当前的代码虽然开启了手动提交enable.auto.commit=false，但存在两个核心问题：

1. 无论消息转发到data微服务是否成功，最终都会执行consumer.commitSync()，导致失败消息的offset被提交，重启后无法重发。
2. 没有检测data微服务状态的逻辑，服务故障时Consumer仍会持续调用poll()拉取消息，造成无效消费。

解决方案 #

要实现「data服务故障时停止消费、重启后重发未成功消息」的需求，需要从故障检测、异常控制消费状态、精准offset提交三个方面改造代码：

1. 新增data服务可用性检测

实现一个简单的健康检查逻辑，判断data服务是否正常：

private boolean isDataServiceAvailable() {
    // 替换为你的data服务健康检查逻辑，比如调用内部健康接口
    try {
        // 示例：使用HttpClient调用健康检查接口
        HttpClient client = HttpClient.newHttpClient();
        HttpRequest request = HttpRequest.newBuilder()
                .uri(URI.create("http://data-service/health"))
                .timeout(Duration.ofSeconds(5))
                .build();
        HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString());
        return response.statusCode() == 200;
    } catch (Exception e) {
        return false;
    }
}

2. 改造消费逻辑：控制消费状态与offset提交

修改原有消费循环，加入故障检测、异常处理、暂停/恢复消费的逻辑：

while (true) {
    // 检查data服务状态，不可用时暂停消费并等待
    if (!isDataServiceAvailable()) {
        consumer.pause(consumer.assignment());
        System.out.println("Data服务不可用，暂停消费");
        Thread.sleep(10000); // 每10秒重试一次
        continue;
    } else {
        // 服务恢复，恢复消费
        consumer.resume(consumer.assignment());
    }

    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    System.out.println("::::::::::" + records.count());
    
    boolean batchSuccess = true;
    OffsetAndMetadata lastCommittedOffset = null;
    
    for (ConsumerRecord<String, String> record : records) {
        try {
            readConsumerRecord(record); // 转发到data服务的业务逻辑
            // 记录当前成功的offset（offset+1表示下一条要消费的位置）
            lastCommittedOffset = new OffsetAndMetadata(record.offset() + 1);
        } catch (Exception e) {
            System.out.println("转发消息失败，offset: " + record.offset() + ", 错误信息: " + e.getMessage());
            batchSuccess = false;
            consumer.pause(consumer.assignment());
            break; // 跳出循环，等待服务恢复
        }
    }
    
    // 仅当批量所有消息都成功时，提交最后一个成功的offset
    if (batchSuccess && lastCommittedOffset != null) {
        consumer.commitSync(Map.of(records.iterator().next().topicPartition(), lastCommittedOffset));
    }
}

关键改动说明 #

• 精准offset提交：只有当消息转发成功时才记录offset，批量全部成功后再提交，确保失败消息的offset不会被确认，重启后能从失败位置重新消费。
• 消费状态控制：通过consumer.pause()和consumer.resume()在服务故障时停止拉取消息，避免无效资源消耗；服务恢复后自动恢复消费。
• 故障检测循环：定期检查data服务状态，避免持续无效poll。

额外注意事项 #

• 避免无限阻塞：如果转发失败次数过多，可以将消息转存到死信队列（DLQ），防止消费进程长期阻塞。
• 配置优化：确保session.timeout.ms（建议30-60秒）和heartbeat.interval.ms（建议session超时的1/3）配置合理，避免暂停消费时Consumer被踢出消费组。
• 配置覆盖检查：确认没有其他代码或框架自动配置覆盖enable.auto.commit=false（比如Spring Boot的spring.kafka.consumer.enable-auto-commit）。

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