NIFI-15483: Route PublishAMQP to failure on undeliverable messages instead of silent success

[rakesh-rsky]

Summary

PublishAMQP silently routes FlowFiles to REL_SUCCESS even when the AMQP broker cannot deliver the message, causing silent data loss.

Two failure modes are addressed:

1. Undeliverable message (basic.return) — broker returns the message when no queue is bound to the exchange/routing-key. The fix uses AMQP Publisher Confirms + basic.return to detect and surface delivery failures, routing the FlowFile to REL_FAILURE.

2. Exchange not found (ShutdownSignalException) — when the exchange does not exist, the broker closes the channel with 404 NOT_FOUND, causing waitForConfirms() to throw ShutdownSignalException. This is now caught and converted to AMQPException so the FlowFile routes to REL_FAILURE instead of causing an unhandled processor failure.

Testing

• Publish to an exchange with no bound queues → FlowFile routes to failure with the broker's return reason
• Publish to a non-existent exchange → FlowFile routes to failure instead of unhandled processor error
• Normal publish (queue bound) → still routes to success
• Added regression tests for all failure scenarios — verified to fail against unfixed code

Fixes: https://issues.apache.org/jira/browse/NIFI-15483

[turcsanyip]

@rakesh-rsky Thanks for working on this issue. Please note the unit tests are failing.

I tried the new error handling in my local environment but I'm getting the following runtime error (instead of routing the FlowFile to failure):

2026-05-06 14:34:55,697 ERROR [Timer-Driven Process Thread-4] o.a.nifi.amqp.processors.PublishAMQP PublishAMQP[id=4fd4ee4e-8e3a-3969-7db4-953498193e4d] Processor failure
com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'dummy' in vhost '/', class-id=60, method-id=40)
	at com.rabbitmq.client.impl.ChannelN.waitForConfirms(ChannelN.java:219)
	at org.apache.nifi.amqp.processors.AMQPPublisher.publish(AMQPPublisher.java:108)
	at org.apache.nifi.amqp.processors.PublishAMQP.processResource(PublishAMQP.java:185)
	at org.apache.nifi.amqp.processors.PublishAMQP.processResource(PublishAMQP.java:52)
	at org.apache.nifi.amqp.processors.AbstractAMQPProcessor.onTrigger(AbstractAMQPProcessor.java:236)
	at org.apache.nifi.processor.AbstractProcessor.onTrigger(AbstractProcessor.java:27)
	at org.apache.nifi.controller.StandardProcessorNode.onTrigger(StandardProcessorNode.java:1292)
	at org.apache.nifi.controller.tasks.ConnectableTask.invoke(ConnectableTask.java:229)
	at org.apache.nifi.controller.scheduling.TimerDrivenSchedulingAgent$1.run(TimerDrivenSchedulingAgent.java:102)
	at org.apache.nifi.engine.FlowEngine.lambda$wrap$1(FlowEngine.java:105)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:572)
	at java.base/java.util.concurrent.FutureTask.runAndReset(FutureTask.java:358)
	at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
	at java.base/java.lang.Thread.run(Thread.java:1583)
Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'dummy' in vhost '/', class-id=60, method-id=40)
	at com.rabbitmq.client.impl.ChannelN.asyncShutdown(ChannelN.java:529)
	at com.rabbitmq.client.impl.ChannelN.processAsync(ChannelN.java:350)
	at com.rabbitmq.client.impl.AMQChannel.handleCompleteInboundCommand(AMQChannel.java:193)
	at com.rabbitmq.client.impl.AMQChannel.handleFrame(AMQChannel.java:125)
	at com.rabbitmq.client.impl.AMQConnection.readFrame(AMQConnection.java:768)
	at com.rabbitmq.client.impl.AMQConnection.access$400(AMQConnection.java:49)
	at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:695)
	... 1 common frames omitted

[rakesh-rsky]

@turcsanyip Thank you for validating this.

Root cause: When the exchange does not exist, the broker closes the channel with 404 NOT_FOUND, causing waitForConfirms() to throw ShutdownSignalException — which was not caught, resulting in an unhandled processor failure instead of routing to REL_FAILURE.

This has been fixed in the latest commits — ShutdownSignalException is now caught and converted to AMQPException, along with regression tests covering this and related broker failure scenarios.

[turcsanyip]

@rakesh-rsky I checked the latest changes and the error handling is correct now functionally.

However, the synchronous wait led to significant performance degradation compared to the original version. According to my measurement:

• local NiFi + local RabbitMQ in Docker: 10x slower
• local NiFi + RabbitMQ on a cloud VM: 300-1000x slower (depending on the cloud location)

So even if the error handling is improved, I don't think we can add the change as-is due to the performance effect. I suggest adding a feature flag property (e.g. Use Message Confirmation = true / false or Delivery Guarantee = At least once / At most once) with proper documentation on the pros and cons. This way, the user can optionally enable this feature if they want the extra failure handling, at the cost of performance.

[rakesh-rsky]

@turcsanyip Thank you for the performance feedback and the suggestion.

I've added a Delivery Guarantee property to PublishAMQP with two options:

• At most once (default) - original fire-and-forget behaviour, no performance impact. Undeliverable messages are logged as a warning and the FlowFile still routes to success.
• At least once - enables Publisher Confirms. The processor waits for a broker ack/nack before routing. Undeliverable messages and NACKs route to failure, preventing silent data loss.

This keeps the default behaviour unchanged while giving users the option to trade throughput for delivery reliability.