Claudio Mattioni created NIFI-16241:
---------------------------------------
Summary: PublishAMQP should route FlowFiles with null or blank
evaluated Routing Key to failure
Key: NIFI-16241
URL: https://issues.apache.org/jira/browse/NIFI-16241
Project: Apache NiFi
Issue Type: Bug
Components: Extensions
Affects Versions: 2.11.0
Reporter: Claudio Mattioni
{{PublishAMQP}} supports Expression Language for the *Routing Key* property.
The configured property is validated as non-empty, but the value obtained after
evaluating Expression Language against a FlowFile can be null or blank.
At runtime, {{PublishAMQP.processResource()}} evaluates the Routing Key and
only checks whether the resulting value is {{{}null{}}}.
A blank value therefore reaches {{{}AMQPPublisher.publish(){}}}, where
{{validateStringProperty()}} rejects null or blank Routing Keys with an
{{{}IllegalArgumentException{}}}.
This exception is not handled as a FlowFile-specific failure. It propagates to
the generic exception handler in {{{}AbstractAMQPProcessor{}}}, which rolls
back the ProcessSession, yields the processor, and closes the AMQP client.
As a result, the affected FlowFile remains in the incoming queue and the same
deterministic error is retried indefinitely.
The same rollback behavior can also occur when evaluation produces
{{{}null{}}}, because the current null check throws an
{{IllegalArgumentException}} instead of routing the FlowFile to {{{}failure{}}}.
h2. Steps to Reproduce
# Configure a {{PublishAMQP}} processor with a valid RabbitMQ connection and
destination.
# Configure *Routing Key* using Expression Language, for example
{{{}${missing}{}}}, where the referenced FlowFile attribute is not present, or
use an expression/attribute that evaluates to a blank value.
# Send a FlowFile without the referenced attribute through the processor.
# Observe that the FlowFile is not transferred to either {{success}} or
{{{}failure{}}}.
# Observe an {{IllegalArgumentException}} related to the Routing Key and the
FlowFile remaining in the incoming queue.
# Allow the processor to run again and observe the same FlowFile being retried
indefinitely.
A test probe reproduced the behavior with:
{{success = 0}}
{{failure = 0}}
{{queued FlowFiles = 1}}
and an {{IllegalArgumentException}} caused by the invalid Routing Key.
h2. Expected Behavior
When the evaluated Routing Key is null or blank, the FlowFile should be routed
to the {{failure}} relationship instead of rolling back the ProcessSession.
This is a deterministic FlowFile-specific configuration/data problem, so
recreating the AMQP client and retrying the same FlowFile cannot resolve it.
h2. Actual Behavior
The invalid evaluated Routing Key results in an
{{{}IllegalArgumentException{}}}.
The exception reaches the generic error handler in
{{{}AbstractAMQPProcessor{}}}, which rolls back the session, yields the
processor, and closes the AMQP client.
The FlowFile therefore remains queued and is retried indefinitely.
h2. Suggested Fix
Validate the evaluated Routing Key in {{PublishAMQP.processResource()}} before
invoking {{AMQPPublisher.publish()}} and route invalid values directly to
{{{}failure{}}}.
For example:
{{if (routingKey == null || routingKey.isBlank()) \{
getLogger().error("Failed to determine a non-empty Routing Key after
evaluating the configured value against FlowFile {}", flowFile);
session.transfer(flowFile, REL_FAILURE);
return;
}}}
This also avoids unnecessarily dropping and recreating the AMQP client for an
error that is unrelated to the connection.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
