Introduce acknowledgeAfterListener flag for custom acknowledge handling

Closes gh-34635

spring-jms/src/main/java/org/springframework/jms/config/AbstractJmsListenerContainerFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2023 the original author or authors.
+ * Copyright 2002-2025 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -63,6 +63,9 @@ public abstract class AbstractJmsListenerContainerFactory<C extends AbstractMess
 	@Nullable
 	private Integer sessionAcknowledgeMode;
 
+	@Nullable
+	private Boolean acknowledgeAfterListener;
+
 	@Nullable
 	private Boolean pubSubDomain;
 
@@ -141,6 +144,14 @@ public abstract class AbstractJmsListenerContainerFactory<C extends AbstractMess
 		this.sessionAcknowledgeMode = sessionAcknowledgeMode;
 	}
 
+	/**
+	 * @since 6.2.6
+	 * @see AbstractMessageListenerContainer#setAcknowledgeAfterListener(boolean)
+	 */
+	public void setAcknowledgeAfterListener(Boolean acknowledgeAfterListener) {
+		this.acknowledgeAfterListener = acknowledgeAfterListener;
+	}
+
 	/**
 	 * @see AbstractMessageListenerContainer#setPubSubDomain(boolean)
 	 */
@@ -209,6 +220,7 @@ public abstract class AbstractJmsListenerContainerFactory<C extends AbstractMess
 		this.observationRegistry = observationRegistry;
 	}
 
+
 	@Override
 	public C createListenerContainer(JmsListenerEndpoint endpoint) {
 		C instance = createContainerInstance();
@@ -234,6 +246,9 @@ public abstract class AbstractJmsListenerContainerFactory<C extends AbstractMess
 		if (this.sessionAcknowledgeMode != null) {
 			instance.setSessionAcknowledgeMode(this.sessionAcknowledgeMode);
 		}
+		if (this.acknowledgeAfterListener != null) {
+			instance.setAcknowledgeAfterListener(this.acknowledgeAfterListener);
+		}
 		if (this.pubSubDomain != null) {
 			instance.setPubSubDomain(this.pubSubDomain);
 		}

spring-jms/src/main/java/org/springframework/jms/listener/AbstractMessageListenerContainer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2024 the original author or authors.
+ * Copyright 2002-2025 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -152,7 +152,8 @@ public abstract class AbstractMessageListenerContainer extends AbstractJmsListen
 		implements MessageListenerContainer {
 
 	private static final boolean micrometerJakartaPresent = ClassUtils.isPresent(
-			"io.micrometer.jakarta9.instrument.jms.JmsInstrumentation", AbstractMessageListenerContainer.class.getClassLoader());
+			"io.micrometer.jakarta9.instrument.jms.JmsInstrumentation",
+			AbstractMessageListenerContainer.class.getClassLoader());
 
 	@Nullable
 	private volatile Object destination;
@@ -170,14 +171,14 @@ public abstract class AbstractMessageListenerContainer extends AbstractJmsListen
 	@Nullable
 	private String subscriptionName;
 
+	private boolean pubSubNoLocal = false;
+
 	@Nullable
 	private Boolean replyPubSubDomain;
 
 	@Nullable
 	private QosSettings replyQosSettings;
 
-	private boolean pubSubNoLocal = false;
-
 	@Nullable
 	private MessageConverter messageConverter;
 
@@ -190,6 +191,8 @@ public abstract class AbstractMessageListenerContainer extends AbstractJmsListen
 	@Nullable
 	private ObservationRegistry observationRegistry;
 
+	private boolean acknowledgeAfterListener = true;
+
 	private boolean exposeListenerSession = true;
 
 	private boolean acceptMessagesWhileStopping = false;
@@ -500,12 +503,7 @@ public abstract class AbstractMessageListenerContainer extends AbstractJmsListen
 	 */
 	@Override
 	public boolean isReplyPubSubDomain() {
-		if (this.replyPubSubDomain != null) {
+		return (this.replyPubSubDomain != null ? this.replyPubSubDomain : isPubSubDomain());
-			return this.replyPubSubDomain;
-		}
-		else {
-			return isPubSubDomain();
-		}
 	}
 
 	/**
@@ -596,6 +594,37 @@ public abstract class AbstractMessageListenerContainer extends AbstractJmsListen
 		return this.observationRegistry;
 	}
 
+	/**
+	 * Specify whether the listener container should automatically acknowledge
+	 * each JMS Message after the message listener returned. This applies in
+	 * case of client acknowledge modes, including vendor-specific modes but
+	 * not in case of auto-acknowledge or a transacted JMS Session.
+	 * <p>As of 6.2, the default is {@code true}: The listener container will
+	 * acknowledge each JMS Message even in case of a vendor-specific mode,
+	 * assuming client-acknowledge style processing for custom vendor modes.
+	 * <p>If the provided listener prefers to manually acknowledge each message in
+	 * the listener itself, in combination with an "individual acknowledge" mode,
+	 * switch this flag to {code false} along with the vendor-specific mode.
+	 * @since 6.2.6
+	 * @see #setSessionAcknowledgeMode
+	 * @see #setMessageListener
+	 * @see Message#acknowledge()
+	 */
+	public void setAcknowledgeAfterListener(boolean acknowledgeAfterListener) {
+		this.acknowledgeAfterListener = acknowledgeAfterListener;
+	}
+
+	/**
+	 * Determine whether the listener container should automatically acknowledge
+	 * each JMS Message after the message listener returned.
+	 * @since 6.2.6
+	 * @see #setAcknowledgeAfterListener
+	 * @see #isClientAcknowledge(Session)
+	 */
+	public boolean isAcknowledgeAfterListener() {
+		return this.acknowledgeAfterListener;
+	}
+
 	/**
 	 * Set whether to expose the listener JMS Session to a registered
 	 * {@link SessionAwareMessageListener} as well as to
@@ -833,7 +862,7 @@ public abstract class AbstractMessageListenerContainer extends AbstractJmsListen
 				JmsUtils.commitIfNecessary(session);
 			}
 		}
-		else if (message != null && isClientAcknowledge(session)) {
+		else if (message != null && isAcknowledgeAfterListener() && isClientAcknowledge(session)) {
 			message.acknowledge();
 		}
 	}