turcsanyip commented on code in PR #11243:
URL: https://github.com/apache/nifi/pull/11243#discussion_r3268942544
##########
nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/AbstractAMQPProcessor.java:
##########
@@ -244,8 +244,10 @@ public final void onTrigger(final ProcessContext context,
final ProcessSession s
context.yield();
closeResource(resource);
} catch (Exception e) {
- getLogger().error("Processor failure", e);
+ getLogger().error("Processor failure, dropping the client", e);
+ session.rollback();
context.yield();
+ closeResource(resource);
}
Review Comment:
@ing-mattioni Yes, correct. Given that `basic.nack` call does not really
contribute to the fix, I would not merge that PR if you and @exceptionfactory
agree.
The real fix is to close the client so that all pending (unacknowledged)
messages become available for the broker to redeliver to the new client or
other existing clients. The negative acknowledgment happens too early, and the
messages may be redelivered to the same client that is about to be closed.
Closing the client is simply enough and handles both the failed messages (that
could be nacked optionally) and the ones in the client's local buffer
(unprocessed yet but already fetched from the broker).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
