[jira] [Assigned] (NIFI-16241) PublishAMQP should route FlowFiles with null or blank evaluated Routing Key to failure

["Claudio Mattioni (Jira)" <ji...@apache.org>]

     [ 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)