Spring Cloud Stream Kafka错误处理：如何实现故障分区消费暂停？

背景概述 #

我在Spring Boot应用中使用spring-cloud-stream-binder-kafka，通过Spring Cloud Function实现数据的upsert（更新插入）操作，处理成功后会将输入消息转发到输出主题。当前配置如下：

SPRING_CLOUD_FUNCTION_DEFINITION: upsertData

SPRING_CLOUD_STREAM_BINDINGS_upsertData-in-0_DESTINATION: UpsertDataEvent
spring.cloud.stream.bindings.upsertData-in-0.group: upsert-data-event-group
spring.cloud.stream.bindings.upsertData-in-0.content-type: application/json

SPRING_CLOUD_STREAM_BINDINGS_upsertData-out-0_DESTINATION: UpsertDataSuccessEvent
spring.cloud.stream.bindings.upsertData-out-0.producer.partitionCount: 1
spring.cloud.stream.bindings.upsertData-out-0.group: upsert-data-success-event-group
spring.cloud.stream.bindings.upsertData-out-0.content-type: application/json

客户提出三个需求，其中前两个已实现：

• 部署12个应用实例，每个实例对应Kafka的一个分区（共12个分区）
• 处理失败时将输入消息路由到错误主题，并携带错误信息作为消息头
• 待实现需求：当消息发送到对应消费者分区的错误主题后，该消费者暂停消费，直到错误主题被清理，同时保持应用的HTTP端点可用

现有实现代码 #

@Component("upsertData")
@AllArgsConstructor
@Slf4j
public class MyCloudFunction implements Function<Message<Data>, Data> {

    private final StreamBridge streamBridge;

    @Override
    public Data apply(Message<Data> message) {
        log.info("Upsert Data Message received: {}", message);
        Data requestPayload = message.getPayload();

        try {
            var searchResult = // 获取数据逻辑

            if (searchResult != null) {
                // 更新数据逻辑
            } else {
                // 创建数据逻辑
            }

            log.info("Upsert Data Event processed successfully");

            return requestPayload;
        } catch (Exception e) {
            log.error("Error processing message. Sending to error queue.", e);
            sendMessageToErrorTopic(e, requestPayload);
            return null;
        }
    }

    private void sendMessageToErrorTopic(Exception e, Data requestPayload) {
        Message<Data> errorMessage =
            MessageBuilder.withPayload(requestPayload)
                .setHeader("error-reason", e.getMessage()) // 注意：直接存Exception对象可能序列化失败，建议存错误信息字符串
                .build();
        streamBridge.send("UpsertDataFailureEvent", errorMessage);
    }

}

需求3的具体实现方案 #

核心思路 #

要实现"错误消息发送后暂停对应分区消费，直到错误主题清理"的逻辑，需要结合Kafka消费者容器控制、错误主题的状态检测以及Spring Cloud Stream的绑定管理来实现，具体步骤如下：

1. 改造错误消息发送逻辑，记录分区信息 #

首先在处理消息时获取原消息的Kafka分区ID，发送错误消息时携带该分区信息，同时触发对应分区的消费暂停：

@Component("upsertData")
@AllArgsConstructor
@Slf4j
public class MyCloudFunction implements Function<Message<Data>, Data> {

    private final StreamBridge streamBridge;
    private final BinderAwareChannelResolver channelResolver;
    // 记录分区暂停状态，key为分区ID，value为是否暂停
    private final ConcurrentHashMap<Integer, Boolean> partitionPauseStatus = new ConcurrentHashMap<>();

    @Override
    public Data apply(Message<Data> message) {
        // 获取原消息的Kafka分区ID
        Integer originalPartition = message.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID, Integer.class);
        log.info("Upsert Data Message received from partition: {}", originalPartition);
        Data requestPayload = message.getPayload();

        try {
            // 原有业务逻辑...
            var searchResult = // 获取数据逻辑
            if (searchResult != null) {
                // 更新数据逻辑
            } else {
                // 创建数据逻辑
            }
            log.info("Upsert Data Event processed successfully");
            return requestPayload;
        } catch (Exception e) {
            log.error("Error processing message from partition {}. Sending to error queue.", originalPartition, e);
            sendMessageToErrorTopic(e, requestPayload, originalPartition);
            // 暂停当前分区的消费
            pausePartitionConsumption(originalPartition);
            return null;
        }
    }

    private void sendMessageToErrorTopic(Exception e, Data requestPayload, Integer originalPartition) {
        Message<Data> errorMessage =
            MessageBuilder.withPayload(requestPayload)
                .setHeader("error-reason", e.getMessage())
                .setHeader("original-partition", originalPartition) // 携带原分区信息
                .build();
        streamBridge.send("UpsertDataFailureEvent", errorMessage);
    }

    // 暂停指定分区的消费
    private void pausePartitionConsumption(Integer partition) {
        MessageChannel inputChannel = channelResolver.resolveDestination("upsertData-in-0");
        if (inputChannel instanceof SubscribableChannel subscribableChannel) {
            subscribableChannel.getSubscribers().forEach(subscriber -> {
                if (subscriber instanceof KafkaMessageListenerContainer container) {
                    TopicPartition targetPartition = new TopicPartition("UpsertDataEvent", partition);
                    container.pause(Collections.singletonList(targetPartition));
                    partitionPauseStatus.put(partition, true);
                    log.info("Paused consumption for partition {}", partition);
                }
            });
        }
    }

    // 对外暴露分区状态，供恢复逻辑使用
    public ConcurrentHashMap<Integer, Boolean> getPartitionPauseStatus() {
        return partitionPauseStatus;
    }
}

