Repository: netflix/conductor-community
Issue: AMQPObservableQueue 'ack' implementation is inconsistent with the interface

**Describe the bug**

The `AMQPObservableQueue` class incorrectly implements the `List<String> ack(List<Message> messages);` function from the `ObservableQueue` interface. The javadoc of `ObservableQueue#ack` states:

```
@return the id of the ones which could not be ack'ed
```

The behaviour of `AMQPObservableQueue` is instead to return the successfully-ack'ed messages. The consequence of this appears to be that `com.netflix.conductor.core.events.queue.DefaultEventQueueProcessor` in conductor-core generates a lot of log-error spam because of this line:

https://github.com/Netflix/conductor/blob/631a04dd790f5d46e76691b47a748cad77e0e20d/core/src/main/java/com/netflix/conductor/core/events/queue/DefaultEventQueueProcessor.java#L155

```java
List<String> failures = queue.ack(Collections.singletonList(msg));
if (!failures.isEmpty()) {
    LOGGER.error("Not able to ack the messages {}", failures);
}
```

The associated `AMQPObservableQueueTest` test-class has a `testAck` test which doesn't currently properly validate this behaviour insofar as all it does is validate that the returned `List<String>` is not-null. It doesn't validate that it is empty.

**Details**
Conductor version: current version on main
Persistence implementation: N/A
Queue implementation: N/A
Lock: N/A
Workflow definition: N/A
Task definition: N/A
Event handler definition: AMQP


**To Reproduce**

See associated test-case.

**Expected behavior**

`AMQPObservableQueue#ack` should only return messages that have thrown an error.

**Screenshots**

N/A

**Additional context**

N/A

Changed files:
event-queue/amqp/src/main/java/com/netflix/conductor/contribs/queue/amqp/AMQPObservableQueue.java
event-queue/amqp/src/test/java/com/netflix/conductor/contribs/queue/amqp/AMQPObservableQueueTest.java

diff --git a/event-queue/amqp/src/main/java/com/netflix/conductor/contribs/queue/amqp/AMQPObservableQueue.java b/event-queue/amqp/src/main/java/com/netflix/conductor/contribs/queue/amqp/AMQPObservableQueue.java
index ff26fc7c..c1ecadd7 100644
--- a/event-queue/amqp/src/main/java/com/netflix/conductor/contribs/queue/amqp/AMQPObservableQueue.java
+++ b/event-queue/amqp/src/main/java/com/netflix/conductor/contribs/queue/amqp/AMQPObservableQueue.java
@@ -187,16 +187,16 @@ public class AMQPObservableQueue implements ObservableQueue {
     }
 
     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/event-queue/amqp/src/test/java/com/netflix/conductor/contribs/queue/amqp/AMQPObservableQueueTest.java b/event-queue/amqp/src/test/java/com/netflix/conductor/contribs/queue/amqp/AMQPObservableQueueTest.java
index 15893be8..0d936ccf 100644
--- a/event-queue/amqp/src/test/java/com/netflix/conductor/contribs/queue/amqp/AMQPObservableQueueTest.java
+++ b/event-queue/amqp/src/test/java/com/netflix/conductor/contribs/queue/amqp/AMQPObservableQueueTest.java
@@ -334,12 +334,12 @@ public class AMQPObservableQueueTest {
     }
 
     @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 class AMQPObservableQueueTest {
         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(