如何在最大RestartSource后捕获alpakka Kafka源流故障

0kjbasz6 于 2023-10-15 发布在 Apache

关注(0)|答案(1)|浏览(112)

如何在最大数量的重启后捕获RestartSource的错误？我想在源失败的最大次数之后做一些事情。我可以在日志中看到源重新启动。我试着添加一个withAttributes，但它从未被调用。

return RestartSource.onFailuresWithBackoff(restartSettings, () -> Consumer
                .committableSource(getConsumerSettings(), topics)
                .log("error on receiver topic")
                .mapMaterializedValue(ctrl -> {
                    control = ctrl;
                    return NotUsed.getInstance();
                })
                .withAttributes(ActorAttributes.withSupervisionStrategy(e -> {
                    log.error("Stream has failed", e);
                    return (Supervision.Directive) Supervision.stop();
                })));

任何建议将不胜感激。

apache-kafka

来源：https://stackoverflow.com/questions/77173979/how-to-capture-alpakka-kafka-source-stream-failure-after-max-restartsource

1条答案

按热度按时间

zf9nrax11#

好了，我搞定了。在这种情况下，属性似乎从未被调用，但这会导致整个流失败，因此结果将在附加流和接收器完成上。在我的情况下，我有这样的东西：

streamCompletion = 
    createSource()
    ...
    runWith(Sink.ignore(), actorSystem);

然后，我们可以像下面这样在streamCompletion上获得失败：

streamCompletion.whenComplete((done, throwable) -> {
        if (throwable != null) {
            log.error("Some error occurred {} {}", done, throwable.getMessage(), throwable);
        }
    });

赞(0）回复(0）举报 2023-10-15

我来回答

如何在最大RestartSource后捕获alpakka Kafka源流故障

1条答案

相关问题

热门标签

最新问答