[NIFI-5117] AMQP Consumer: Error during creation of Flow File results in lost message

• Type: Bug
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 1.3.0, 1.4.0, 1.5.0, 1.6.0
• Component/s: Extensions
• Assignee: Unassigned
• Reporter: Edward Armes
• Created: 24/Apr/18 21:03
• Updated: 29/Jan/19 17:15

Description

The AMQP Consumer performs a "basicGet()". The was this basicGet is called results in the message being dequeued from the AMQP queue.

If a processor instances fails to submit a flow file to the output as a result in the case of an error in "session.write()" or the processor is unexpectedly halted before the flow file is created and persisted, the message consumer from an AMQP queue is lost and can't be recovered.

Reference: https://rabbitmq.github.io/rabbitmq-java-client/api/current/com/rabbitmq/client/Channel.html#basicGet-java.lang.String-boolean-:

A potential fix here would be to:

1. AMQPConsumer.java: Change the call "basicGet(this.queueName, true)" -> "basicGet(this.queueName, false)"
2. AMQPConsumer.java: New method that wraps the basicAck() and basicNack() methods to taking a long (the delivery tag) and boolean (successes) if successes is true basicAck() is called is false basicNack() with requeue is called
3. ConsumerAMQP.java: An additional call(s) to "consumer" to call the new method as needed in case of successes and error.