如何在Apache Flink中处理瞬态/应用程序故障？

Question

我的Flink处理器监听Kafka，处理器中的业务逻辑涉及调用外部REST服务，服务可能会停止。我想将元组重放回处理器中，是否仍有这样做的方法？我使用了Storm，我们将能够使元组失败，这样元组就不会被确认。因此，相同的元组将重播到处理器。

在Flink中，一旦消息被Flink Kafka消费者消费，元组就会被自动确认。有很多方法可以解决这个问题。其中一种方法是将消息发布回同一队列/重试队列。但我正在寻找一种类似于Storm的解决方案。

我知道Flink的保存点/检查点将用于容错。但据我所知，在Flink失败的情况下，元组将重演。我想知道如何处理暂时性故障。

谢谢你

Answer 1

与外部系统交互时，我建议使用Flink的异步I/O运算符。它允许您执行异步任务而不会阻止运算符的执行。

如果要重试失败的操作，而不从最后一个成功的检查点重新启动Flink作业，那么我建议您自己实现重试策略。它可以如下所示：

new AsyncFunction<IN, OUT>() {
    @Override
    public void asyncInvoke(IN input, ResultFuture<OUT> resultFuture) throws Exception {
        FutureUtils
            .retrySuccessfulWithDelay(
                () -> triggerAsyncOperation(input),
                Time.seconds(1L),
                Deadline.fromNow(Duration.ofSeconds(10L)),
                this::decideWhetherToRetry,
                new ScheduledExecutorServiceAdapter(new DirectScheduledExecutorService()))
            .whenComplete((result, throwable) -> {
                if (result != null) {
                    resultFuture.complete(Collections.singleton(result));
                } else {
                    resultFuture.completeExceptionally(throwable);
                }
            })
    }
}

使用triggerAsyncAction封装您的异步操作和decdeWhetherToRetry封装您的重试策略。如果decdeWhetherToRetry返回true，则结果未来将使用此操作尝试的值完成。

如果异常完成了resultFuture，那么它将触发故障转移，这将导致作业从最后一个成功的检查点重新启动。