如何捕获RestartSource达最大重启次数后的Alpakka Kafka源流故障？

你的核心问题在于：内部流的SupervisionStrategy无法捕获RestartSource达到最大重启次数后的终态错误——因为RestartSource会接管内部流的失败并重启，直到耗尽重试次数，此时整个RestartSource流才会终止，而非内部流触发终止指令。

正确的处理方式 #

要捕获最大重启次数后的错误并执行自定义操作，需要监听整个RestartSource流的终止状态，而非在内部流上设置SupervisionStrategy。以下是两种可行方案：

方案1：使用watchTermination监听流终止

通过watchTermination获取整个流的终止Future，当RestartSource耗尽重试次数后，该Future会以失败状态完成，此时即可执行自定义逻辑：

return RestartSource.onFailuresWithBackoff(restartSettings, () -> 
        Consumer.committableSource(getConsumerSettings(), topics)
                .log("error on receiver topic")
                .mapMaterializedValue(ctrl -> {
                    control = ctrl;
                    return NotUsed.getInstance();
                })
)
// 监听整个流的终止状态
.watchTermination((materializedValue, terminationFuture) -> {
    terminationFuture.whenComplete((unused, throwable) -> {
        if (throwable != null) {
            log.error("RestartSource已达最大重启次数，流终止", throwable);
            // 在这里执行你需要的后续操作，比如告警、清理资源等
        }
    });
    return materializedValue;
})
// 可选：设置元素级错误的处理策略（仅处理单个元素处理失败，不触发流终止）
.withAttributes(ActorAttributes.withSupervisionStrategy(e -> {
    log.error("元素处理失败", e);
    return Supervision.resume(); // 根据需求选择resume/stop/restart
}));

方案2：在流末端添加Sink.onComplete

如果你的流有明确的Sink，可以在末端通过Sink.onComplete捕获终止信号：

RestartSource.onFailuresWithBackoff(restartSettings, () -> 
        Consumer.committableSource(getConsumerSettings(), topics)
                .log("error on receiver topic")
                .mapMaterializedValue(ctrl -> {
                    control = ctrl;
                    return NotUsed.getInstance();
                })
)
// 替换成你实际的业务处理Sink
.to(Sink.foreach(record -> { /* 业务逻辑 */ }))
// 监听流的完成状态
.run(streamMaterializer)
.whenComplete((unused, throwable) -> {
    if (throwable != null) {
        log.error("RestartSource已达最大重启次数，流终止", throwable);
        // 执行自定义操作
    }
});

关键原理说明 #

1. 内部流的SupervisionStrategy作用范围：仅处理单个元素处理过程中抛出的异常，不会影响整个源的生命周期。当整个源（比如Kafka Consumer连接失败）崩溃时，RestartSource会直接捕获并重启内部流，不会触发内部流的SupervisionStrategy。
2. RestartSource的终态逻辑：当耗尽restartSettings中配置的最大重启次数后，RestartSource会停止重试，整个流以失败状态终止，此时通过watchTermination或流的materialized Future可以捕获到这个终止信号。

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