[
https://issues.apache.org/jira/browse/NIFI-16241?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claudio Mattioni reassigned NIFI-16241:
---------------------------------------
Assignee: Claudio Mattioni
> 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
> Assignee: Claudio Mattioni
> Priority: Major
> Labels: AMQP
> Time Spent: 10m
> Remaining Estimate: 0h
>
> {{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)
