Quarkus Mutiny中Uni包装Future失败重试不生效问题排查

你当前的代码逻辑存在一个关键问题：在创建Uni之前就提前执行了performFutureOperation并获取了Future实例。这个Future在创建时就已经通过supplyAsync启动了异步执行，当Uni订阅它时，它可能已经处于失败状态。而Mutiny的重试机制是针对Uni的订阅过程，重试时只会重新订阅已有的Future实例，不会重新触发performFutureOperation的调用——因为原Future已经完成（失败），再次订阅只会直接返回失败结果，不会重新执行异步任务。

要让重试时重新执行异步操作，你需要延迟Future的创建，让每次订阅（包括重试）都生成一个新的Future实例。可以通过Uni.createFrom().future(Supplier<CompletableFuture<T>>)方法实现，该方法会在每次订阅时调用Supplier来创建新的Future：

public Uni<String> nonBlockingMethod(String val) {
    // 用Supplier延迟创建Future，每次订阅（重试）都会调用performFutureOperation
    return Uni.createFrom().future(() -> performFutureOperation(val))
            .onFailure().retry().atMost(1);
}

private static CompletableFuture<String> performFutureOperation(String val) {
    return CompletableFuture.supplyAsync(() -> {
        log.info("performFutureOperation called!");
        throw new RuntimeException("test exception");
        // return "processed "+val;
    });
}

这样配置后，当第一次调用失败时，重试机制会触发重新订阅Uni，此时Supplier会再次调用performFutureOperation，生成新的CompletableFuture并执行异步任务，你就能看到performFutureOperation called!被打印两次。

另外你的测试方法中，断言了成功返回"processed test-id"，但当前performFutureOperation总是抛出异常，所以测试会失败。如果要验证重试逻辑，可以调整测试断言失败状态，或者修改performFutureOperation让它在第二次调用时成功，比如：

private static AtomicInteger retryCount = new AtomicInteger(0);
private static CompletableFuture<String> performFutureOperation(String val) {
    return CompletableFuture.supplyAsync(() -> {
        log.info("performFutureOperation called!");
        if (retryCount.getAndIncrement() == 0) {
            throw new RuntimeException("test exception");
        }
        return "processed "+val;
    });
}

@Test
public void testNonBlockingMethod() {
    UniAssertSubscriber<String> subscriber = nonBlockingMethod("test-id").subscribe()
            .withSubscriber(UniAssertSubscriber.create());
    subscriber.awaitItem().assertItem("processed test-id").assertCompleted();
}

这样测试就能通过，同时验证重试机制生效。

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