Merge pull request #242 from jcockrill-rq/fix-AMQPObservableQueue-ack…

…-behaviour Issue#241 fix AMQPObservableQueue behaviour to return failed message…
Netflix · Sep 2, 2023 · 941f072 · 941f072
2 parents c67c136 + 5a3792e
commit 941f072
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 5 deletions.
diff --git a/...eue/amqp/src/main/java/com/netflix/conductor/contribs/queue/amqp/AMQPObservableQueue.java b/...eue/amqp/src/main/java/com/netflix/conductor/contribs/queue/amqp/AMQPObservableQueue.java
@@ -187,16 +187,16 @@ public Address[] getAddresses() {
     }
 
     public List<String> ack(List<Message> messages) {
-        final List<String> processedDeliveryTags = new ArrayList<>();
+        final List<String> failedMessages = new ArrayList<>();
         for (final Message message : messages) {
             try {
                 ackMsg(message);
-                processedDeliveryTags.add(message.getReceipt());
             } catch (final Exception e) {
                 LOGGER.error("Cannot ACK message with delivery tag {}", message.getReceipt(), e);
+                failedMessages.add(message.getReceipt());
             }
         }
-        return processedDeliveryTags;
+        return failedMessages;
     }
 
     public void ackMsg(Message message) throws Exception {

diff --git a/...amqp/src/test/java/com/netflix/conductor/contribs/queue/amqp/AMQPObservableQueueTest.java b/...amqp/src/test/java/com/netflix/conductor/contribs/queue/amqp/AMQPObservableQueueTest.java
@@ -386,8 +386,9 @@ public void testAck() throws IOException, TimeoutException {
         msg.setPayload("Payload");
         msg.setReceipt("1");
         messages.add(msg);
-        List<String> deliveredTags = observableQueue.ack(messages);
-        assertNotNull(deliveredTags);
+        List<String> failedMessages = observableQueue.ack(messages);
+        assertNotNull(failedMessages);
+        assertTrue(failedMessages.isEmpty());
     }
 
     private void testGetMessagesFromExchangeAndDefaultConfiguration(