Repository: nifi Updated Branches: refs/heads/master 895323f3c -> b4c8e0179
NIFI-3531 Moved session.recover in JMSConsumer to exceptional situations Project: http://git-wip-us.apache.org/repos/asf/nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/83097478 Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/83097478 Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/83097478 Branch: refs/heads/master Commit: 8309747889c3187005a0ce4e4c0c11b1e3281e10 Parents: b4894c5 Author: Mike Moser <[email protected]> Authored: Wed Aug 1 20:11:35 2018 +0000 Committer: Mike Moser <[email protected]> Committed: Wed Aug 1 20:11:40 2018 +0000 ---------------------------------------------------------------------- .../java/org/apache/nifi/jms/processors/JMSConsumer.java | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi/blob/83097478/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/JMSConsumer.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/JMSConsumer.java b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/JMSConsumer.java index 8092277..cf22175 100644 --- a/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/JMSConsumer.java +++ b/nifi-nar-bundles/nifi-jms-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/JMSConsumer.java @@ -85,10 +85,6 @@ final class JMSConsumer extends JMSWorker { this.jmsTemplate.execute(new SessionCallback<Void>() { @Override public Void doInJms(final Session session) throws JMSException { - // We need to call recover to ensure that in in the event of - // abrupt end or exception the current session will stop message - // delivery and restarts with the oldest unacknowledged message - session.recover(); final MessageConsumer msgConsumer = createMessageConsumer(session, destinationName, durable, shared, subscriberName); try { @@ -126,6 +122,12 @@ final class JMSConsumer extends JMSWorker { // and if CLIENT_ACKNOWLEDGE is set. consumerCallback.accept(response); acknowledge(message, session); + } catch (JMSException e) { + // We need to call recover to ensure that in the event of + // abrupt end or exception the current session will stop message + // delivery and restart with the oldest unacknowledged message + session.recover(); + throw e; } finally { JmsUtils.closeMessageConsumer(msgConsumer); }
