Issue#3408 fix AMQPObservableQueue behaviour to return failed message…

…s, not successful ones
Netflix · Jul 6, 2023 · 5a3792e · 5a3792e
1 parent c0fe98c
commit 5a3792e
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 8 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
@@ -334,12 +334,12 @@ void runObserve(
     }
 
     @Test
-    public void testGetMessagesFromExistingExchangeWithDefaultConfiguration() throws IOException, TimeoutException {
+    public void testGetMessagesFromExistingExchangeWithDefaultConfiguration()
+            throws IOException, TimeoutException {
         // Mock channel and connection
         Channel channel = mockBaseChannel();
         Connection connection = mockGoodConnection(channel);
-        testGetMessagesFromExchangeAndDefaultConfiguration(
-                channel, connection, true, true);
+        testGetMessagesFromExchangeAndDefaultConfiguration(channel, connection, true, true);
     }
 
     @Test
@@ -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(