2. 实现分区消费恢复逻辑 #

提供两种恢复方式，可根据业务需求选择：

方案一：手动HTTP触发恢复（推荐，可控性强）

创建HTTP接口，允许运维人员在清理错误主题后，手动指定分区恢复消费：

@RestController
@RequestMapping("/consumer-control")
@AllArgsConstructor
@Slf4j
public class ConsumerControlController {

    private final MyCloudFunction myCloudFunction;
    private final BinderAwareChannelResolver channelResolver;

    @PostMapping("/resume/{partition}")
    public ResponseEntity<String> resumePartition(@PathVariable Integer partition) {
        MessageChannel inputChannel = channelResolver.resolveDestination("upsertData-in-0");
        if (inputChannel instanceof SubscribableChannel subscribableChannel) {
            subscribableChannel.getSubscribers().forEach(subscriber -> {
                if (subscriber instanceof KafkaMessageListenerContainer container) {
                    TopicPartition targetPartition = new TopicPartition("UpsertDataEvent", partition);
                    container.resume(Collections.singletonList(targetPartition));
                    myCloudFunction.getPartitionPauseStatus().put(partition, false);
                    log.info("Resumed consumption for partition {}", partition);
                }
            });
            return ResponseEntity.ok("Successfully resumed partition " + partition);
        }
        return ResponseEntity.badRequest().body("Failed to resume partition " + partition);
    }
}

方案二：自动监听错误主题清理

通过定时任务检查错误主题的分区消息量，当对应分区无消息时自动恢复原分区消费：

@Component
@AllArgsConstructor
@Slf4j
public class ErrorTopicMonitor {

    private final KafkaAdmin kafkaAdmin;
    private final MyCloudFunction myCloudFunction;
    private final BinderAwareChannelResolver channelResolver;
    // 注入Kafka配置属性
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Scheduled(fixedRate = 60000) // 每分钟检查一次，可根据需求调整
    public void checkErrorTopicAndResume() {
        DescribeTopicsResult describeTopics = kafkaAdmin.describeTopics(Collections.singletonList("UpsertDataFailureEvent"));
        describeTopics.all().whenComplete((topicMap, throwable) -> {
            if (throwable != null) {
                log.error("Failed to describe error topic", throwable);
                return;
            }
            TopicDescription topicDescription = topicMap.get("UpsertDataFailureEvent");
            if (topicDescription == null) {
                log.warn("Error topic UpsertDataFailureEvent does not exist");
                return;
            }
            // 遍历错误主题的每个分区
            topicDescription.partitions().forEach(partitionInfo -> {
                Integer errorPartitionId = partitionInfo.partition();
                // 查询该分区是否有消息
                try (KafkaConsumer<String, Object> consumer = createKafkaConsumer()) {
                    TopicPartition errorTopicPartition = new TopicPartition("UpsertDataFailureEvent", errorPartitionId);
                    consumer.assign(Collections.singletonList(errorTopicPartition));
                    long endOffset = consumer.endOffsets(Collections.singletonList(errorTopicPartition)).get(errorTopicPartition);
                    long beginningOffset = consumer.beginningOffsets(Collections.singletonList(errorTopicPartition)).get(errorTopicPartition);
                    // 无消息且原分区处于暂停状态，则恢复消费
                    if (endOffset == beginningOffset) {
                        Integer originalPartition = errorPartitionId; // 假设错误主题与原主题分区数一致，一一对应
                        if (myCloudFunction.getPartitionPauseStatus().getOrDefault(originalPartition, false)) {
                            resumeOriginalPartitionConsumption(originalPartition);
                        }
                    }
                } catch (Exception e) {
                    log.error("Failed to check error topic partition {}", errorPartitionId, e);
                }
            });
        });
    }

    private KafkaConsumer<String, Object> createKafkaConsumer() {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "error-topic-monitor-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        return new KafkaConsumer<>(props);
    }

    private void resumeOriginalPartitionConsumption(Integer originalPartition) {
        MessageChannel inputChannel = channelResolver.resolveDestination("upsertData-in-0");
        if (inputChannel instanceof SubscribableChannel subscribableChannel) {
            subscribableChannel.getSubscribers().forEach(subscriber -> {
                if (subscriber instanceof KafkaMessageListenerContainer container) {
                    TopicPartition targetPartition = new TopicPartition("UpsertDataEvent", originalPartition);
                    container.resume(Collections.singletonList(targetPartition));
                    myCloudFunction.getPartitionPauseStatus().put(originalPartition, false);
                    log.info("Automatically resumed consumption for partition {}", originalPartition);
                }
            });
        }
    }
}

3. 关键注意事项 #

• 错误消息头不要直接存储Exception对象，避免序列化失败，建议存储错误信息字符串（如e.getMessage()或e.toString()）
• 每个应用实例仅负责自身绑定的Kafka分区，暂停/恢复操作需精准定位到对应分区
• HTTP控制接口需添加权限校验，防止未授权操作
• 自动监听方案中，定时任务频率需根据业务场景调整，避免频繁调用Kafka API造成性能开销

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