Kafka Streams应用调用外部服务异常时，如何重读Kafka消息？

核心逻辑 #

通过异常分类捕获+分层重试机制+延迟队列的组合方式，针对两类错误场景实现差异化处理：

• Final Service连接超时/不可用：将消息路由至延迟重试队列，到达指定时间后重新消费
• 技术异常（如服务内部错误）：触发即时重试，快速循环处理

具体实现步骤 #

1. 精准区分异常类型 #

在调用Final Service的代码中，针对性捕获两类异常并抛出自定义标记异常：

try {
    // 调用Final Service的REST请求
    restTemplate.postForObject(finalServiceUrl, message, String.class);
} catch (ResourceAccessException e) {
    // 场景1：连接超时/服务不可用
    if (e.getCause() instanceof ConnectTimeoutException) {
        throw new DelayedRetryException(message); // 标记为需延迟重试的异常
    }
} catch (HttpServerErrorException | HttpClientErrorException e) {
    // 场景2：技术异常（如5xx内部错误、可重试的4xx错误）
    throw new ImmediateRetryException(message); // 标记为需即时重试的异常
}

2. 即时重试配置 #

针对ImmediateRetryException，直接通过Kafka Streams内置的重试参数实现快速重试：
在Streams配置文件中添加：

# 最大重试次数，按需调整
max.retries=5
# 即时重试间隔设为极小值（如100ms）
retry.backoff.ms=100
processing.guarantee=exactly_once_v2

当抛出ImmediateRetryException时，Streams会立即触发重试，达到最大次数后可将消息转至死信队列（DLQ）归档。

3. 延迟重试实现（场景1） #

针对服务不可用的场景，采用「延迟重试主题+时间过滤消费」的方案：

• 创建专用延迟重试主题（如topic-retry-delayed），发送消息时附加延迟时间戳
• 启动独立消费者，仅消费已到重试时间的消息，再转发回原处理主题

示例代码（发送延迟重试消息）：

ProducerRecord<String, String> retryRecord = new ProducerRecord<>(
    "topic-retry-delayed",
    message.getKey(),
    message.getValue()
);
// 设置5分钟后重试：当前时间 + 300000ms
long delayTime = System.currentTimeMillis() + 300000;
retryRecord.headers().add("retry-time", String.valueOf(delayTime).getBytes());
kafkaProducer.send(retryRecord);

延迟消费者核心逻辑：

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
    for (ConsumerRecord<String, String> record : records) {
        long retryTime = Long.parseLong(new String(record.headers().lastHeader("retry-time").value()));
        if (System.currentTimeMillis() >= retryTime) {
            // 转发回原处理主题
            producer.send(new ProducerRecord<>("original-topic", record.key(), record.value()));
            consumer.commitSync();
        }
        // 未到重试时间，不提交偏移量，下次轮询再检查
    }
}

4. 死信队列兜底 #

当重试次数耗尽仍失败时，将消息路由至死信队列（如topic-dlq），用于后续人工排查：

streams.setUncaughtExceptionHandler((thread, throwable) -> {
    if (throwable instanceof DelayedRetryException || throwable instanceof ImmediateRetryException) {
        int retryCount = extractRetryCountFromMessage(throwable.getMessage());
        if (retryCount >= MAX_RETRY_LIMIT) {
            sendToDlq(throwable.getMessage());
        }
    }
});

关键注意点 #

• 确保Final Service支持幂等性，避免重试导致重复处理
• 延迟队列消费者需配置可靠的消费者组，防止消息丢失
• 重试次数、延迟间隔需根据业务实际压力调整，避免压垮下游服务

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