Merge pull request #85 from commercetools/idempotent-nack

Make ack/nack idempotent on an empty batch
commercetools · Nov 8, 2024 · d7f254f · d7f254f
2 parents 3e70b19 + ca39f85
commit d7f254f
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 2 deletions.
diff --git a/aws/sqs/src/main/scala/com/commercetools/queue/aws/sqs/SQSMessageBatch.scala b/aws/sqs/src/main/scala/com/commercetools/queue/aws/sqs/SQSMessageBatch.scala
@@ -34,7 +34,9 @@ private class SQSMessageBatch[F[_], T](
 
   override def messages: Chunk[Message[F, T]] = payload
 
-  override def ackAll: F[List[MessageId]] =
+  override def ackAll: F[List[MessageId]] = if (payload.isEmpty) {
+    F.pure(List.empty[MessageId])
+  } else {
     F.fromCompletableFuture {
       F.delay {
         client.deleteMessageBatch(
@@ -52,8 +54,11 @@ private class SQSMessageBatch[F[_], T](
         )
       }
     }.map(res => res.failed().asScala.map(message => MessageId(message.id())).toList)
+  }
 
-  override def nackAll: F[List[MessageId]] =
+  override def nackAll: F[List[MessageId]] = if (payload.isEmpty) {
+    F.pure(List.empty[MessageId])
+  } else {
     F.fromCompletableFuture {
       F.delay {
         client.changeMessageVisibilityBatch(
@@ -74,4 +79,5 @@ private class SQSMessageBatch[F[_], T](
         )
       }
     }.map(res => res.failed().asScala.map(message => MessageId(message.id())).toList)
+  }
 }
diff --git a/testkit/src/main/scala/com/commercetools/queue/testkit/QueueSubscriberSuite.scala b/testkit/src/main/scala/com/commercetools/queue/testkit/QueueSubscriberSuite.scala
@@ -169,6 +169,21 @@ trait QueueSubscriberSuite extends CatsEffectSuite { self: QueueClientSuite =>
     }
   }
 
+  withQueue.test("messageBatch noop ack/nack on empty batches") { queueName =>
+    val client = clientFixture()
+    client.subscribe(queueName).puller.use { puller =>
+      for {
+        msgBatch <- puller.pullMessageBatch(10, waitingTime)
+        _ = assertEquals(msgBatch.messages.size, 0)
+        _ <- msgBatch.nackAll
+        _ <- msgBatch.ackAll
+        _ <- assertIOBoolean(
+          puller.pullMessageBatch(10, waitingTime).map(_.messages.isEmpty)
+        )
+      } yield ()
+    }
+  }
+
   withQueue.test("process respects the decision from the handler") { queueName =>
     val client = clientFixture()
     for {