Aws sqs nack

smallrye-reactive-messaging-aws-sqs/src/main/java/io/smallrye/reactive/messaging/aws/sqs/SqsConnector.java

@@ -49,6 +49,7 @@
 @ConnectorAttribute(name = "receive.request.retries", type = "long", direction = ConnectorAttribute.Direction.INCOMING, description = "If set to a positive number, the connector will try to retry the request that was not delivered successfully (with a potentially transient error) until the number of retries is reached. If set to 0, retries are disabled.", defaultValue = "2147483647")
 @ConnectorAttribute(name = "receive.request.pause.resume", type = "boolean", direction = ConnectorAttribute.Direction.INCOMING, description = "Whether the polling must be paused when the application does not request items and resume when it does. This allows implementing back-pressure based on the application capacity. Note that polling is not stopped, but will not retrieve any records when paused.", defaultValue = "true")
 @ConnectorAttribute(name = "ack.delete", type = "boolean", direction = ConnectorAttribute.Direction.INCOMING, description = "Whether the acknowledgement deletes the message from the queue", defaultValue = "true")
+@ConnectorAttribute(name = "nack.visibility-timeout", type = "int", direction = ConnectorAttribute.Direction.INCOMING, description = "The duration in seconds that the nacked messages to set Visibility Timeout", defaultValue = "0")
 public class SqsConnector implements InboundConnector, OutboundConnector, HealthReporter {
 
     @Inject

smallrye-reactive-messaging-aws-sqs/src/main/java/io/smallrye/reactive/messaging/aws/sqs/SqsInboundChannel.java

@@ -63,7 +63,7 @@ public SqsInboundChannel(SqsConnectorIncomingConfiguration conf, Vertx vertx, Sq
         this.customizer = customizer;
 
         SqsAckHandler ackHandler = conf.getAckDelete() ? new SqsDeleteAckHandler(client, queueUrlUni)
-                : new SqsNothingAckHandler();
+                : new SqsNothingAckHandler(client, queueUrlUni);
         PausablePollingStream<List<software.amazon.awssdk.services.sqs.model.Message>, software.amazon.awssdk.services.sqs.model.Message> pollingStream = new PausablePollingStream<>(
                 channel, request(null, 0), (messages, processor) -> {
                     if (messages != null) {

smallrye-reactive-messaging-aws-sqs/src/main/java/io/smallrye/reactive/messaging/aws/sqs/SqsMessage.java

@@ -3,7 +3,6 @@
 import static io.smallrye.reactive.messaging.providers.locals.ContextAwareMessage.captureContextMetadata;
 
 import java.util.Map;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.function.BiFunction;
 import java.util.function.Function;
@@ -118,9 +117,7 @@ public Function<Metadata, CompletionStage<Void>> getAckWithMetadata() {
 
     @Override
     public CompletionStage<Void> nack(Throwable reason, Metadata metadata) {
-        CompletableFuture<Void> nack = new CompletableFuture<>();
-        runOnMessageContext(() -> nack.complete(null));
-        return nack;
+        return ackHandler.handle(this).subscribeAsCompletionStage();
     }
 
     @Override

smallrye-reactive-messaging-aws-sqs/src/main/java/io/smallrye/reactive/messaging/aws/sqs/ack/SqsNothingAckHandler.java

@@ -3,12 +3,28 @@
 import io.smallrye.mutiny.Uni;
 import io.smallrye.reactive.messaging.aws.sqs.SqsAckHandler;
 import io.smallrye.reactive.messaging.aws.sqs.SqsMessage;
+import software.amazon.awssdk.services.sqs.SqsAsyncClient;
+import software.amazon.awssdk.services.sqs.model.ChangeMessageVisibilityRequest;
 
 public class SqsNothingAckHandler implements SqsAckHandler {
 
+    private final SqsAsyncClient client;
+    private final Uni<String> queueUrlUni;
+
+    public SqsNothingAckHandler(SqsAsyncClient client, Uni<String> queueUrlUni) {
+        this.client = client;
+        this.queueUrlUni = queueUrlUni;
+    }
+
     @Override
     public Uni<Void> handle(SqsMessage message) {
-        return Uni.createFrom().voidItem()
+        return queueUrlUni.map(queueUrl -> ChangeMessageVisibilityRequest.builder()
+                .queueUrl(queueUrl)
+                .receiptHandle(message.getMessage().receiptHandle())
+                .visibilityTimeout(0)
+                .build())
+                .chain(request -> Uni.createFrom().completionStage(() -> client.changeMessageVisibility(request)))
+                .replaceWithVoid()
                 .emitOn(message::runOnMessageContext);
     }
 }

smallrye-reactive-messaging-aws-sqs/src/test/java/io/smallrye/reactive/messaging/aws/sqs/locals/LocalPropagationAckTest.java

@@ -54,7 +54,7 @@ public void testChannelWithAckOnMessageContextNothingAck() {
         SqsClientProvider.client = getSqsClient();
         addBeans(SqsClientProvider.class);
         IncomingChannelWithAckOnMessageContext bean = runApplication(dataconfig()
-                .with("mp.messaging.incoming.data.ack.delete", false),
+                .with("mp.messaging.incoming.data.ack.delete", true),
                 IncomingChannelWithAckOnMessageContext.class);
         bean.process(i -> i + 1);
         await().until(() -> bean.getResults().size() >= 5);