GH-3079: Allow ACKs within the ShutdownTimeout period

Fixes: #3079
Issue link: #3079

Remove `cancelled()` condition from the `BlockingQueueConsumer.commitIfNecessary()` to allow initiated acks to be committed.
The cancelled consumer means no new deliveries, but that does not mean we cannot ack outstanding messages.
Actually, this is a leftover after previous fix with assumption that messages have to be returned to the queue with `Recovery` command when consumer is cancelled.

Signed-off-by: Marco Cozzi <[email protected]>

**Auto-cherry-pick to `3.2.x` & `3.1.x`**

Author: marcogramy

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java

@@ -888,7 +888,7 @@ public void rollbackOnExceptionIfNecessary(Throwable ex, long tag) {
 
 	/**
 	 * Perform a commit or message acknowledgement, as appropriate.
-	 * NOTE: This method was never been intended tobe public.
+	 * NOTE: This method was never intended to be public.
 	 * @param localTx Whether the channel is locally transacted.
 	 * @param forceAck perform {@link Channel#basicAck(long, boolean)} independently of {@link #acknowledgeMode}.
 	 * @return true if at least one delivery tag exists.
@@ -909,7 +909,7 @@ boolean commitIfNecessary(boolean localTx, boolean forceAck) {
 		try {
 			boolean ackRequired = forceAck || (!this.acknowledgeMode.isAutoAck() && !this.acknowledgeMode.isManual());
 
-			if (ackRequired && (!this.transactional || (isLocallyTransacted && !cancelled()))) {
+			if (ackRequired && (!this.transactional || isLocallyTransacted)) {
 				OptionalLong deliveryTag = this.deliveryTags.stream().mapToLong(l -> l).max();
 				deliveryTag.ifPresent((tag) -> {
 					try {