Merge branch 'pr2931'

adyoun2 · Oct 2, 2018 · b4c8e01 · b4c8e01
2 parents 895323f + 76a9f98
commit b4c8e01
Showing 1 changed file with 6 additions and 4 deletions.
diff --git a/...-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/JMSConsumer.java b/...-bundle/nifi-jms-processors/src/main/java/org/apache/nifi/jms/processors/JMSConsumer.java
@@ -85,10 +85,6 @@ public void consume(final String destinationName, final boolean durable, final b
         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 @@ public Void doInJms(final Session session) throws JMSException {
                     // and if CLIENT_ACKNOWLEDGE is set.
                     consumerCallback.accept(response);
                     acknowledge(message, session);
+                } catch (Exception 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);
                 }