Branched Exception Handling In Mutiny reactive

[prakashelango]

Describe the bug

import io.quarkus.logging.Log;
import io.smallrye.mutiny.Uni;
import software.amazon.awssdk.enhanced.dynamodb.DynamoDbAsyncTable;
import software.amazon.awssdk.enhanced.dynamodb.Expression;
import software.amazon.awssdk.enhanced.dynamodb.model.PutItemEnhancedRequest;
import software.amazon.awssdk.services.dynamodb.model.ConditionalCheckFailedException;
import software.amazon.awssdk.services.dynamodb.model.InternalServerErrorException;
import software.amazon.awssdk.services.dynamodb.model.ProvisionedThroughputExceededException;
import software.amazon.awssdk.services.dynamodb.model.RequestLimitExceededException;

public class GenericRepositoryHelper<T> {

    private final DynamoDbAsyncTable<T> table;

    public GenericRepositoryHelper(DynamoDbAsyncTable<T> table) {
        this.table = table;
    }

    public Uni<T> save(final T entity) {
        return Uni.createFrom().completionStage(table.putItemWithResponse(toPutItemRequest(entity)))
            .map(resp -> entity)
            .onFailure(ProvisionedThroughputExceededException.class).recoverWithUni(fail -> handleRetryableException(fail, "ProvisionedThroughputExceededException occurred during put item", entity))
            .onFailure(RequestLimitExceededException.class).recoverWithUni(fail -> handleRetryableException(fail, "RequestLimitExceededException occurred during put item operation", entity))
            .onFailure(InternalServerErrorException.class).recoverWithUni(fail -> handleRetryableException(fail, "InternalServerErrorException occurred during put item operation", entity))
            .onFailure(ConditionalCheckFailedException.class).recoverWithUni(fail -> {
                Log.errorf("ConditionalCheckFailedException occurred during put item operation: %s", fail.getMessage());
                return Uni.createFrom().failure(new NonRetryableException(fail.getMessage(), fail));
            })
            .onFailure().recoverWithUni(fail -> handleRetryableException(fail, "GenericException occurred during put item operation", entity));
    }

    private PutItemEnhancedRequest<T> toPutItemRequest(final T entity) {
        return PutItemEnhancedRequest.builder(entity.getClass())
            .item(entity)
            .conditionExpression(Expression.builder().expression("attribute_not_exists(pk) and attribute_not_exists(sk)").build())
            .build();
    }

    private Uni<T> handleRetryableException(final Throwable failure, final String message, final T entity) {
        Log.errorf("%s %s", message, failure);
        if (failure instanceof NonRetryableException) {
            return Uni.createFrom().failure(failure);
        }
        return Uni.createFrom().failure(new RetryableException(failure.getMessage(), new UnprocessedItemsException(entity)));
    }
}

 .onFailure(ConditionalCheckFailedException.class).recoverWithUni(fail -> { even with specified exception generic exception handler gets invoked which leads to add another condition in retryableException expectation is similar to try catch block will be preferred
                Log.errorf("ConditionalCheckFailedException occurred during put item operation: %s", fail.getMessage());
                return Uni.createFrom().failure(new NonRetryableException(fail.getMessage(), fail));
            })
            .onFailure().recoverWithUni(fail -> handleRetryableException(fail, "GenericException occurred during put item operation", entity));

Note: Changing order of exception handling pipes doesn't take any effect

Expected behavior

No response

Actual behavior

No response

How to Reproduce?

No response

Output of uname -a or ver

No response

Output of java -version

21

Quarkus version or git rev

No response

Build tool (ie. output of mvnw --version or gradlew --version)

gradle

Additional information

No response

[quarkus-bot]

/cc @cescoffier (mutiny), @jponge (mutiny)

[jponge]

Can you please provide an explanation of what you think is wrong?

[prakashelango]

.onFailure(ProvisionedThroughputExceededException.class).recoverWithUni(fail -> handleRetryableException(fail, "ProvisionedThroughputExceededException occurred during put item", entity))
            .onFailure(RequestLimitExceededException.class).recoverWithUni(fail -> handleRetryableException(fail, "RequestLimitExceededException occurred during put item operation", entity))
            .onFailure(InternalServerErrorException.class).recoverWithUni(fail -> handleRetryableException(fail, "InternalServerErrorException occurred during put item operation", entity))
            .onFailure(ConditionalCheckFailedException.class).recoverWithUni(fail -> {
                Log.errorf("ConditionalCheckFailedException occurred during put item operation: %s", fail.getMessage());
                return Uni.createFrom().failure(new NonRetryableException(fail.getMessage(), fail));
            })
            .onFailure().recoverWithUni(fail -> handleRetryableException(fail, "GenericException occurred during put item operation", entity));

Given that failure of Ex: ConditionalCheckFailedException reactive pipeline should return Uni of (NonRetryableException) method execution should be passed back to caller. whereas Uni(NonRetryableException) being carried over to generic exception handler .onFailure().recoverWithUni( leading to invocation of handleRetryableException one more which adds additional ```
if (failure instanceof NonRetryableException) {
return Uni.createFrom().failure(failure);
}

[prakashelango]

@jponge can help on this.

[cescoffier]

There is an open PR on Mutiny to add that feature. However, it's a breaking change, and it will have to wait a Mutiny 3.0.
See smallrye/smallrye-mutiny#1848