Attempt to fix AMQP link race with the lock

* Wrap main loop in the `AmqpMessageListenerContainer.AmqpConsumer`
into a lock
* Use same lock around `this.receiver.close();` to ensure that no other
link flow interactions are in between

Author: artembilan

spring-amqp-client/src/main/java/org/springframework/amqp/client/listener/AmqpMessageListenerContainer.java

@@ -468,6 +468,8 @@ private class AmqpConsumer implements SchedulingAwareRunnable, AutoCloseable {
 
 		private final ClientReceiver receiver;
 
+		private final Lock receiverLock = new ReentrantLock();
+
 		private final ProtonReceiver protonReceiver;
 
 		private final ProtonLinkCreditState creditState;
@@ -494,31 +496,37 @@ int queuedDeliveries() {
 
 		@Override
 		public void run() {
-			while (this.running) {
-				try {
-					Delivery delivery =
-							this.receiver.receive(AmqpMessageListenerContainer.this.receiveTimeout.toMillis(),
-									TimeUnit.MILLISECONDS);
-					if (delivery != null) {
-						doInvokeListener(delivery, this::replenishCredit);
+			this.receiverLock.lock();
+			try {
+				while (this.running) {
+					try {
+						Delivery delivery =
+								this.receiver.receive(AmqpMessageListenerContainer.this.receiveTimeout.toMillis(),
+										TimeUnit.MILLISECONDS);
+						if (delivery != null) {
+							doInvokeListener(delivery, this::replenishCredit);
+						}
 					}
-				}
-				catch (Exception ex) {
-					if (this.running) {
-						AmqpException amqpException = ProtonUtils.toAmqpException(ex);
-						ErrorHandler errorHandlerToUse = AmqpMessageListenerContainer.this.errorHandler;
-						if (errorHandlerToUse != null) {
-							errorHandlerToUse.handleError(amqpException);
+					catch (Exception ex) {
+						if (this.running) {
+							AmqpException amqpException = ProtonUtils.toAmqpException(ex);
+							ErrorHandler errorHandlerToUse = AmqpMessageListenerContainer.this.errorHandler;
+							if (errorHandlerToUse != null) {
+								errorHandlerToUse.handleError(amqpException);
+							}
+							else {
+								throw amqpException;
+							}
 						}
 						else {
-							throw amqpException;
+							LOG.debug(ex, "Consumer stopped");
 						}
 					}
-					else {
-						LOG.debug(ex, "Consumer stopped");
-					}
 				}
 			}
+			finally {
+				this.receiverLock.unlock();
+			}
 		}
 
 		/**
@@ -534,7 +542,9 @@ private void replenishCredit() {
 						if (potentialPrefetch <= AmqpMessageListenerContainer.this.initialCredits * 0.7) {
 							int additionalCredit = AmqpMessageListenerContainer.this.initialCredits - potentialPrefetch;
 
-							this.receiver.addCredit(additionalCredit);
+							if (!this.paused && this.running) {
+								this.receiver.addCredit(additionalCredit);
+							}
 						}
 					}
 				}
@@ -549,15 +559,15 @@ private void replenishCredit() {
 		 * so rely on the reflection to imitate behavior with resetting credits to zero.
 		 */
 		void pause() {
-			if (this.running) {
+			if (this.running && !this.paused) {
 				this.paused = true;
 				this.creditState.updateCredit(0);
 				ReflectionUtils.invokeMethod(WRITE_FLOW_METHOD, this.sessionWindow, this.protonReceiver);
 			}
 		}
 
 		void resume() {
-			if (this.running) {
+			if (this.running && this.paused) {
 				this.paused = false;
 				try {
 					this.receiver.addCredit(AmqpMessageListenerContainer.this.initialCredits);
@@ -576,7 +586,13 @@ public boolean isLongLived() {
 		@Override
 		public void close() {
 			this.running = false;
-			this.receiver.close();
+			this.receiverLock.lock();
+			try {
+				this.receiver.close();
+			}
+			finally {
+				this.receiverLock.unlock();
+			}
 		}
 
 	}