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:
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.
Describe the bug
The
AMQPObservableQueue
class incorrectly implements theList<String> ack(List<Message> messages);
function from theObservableQueue
interface. The javadoc ofObservableQueue#ack
states:The behaviour of
AMQPObservableQueue
is instead to return the successfully-ack'ed messages. The consequence of this appears to be thatcom.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
The associated
AMQPObservableQueueTest
test-class has atestAck
test which doesn't currently properly validate this behaviour insofar as all it does is validate that the returnedList<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