Kafka Stream处理长耗时事件：如何配置实现错误抛出与DLQ推送？

我有一个Spring Boot Kafka Stream应用，读取记录后执行HTTP调用并转换格式，再发送给生产者。核心代码如下：

final KStream<String, String> toSquare = builder.stream(eventTopic,
                Consumed.with(Serdes.String(), Serdes.String()));
toSquare.mapValues(recodProcessor::processMessage).to(notificationTopic,
                Produced.with(Serdes.String(), Serdes.String()));

有时processMessage方法处理输入事件耗时超过5分钟，而max.poll.interval.ms配置为5分钟，这会触发重平衡，还会导致对应分区出现延迟堆积。需要配置Kafka Stream，使其在该场景下抛出错误、将记录推送至DLQ并处理下一条记录。

1. 给processMessage添加超时控制 #

首先在HTTP调用环节加入超时限制，避免单个消息处理无限阻塞，超过阈值时直接抛出异常。可以用CompletableFuture配合超时实现：

public String processMessage(String message) throws TimeoutException, ExecutionException, InterruptedException {
    // 将HTTP调用包装成异步任务
    CompletableFuture<String> httpTask = CompletableFuture.supplyAsync(() -> {
        // 执行HTTP调用逻辑
        return yourHttpClient.callExternalService(message);
    });
    // 设置超时时间（建议略小于max.poll.interval.ms，比如290秒）
    return httpTask.get(290, TimeUnit.SECONDS);
}

2. 配置异常处理与DLQ转发 #

方式一：全局异常处理器（推荐） #

在Spring Boot配置类中定义自定义配置，指定异常处理逻辑：

@Bean
public StreamsBuilderFactoryBeanCustomizer streamsCustomizer() {
    return factoryBean -> {
        factoryBean.setStreamsConfig(new StreamsConfig(Map.of(
            StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class,
            StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, CustomDLQExceptionHandler.class
        )));
    };
}

实现自定义异常处理器，将失败消息发送到DLQ：

public class CustomDLQExceptionHandler implements ProductionExceptionHandler {
    private KafkaProducer<String, String> dlqProducer;

    @Override
    public void configure(Map<String, ?> configs) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, configs.get(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG));
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        dlqProducer = new KafkaProducer<>(props);
    }

    @Override
    public ProductionExceptionHandlerResponse handle(ProducerRecord<byte[], byte[]> record, Exception exception) {
        // 发送失败消息到DLQ主题
        dlqProducer.send(new ProducerRecord<>("your-dlq-topic", record.key(), record.value()));
        // 返回CONTINUE，让流继续处理下一条消息
        return ProductionExceptionHandlerResponse.CONTINUE;
    }
}

方式二：流处理逻辑内捕获异常 #

用flatMapValues替代mapValues，在逻辑内捕获异常并转发到DLQ：

@Autowired
private KafkaTemplate<String, String> dlqKafkaTemplate;

// 流处理逻辑
final KStream<String, String> toSquare = builder.stream(eventTopic,
                Consumed.with(Serdes.String(), Serdes.String()));

toSquare.flatMapValues(message -> {
    try {
        String processed = recodProcessor.processMessage(message);
        return Collections.singletonList(processed);
    } catch (Exception e) {
        // 推送原消息到DLQ
        dlqKafkaTemplate.send("your-dlq-topic", message);
        // 返回空列表，跳过当前消息的目标主题发送
        return Collections.emptyList();
    }
}).to(notificationTopic, Produced.with(Serdes.String(), Serdes.String()));

3. 调整Kafka Streams参数优化 #

• 适当调大max.poll.interval.ms，但必须配合超时控制确保不会无限阻塞：

spring:
  kafka:
    streams:
      properties:
        max.poll.interval.ms: 3600000  # 6分钟

• 调小max.poll.records，控制单次拉取消息数量，避免批次处理时间过长：

spring:
  kafka:
    streams:
      properties:
        max.poll.records: 10

4. 准备DLQ主题 #

提前创建DLQ主题，配置合适的分区数和副本数，可设置消息过期时间避免无限制存储失败消息。

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