GH-2994: Add RabbitAmqpListenerContainer infrastructure

Related to: https://github.com/spring-projects/spring-amqp/issues/2994

* Add `RabbitAmqpMessageListener` for RabbitMQ AMQP 1.0 native message consumption
* Add `RabbitAmqpMessageListenerAdapter` for `@RabbitLister` API
* Add `RabbitAmqpListenerContainer` and respective `RabbitAmqpListenerContainerFactory`
* Add `AmqpAcknowledgment` as a general abstraction.
In the `RabbitAmqpListenerContainer` delegates to the `Consumer.Context` for manual settlement
* Extract `RabbitAmqpUtils` for conversion to/from AMQP 1.0 native message
* Add convenient `ContainerUtils.isImmediateAcknowledge()` and `ContainerUtils.isAmqpReject()` utilities
* Expose `AmqpAcknowledgment` as a `MessageProperties.amqpAcknowledgment` for generic `MessageListener` use-cases
* Remove `io.micrometer` dependecies from the `spring-rabbitmq-client` module since metrics and observation handled
thoroughly in the `com.rabbitmq.client:amqp-client`

Not tests for the listener yet.
Therefore no fixing for the issue.

build.gradle

@@ -479,7 +479,6 @@ project('spring-rabbitmq-client') {
 	dependencies {
 		api project(':spring-rabbit')
 		api "com.rabbitmq.client:amqp-client:$rabbitmqAmqpClientVersion"
-		api 'io.micrometer:micrometer-observation'
 
 		testApi project(':spring-rabbit-junit')
 
@@ -488,10 +487,6 @@ project('spring-rabbitmq-client') {
 		testImplementation 'org.testcontainers:rabbitmq'
 		testImplementation 'org.testcontainers:junit-jupiter'
 		testImplementation 'org.apache.logging.log4j:log4j-slf4j-impl'
-		testImplementation 'io.micrometer:micrometer-observation-test'
-		testImplementation 'io.micrometer:micrometer-tracing-bridge-brave'
-		testImplementation 'io.micrometer:micrometer-tracing-test'
-		testImplementation 'io.micrometer:micrometer-tracing-integration-test'
 	}
 }
 

spring-amqp/src/main/java/org/springframework/amqp/core/AmqpAcknowledgment.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.amqp.core;
+
+/**
+ * An abstraction over acknowledgments.
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.0
+ */
+@FunctionalInterface
+public interface AmqpAcknowledgment {
+
+	/**
+	 * Acknowledge the message.
+	 * @param status the status.
+	 */
+	void acknowledge(Status status);
+
+	default void acknowledge() {
+		acknowledge(Status.ACCEPT);
+	}
+
+	enum Status {
+
+		/**
+		 * Mark the message as accepted.
+		 */
+		ACCEPT,
+
+		/**
+		 * Mark the message as rejected.
+		 */
+		REJECT,
+
+		/**
+		 * Reject the message and requeue so that it will be redelivered.
+		 */
+		REQUEUE
+
+	}
+
+}

spring-amqp/src/main/java/org/springframework/amqp/core/MessageProperties.java

@@ -157,6 +157,8 @@ public class MessageProperties implements Serializable {
 
 	private transient @Nullable Object targetBean;
 
+	private transient @Nullable AmqpAcknowledgment amqpAcknowledgment;
+
 	public void setHeader(String key, Object value) {
 		this.headers.put(key, value);
 	}
@@ -641,6 +643,25 @@ public class MessageProperties implements Serializable {
 		}
 	}
 
+	/**
+	 * Return the {@link AmqpAcknowledgment} for consumer if any.
+	 * @return the {@link AmqpAcknowledgment} for consumer if any.
+	 * @since 4.0
+	 */
+	public @Nullable AmqpAcknowledgment getAmqpAcknowledgment() {
+		return this.amqpAcknowledgment;
+	}
+
+	/**
+	 * Set an {@link AmqpAcknowledgment} for manual acks in the target message processor.
+	 * This is only in-application a consumer side logic.
+	 * @param amqpAcknowledgment the {@link AmqpAcknowledgment} to use in the application.
+	 * @since 4.0
+	 */
+	public void setAmqpAcknowledgment(AmqpAcknowledgment amqpAcknowledgment) {
+		this.amqpAcknowledgment = amqpAcknowledgment;
+	}
+
 	@Override // NOSONAR complexity
 	public int hashCode() {
 		final int prime = 31;

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

@@ -752,7 +752,7 @@ public abstract class AbstractMessageListenerContainer extends ObservableListene
 	 * to be sent to the dead letter exchange. Setting to false causes all rejections to not
 	 * be requeued. When true, the default can be overridden by the listener throwing an
 	 * {@link AmqpRejectAndDontRequeueException}. Default true.
-	 * @param defaultRequeueRejected true to reject by default.
+	 * @param defaultRequeueRejected true to requeue by default.
 	 */
 	public void setDefaultRequeueRejected(boolean defaultRequeueRejected) {
 		this.defaultRequeueRejected = defaultRequeueRejected;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2022 the original author or authors.
+ * Copyright 2018-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.
@@ -17,8 +17,10 @@
 package org.springframework.amqp.rabbit.listener.support;
 
 import org.apache.commons.logging.Log;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.amqp.AmqpRejectAndDontRequeueException;
+import org.springframework.amqp.ImmediateAcknowledgeAmqpException;
 import org.springframework.amqp.ImmediateRequeueAmqpException;
 import org.springframework.amqp.rabbit.listener.exception.MessageRejectedWhileStoppingException;
 
@@ -59,7 +61,11 @@ public final class ContainerUtils {
 				shouldRequeue = true;
 				break;
 			}
-			t = t.getCause();
+			Throwable cause = t.getCause();
+			if (cause == t) {
+				break;
+			}
+			t = cause;
 		}
 		if (logger.isDebugEnabled()) {
 			logger.debug("Rejecting messages (requeue=" + shouldRequeue + ")");
@@ -75,8 +81,41 @@ public final class ContainerUtils {
 	 * @since 2.2
 	 */
 	public static boolean isRejectManual(Throwable ex) {
-		return ex instanceof AmqpRejectAndDontRequeueException aradrex
-				&& aradrex.isRejectManual();
+		AmqpRejectAndDontRequeueException amqpRejectAndDontRequeueException =
+				findInCause(ex, AmqpRejectAndDontRequeueException.class);
+		return amqpRejectAndDontRequeueException != null && amqpRejectAndDontRequeueException.isRejectManual();
+	}
+
+	/**
+	 * Return true for {@link ImmediateAcknowledgeAmqpException}.
+	 * @param ex the exception to traverse.
+	 * @return true if an {@link ImmediateAcknowledgeAmqpException} is present in the cause chain.
+	 * @since 4.0
+	 */
+	public static boolean isImmediateAcknowledge(Throwable ex) {
+		return findInCause(ex, ImmediateAcknowledgeAmqpException.class) != null;
+	}
+
+	/**
+	 * Return true for {@link AmqpRejectAndDontRequeueException}.
+	 * @param ex the exception to traverse.
+	 * @return true if an {@link AmqpRejectAndDontRequeueException} is present in the cause chain.
+	 * @since 4.0
+	 */
+	public static boolean isAmqpReject(Throwable ex) {
+		return findInCause(ex, AmqpRejectAndDontRequeueException.class) != null;
+	}
+
+	@SuppressWarnings("unchecked")
+	private static <T extends Throwable> @Nullable T findInCause(Throwable throwable, Class<T> exceptionToFind) {
+		if (exceptionToFind.isAssignableFrom(throwable.getClass())) {
+			return (T) throwable;
+		}
+		Throwable cause = throwable.getCause();
+		if (cause == null || cause == throwable) {
+			return null;
+		}
+		return findInCause(cause, exceptionToFind);
 	}
 
 }

spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplate.java

@@ -16,11 +16,7 @@
 
 package org.springframework.amqp.rabbitmq.client;
 
-import java.nio.charset.StandardCharsets;
 import java.time.Duration;
-import java.util.Date;
-import java.util.Map;
-import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
@@ -182,38 +178,20 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, InitializingBean,
 	private CompletableFuture<Boolean> doSend(@Nullable String exchange, @Nullable String routingKey,
 			@Nullable String queue, Message message) {
 
-		MessageProperties messageProperties = message.getMessageProperties();
-
-		com.rabbitmq.client.amqp.Message amqpMessage =
-				this.publisher.message(message.getBody())
-						.contentEncoding(messageProperties.getContentEncoding())
-						.contentType(messageProperties.getContentType())
-						.messageId(messageProperties.getMessageId())
-						.correlationId(messageProperties.getCorrelationId())
-						.priority(messageProperties.getPriority().byteValue())
-						.replyTo(messageProperties.getReplyTo());
-
+		com.rabbitmq.client.amqp.Message amqpMessage = this.publisher.message();
 		com.rabbitmq.client.amqp.Message.MessageAddressBuilder address = amqpMessage.toAddress();
-
-		Map<String, @Nullable Object> headers = messageProperties.getHeaders();
-		if (!headers.isEmpty()) {
-			headers.forEach((key, val) -> mapProp(key, val, amqpMessage));
-		}
-
 		JavaUtils.INSTANCE
-				.acceptIfNotNull(messageProperties.getUserId(),
-						(userId) -> amqpMessage.userId(userId.getBytes(StandardCharsets.UTF_8)))
-				.acceptIfNotNull(messageProperties.getTimestamp(),
-						(timestamp) -> amqpMessage.creationTime(timestamp.getTime()))
-				.acceptIfNotNull(messageProperties.getExpiration(),
-						(expiration) -> amqpMessage.absoluteExpiryTime(Long.parseLong(expiration)))
 				.acceptIfNotNull(exchange, address::exchange)
 				.acceptIfNotNull(routingKey, address::key)
 				.acceptIfNotNull(queue, address::queue);
 
+		amqpMessage = address.message();
+
+		RabbitAmqpUtils.toAmqpMessage(message, amqpMessage);
+
 		CompletableFuture<Boolean> publishResult = new CompletableFuture<>();
 
-		this.publisher.publish(address.message(),
+		this.publisher.publish(amqpMessage,
 				(context) -> {
 					switch (context.status()) {
 						case ACCEPTED -> publishResult.complete(true);
@@ -299,7 +277,7 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, InitializingBean,
 						.priority(10)
 						.messageHandler((context, message) -> {
 							context.accept();
-							messageFuture.complete(fromAmqpMessage(message));
+							messageFuture.complete(RabbitAmqpUtils.fromAmqpMessage(message, null));
 						})
 						.build();
 
@@ -452,62 +430,4 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, InitializingBean,
 		throw new UnsupportedOperationException();
 	}
 
-	private static void mapProp(String key, @Nullable Object val, com.rabbitmq.client.amqp.Message amqpMessage) {
-		if (val == null) {
-			return;
-		}
-		if (val instanceof String string) {
-			amqpMessage.property(key, string);
-		}
-		else if (val instanceof Long longValue) {
-			amqpMessage.property(key, longValue);
-		}
-		else if (val instanceof Integer intValue) {
-			amqpMessage.property(key, intValue);
-		}
-		else if (val instanceof Short shortValue) {
-			amqpMessage.property(key, shortValue);
-		}
-		else if (val instanceof Byte byteValue) {
-			amqpMessage.property(key, byteValue);
-		}
-		else if (val instanceof Double doubleValue) {
-			amqpMessage.property(key, doubleValue);
-		}
-		else if (val instanceof Float floatValue) {
-			amqpMessage.property(key, floatValue);
-		}
-		else if (val instanceof Character character) {
-			amqpMessage.property(key, character);
-		}
-		else if (val instanceof UUID uuid) {
-			amqpMessage.property(key, uuid);
-		}
-		else if (val instanceof byte[] bytes) {
-			amqpMessage.property(key, bytes);
-		}
-		else if (val instanceof Boolean booleanValue) {
-			amqpMessage.property(key, booleanValue);
-		}
-	}
-
-	private static Message fromAmqpMessage(com.rabbitmq.client.amqp.Message amqpMessage) {
-		MessageProperties messageProperties = new MessageProperties();
-
-		JavaUtils.INSTANCE
-				.acceptIfNotNull(amqpMessage.messageIdAsString(), messageProperties::setMessageId)
-				.acceptIfNotNull(amqpMessage.userId(),
-						(usr) -> messageProperties.setUserId(new String(usr, StandardCharsets.UTF_8)))
-				.acceptIfNotNull(amqpMessage.correlationIdAsString(), messageProperties::setCorrelationId)
-				.acceptIfNotNull(amqpMessage.contentType(), messageProperties::setContentType)
-				.acceptIfNotNull(amqpMessage.contentEncoding(), messageProperties::setContentEncoding)
-				.acceptIfNotNull(amqpMessage.absoluteExpiryTime(),
-						(exp) -> messageProperties.setExpiration(Long.toString(exp)))
-				.acceptIfNotNull(amqpMessage.creationTime(), (time) -> messageProperties.setTimestamp(new Date(time)));
-
-		amqpMessage.forEachProperty(messageProperties::setHeader);
-
-		return new Message(amqpMessage.body(), messageProperties);
-	}
-
 }

spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpUtils.java

@@ -0,0 +1,145 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.amqp.rabbitmq.client;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Date;
+import java.util.Map;
+import java.util.UUID;
+
+import com.rabbitmq.client.amqp.Consumer;
+import org.jspecify.annotations.Nullable;
+
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.utils.JavaUtils;
+
+/**
+ * The utilities for RabbitMQ AMQP 1.0 protocol API.
+ */
+public final class RabbitAmqpUtils {
+
+	/**
+	 * Convert {@link com.rabbitmq.client.amqp.Message} into {@link Message}.
+	 * @param amqpMessage the {@link com.rabbitmq.client.amqp.Message} convert from.
+	 * @param context the {@link Consumer.Context} for manual message settlement.
+	 * @return the {@link Message} mapped from a {@link com.rabbitmq.client.amqp.Message}.
+	 */
+	public static Message fromAmqpMessage(com.rabbitmq.client.amqp.Message amqpMessage,
+			Consumer.@Nullable Context context) {
+
+		MessageProperties messageProperties = new MessageProperties();
+
+		JavaUtils.INSTANCE
+				.acceptIfNotNull(amqpMessage.messageIdAsString(), messageProperties::setMessageId)
+				.acceptIfNotNull(amqpMessage.userId(),
+						(usr) -> messageProperties.setUserId(new String(usr, StandardCharsets.UTF_8)))
+				.acceptIfNotNull(amqpMessage.correlationIdAsString(), messageProperties::setCorrelationId)
+				.acceptIfNotNull(amqpMessage.contentType(), messageProperties::setContentType)
+				.acceptIfNotNull(amqpMessage.contentEncoding(), messageProperties::setContentEncoding)
+				.acceptIfNotNull(amqpMessage.absoluteExpiryTime(),
+						(exp) -> messageProperties.setExpiration(Long.toString(exp)))
+				.acceptIfNotNull(amqpMessage.creationTime(), (time) -> messageProperties.setTimestamp(new Date(time)));
+
+		amqpMessage.forEachProperty(messageProperties::setHeader);
+
+		if (context != null) {
+			messageProperties.setAmqpAcknowledgment((status) -> {
+				switch (status) {
+					case ACCEPT -> context.accept();
+					case REJECT -> context.discard();
+					case REQUEUE -> context.requeue();
+				}
+			});
+		}
+
+		return new Message(amqpMessage.body(), messageProperties);
+	}
+
+	/**
+	 * Convert {@link com.rabbitmq.client.amqp.Message} into {@link Message}.
+	 * @param amqpMessage the {@link com.rabbitmq.client.amqp.Message} convert from.
+	 */
+	public static void toAmqpMessage(Message message, com.rabbitmq.client.amqp.Message amqpMessage) {
+		MessageProperties messageProperties = message.getMessageProperties();
+
+		amqpMessage
+				.body(message.getBody())
+				.contentEncoding(messageProperties.getContentEncoding())
+				.contentType(messageProperties.getContentType())
+				.messageId(messageProperties.getMessageId())
+				.correlationId(messageProperties.getCorrelationId())
+				.priority(messageProperties.getPriority().byteValue())
+				.replyTo(messageProperties.getReplyTo());
+
+		Map<String, @Nullable Object> headers = messageProperties.getHeaders();
+		if (!headers.isEmpty()) {
+			headers.forEach((key, val) -> mapProp(key, val, amqpMessage));
+		}
+
+		JavaUtils.INSTANCE
+				.acceptIfNotNull(messageProperties.getUserId(),
+						(userId) -> amqpMessage.userId(userId.getBytes(StandardCharsets.UTF_8)))
+				.acceptIfNotNull(messageProperties.getTimestamp(),
+						(timestamp) -> amqpMessage.creationTime(timestamp.getTime()))
+				.acceptIfNotNull(messageProperties.getExpiration(),
+						(expiration) -> amqpMessage.absoluteExpiryTime(Long.parseLong(expiration)));
+	}
+
+	private static void mapProp(String key, @Nullable Object val, com.rabbitmq.client.amqp.Message amqpMessage) {
+		if (val == null) {
+			return;
+		}
+		if (val instanceof String string) {
+			amqpMessage.property(key, string);
+		}
+		else if (val instanceof Long longValue) {
+			amqpMessage.property(key, longValue);
+		}
+		else if (val instanceof Integer intValue) {
+			amqpMessage.property(key, intValue);
+		}
+		else if (val instanceof Short shortValue) {
+			amqpMessage.property(key, shortValue);
+		}
+		else if (val instanceof Byte byteValue) {
+			amqpMessage.property(key, byteValue);
+		}
+		else if (val instanceof Double doubleValue) {
+			amqpMessage.property(key, doubleValue);
+		}
+		else if (val instanceof Float floatValue) {
+			amqpMessage.property(key, floatValue);
+		}
+		else if (val instanceof Character character) {
+			amqpMessage.property(key, character);
+		}
+		else if (val instanceof UUID uuid) {
+			amqpMessage.property(key, uuid);
+		}
+		else if (val instanceof byte[] bytes) {
+			amqpMessage.property(key, bytes);
+		}
+		else if (val instanceof Boolean booleanValue) {
+			amqpMessage.property(key, booleanValue);
+		}
+	}
+
+	private RabbitAmqpUtils() {
+	}
+
+}

spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/config/RabbitAmqpListenerContainerFactory.java

@@ -0,0 +1,97 @@
+/*
+ * Copyright 2021-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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.amqp.rabbitmq.client.config;
+
+import com.rabbitmq.client.amqp.Connection;
+import org.aopalliance.aop.Advice;
+import org.jspecify.annotations.Nullable;
+
+import org.springframework.amqp.rabbit.config.BaseRabbitListenerContainerFactory;
+import org.springframework.amqp.rabbit.config.ContainerCustomizer;
+import org.springframework.amqp.rabbit.listener.MethodRabbitListenerEndpoint;
+import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
+import org.springframework.amqp.rabbitmq.client.listener.RabbitAmqpListenerContainer;
+import org.springframework.amqp.rabbitmq.client.listener.RabbitAmqpMessageListenerAdapter;
+import org.springframework.amqp.utils.JavaUtils;
+
+/**
+ * Factory for {@link RabbitAmqpListenerContainer}.
+ * To use it as default one, has to be configured with a
+ * {@link org.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor#DEFAULT_RABBIT_LISTENER_CONTAINER_FACTORY_BEAN_NAME}.
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.0
+ *
+ */
+public class RabbitAmqpListenerContainerFactory
+		extends BaseRabbitListenerContainerFactory<RabbitAmqpListenerContainer> {
+
+	private final Connection connection;
+
+	private @Nullable ContainerCustomizer<RabbitAmqpListenerContainer> containerCustomizer;
+
+	/**
+	 * Construct an instance using the provided amqpConnection.
+	 * @param amqpConnection the connection.
+	 */
+	public RabbitAmqpListenerContainerFactory(Connection amqpConnection) {
+		this.connection = amqpConnection;
+	}
+
+	/**
+	 * Set a {@link ContainerCustomizer} that is invoked after a container is created and
+	 * configured to enable further customization of the container.
+	 * @param containerCustomizer the customizer.
+	 */
+	public void setContainerCustomizer(ContainerCustomizer<RabbitAmqpListenerContainer> containerCustomizer) {
+		this.containerCustomizer = containerCustomizer;
+	}
+
+	@Override
+	public RabbitAmqpListenerContainer createListenerContainer(@Nullable RabbitListenerEndpoint endpoint) {
+		if (endpoint instanceof MethodRabbitListenerEndpoint methodRabbitListenerEndpoint) {
+			methodRabbitListenerEndpoint.setAdapterProvider(
+					(batch, bean, method, returnExceptions, errorHandler, batchingStrategy) ->
+							new RabbitAmqpMessageListenerAdapter(bean, method, returnExceptions, errorHandler));
+		}
+		RabbitAmqpListenerContainer container = createContainerInstance();
+		Advice[] adviceChain = getAdviceChain();
+		JavaUtils.INSTANCE
+				.acceptIfNotNull(adviceChain, container::setAdviceChain)
+				.acceptIfNotNull(getDefaultRequeueRejected(), container::setDefaultRequeue);
+
+		applyCommonOverrides(endpoint, container);
+
+		if (endpoint != null) {
+			JavaUtils.INSTANCE
+					.acceptIfNotNull(endpoint.getAckMode(),
+							(ackMode) -> container.setAutoSettle(!ackMode.isManual()))
+					.acceptIfNotNull(endpoint.getConcurrency(),
+							(concurrency) -> container.setConsumersPerQueue(Integer.parseInt(concurrency)));
+		}
+		if (this.containerCustomizer != null) {
+			this.containerCustomizer.configure(container);
+		}
+		return container;
+	}
+
+	protected RabbitAmqpListenerContainer createContainerInstance() {
+		return new RabbitAmqpListenerContainer(this.connection);
+	}
+
+}

spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/config/package-info.java

@@ -0,0 +1,5 @@
+/**
+ * Provides classes for Spring application context support.
+ */
+@org.jspecify.annotations.NullMarked
+package org.springframework.amqp.rabbitmq.client.config;

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

@@ -0,0 +1,392 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.amqp.rabbitmq.client.listener;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+import com.rabbitmq.client.amqp.Connection;
+import com.rabbitmq.client.amqp.Consumer;
+import com.rabbitmq.client.amqp.Resource;
+import org.aopalliance.aop.Advice;
+import org.apache.commons.logging.LogFactory;
+import org.jspecify.annotations.Nullable;
+
+import org.springframework.amqp.AmqpRejectAndDontRequeueException;
+import org.springframework.amqp.core.AcknowledgeMode;
+import org.springframework.amqp.core.AmqpAcknowledgment;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageListener;
+import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler;
+import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.support.ContainerUtils;
+import org.springframework.amqp.rabbitmq.client.RabbitAmqpUtils;
+import org.springframework.aop.framework.ProxyFactory;
+import org.springframework.aop.support.DefaultPointcutAdvisor;
+import org.springframework.core.log.LogAccessor;
+import org.springframework.util.Assert;
+import org.springframework.util.ErrorHandler;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+import org.springframework.util.ObjectUtils;
+
+/**
+ * A listener container for RabbitMQ AMQP 1.0 Consumer.
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.0
+ *
+ */
+public class RabbitAmqpListenerContainer implements MessageListenerContainer {
+
+	private static final LogAccessor LOG = new LogAccessor(LogFactory.getLog(RabbitAmqpListenerContainer.class));
+
+	private final Lock lock = new ReentrantLock();
+
+	private final Connection connection;
+
+	private final MultiValueMap<String, Consumer> queueToConsumers = new LinkedMultiValueMap<>();
+
+	private String @Nullable [] queues;
+
+	private Advice @Nullable [] adviceChain;
+
+	private int initialCredits = 100;
+
+	private int priority;
+
+	private Resource.StateListener @Nullable [] stateListeners;
+
+	private boolean autoSettle = true;
+
+	private boolean defaultRequeue = true;
+
+	private int consumersPerQueue = 1;
+
+	private @Nullable MessageListener messageListener;
+
+	private ErrorHandler errorHandler = new ConditionalRejectingErrorHandler();
+
+	private boolean autoStartup = true;
+
+	private @Nullable String listenerId;
+
+	private Duration gracefulShutdownPeriod = Duration.ofSeconds(30);
+
+	/**
+	 * Construct an instance using the provided connection.
+	 * @param connection to use.
+	 */
+	public RabbitAmqpListenerContainer(Connection connection) {
+		this.connection = connection;
+	}
+
+	@Override
+	public void setQueueNames(String... queueNames) {
+		this.queues = Arrays.copyOf(queueNames, queueNames.length);
+	}
+
+	public void setInitialCredits(int initialCredits) {
+		this.initialCredits = initialCredits;
+	}
+
+	public void setPriority(int priority) {
+		this.priority = priority;
+	}
+
+	public void setStateListeners(Resource.StateListener... stateListeners) {
+		this.stateListeners = Arrays.copyOf(stateListeners, stateListeners.length);
+	}
+
+	@Override
+	public void setAutoStartup(boolean autoStart) {
+		this.autoStartup = autoStart;
+	}
+
+	@Override
+	public boolean isAutoStartup() {
+		return this.autoStartup;
+	}
+
+	/**
+	 * Set an advice chain to apply to the listener.
+	 * @param advices the advice chain.
+	 * @since 2.4.5
+	 */
+	public void setAdviceChain(Advice... advices) {
+		Assert.notNull(advices, "'advices' cannot be null");
+		Assert.noNullElements(advices, "'advices' cannot have null elements");
+		this.adviceChain = Arrays.copyOf(advices, advices.length);
+	}
+
+	/**
+	 * Set to {@code false} to propagate a
+	 * {@link org.springframework.amqp.core.MessageProperties#setAmqpAcknowledgment(AmqpAcknowledgment)}
+	 * for target {@link MessageListener} manual settlement.
+	 * In case of {@link RabbitAmqpMessageListener}, the native {@link Consumer.Context}
+	 * should be used for manual settlement.
+	 * @param autoSettle to call {@link Consumer.Context#accept()} automatically.
+	 */
+	public void setAutoSettle(boolean autoSettle) {
+		this.autoSettle = autoSettle;
+	}
+
+	/**
+	 * Set the default behavior when a message processing has failed.
+	 * When true, messages will be requeued, when false, they will be discarded.
+	 * When true, the default can be overridden by the listener throwing an
+	 * {@link AmqpRejectAndDontRequeueException}. Default true.
+	 * @param defaultRequeue true to requeue by default.
+	 */
+	public void setDefaultRequeue(boolean defaultRequeue) {
+		this.defaultRequeue = defaultRequeue;
+	}
+
+	public void setGracefulShutdownPeriod(Duration gracefulShutdownPeriod) {
+		this.gracefulShutdownPeriod = gracefulShutdownPeriod;
+	}
+
+	/**
+	 * Each queue runs in its own consumer; set this property to create multiple
+	 * consumers for each queue.
+	 * Can be treated as {@code concurrency}, but per queue.
+	 * @param consumersPerQueue the consumers per queue.
+	 */
+	public void setConsumersPerQueue(int consumersPerQueue) {
+		this.consumersPerQueue = consumersPerQueue;
+	}
+
+	public void setErrorHandler(ErrorHandler errorHandler) {
+		this.errorHandler = errorHandler;
+	}
+
+	@Override
+	public void setListenerId(String id) {
+		this.listenerId = id;
+	}
+
+	@Override
+	public void setupMessageListener(MessageListener messageListener) {
+		this.messageListener = messageListener;
+		if (!ObjectUtils.isEmpty(this.adviceChain)) {
+			ProxyFactory factory = new ProxyFactory(messageListener);
+			for (Advice advice : this.adviceChain) {
+				factory.addAdvisor(new DefaultPointcutAdvisor(advice));
+			}
+			factory.setInterfaces(messageListener.getClass().getInterfaces());
+			this.messageListener = (MessageListener) factory.getProxy(getClass().getClassLoader());
+		}
+	}
+
+	@Override
+	public @Nullable Object getMessageListener() {
+		return this.messageListener;
+	}
+
+	@Override
+	public void afterPropertiesSet() {
+		Assert.state(this.queues != null, "At least one queue has to be provided for consuming.");
+		Assert.state(this.messageListener != null, "The 'messageListener' must be provided.");
+
+		this.messageListener.containerAckMode(this.autoSettle ? AcknowledgeMode.AUTO : AcknowledgeMode.MANUAL);
+	}
+
+	@Override
+	public boolean isRunning() {
+		this.lock.lock();
+		try {
+			return !this.queueToConsumers.isEmpty();
+		}
+		finally {
+			this.lock.unlock();
+		}
+	}
+
+	@Override
+	@SuppressWarnings("NullAway") // Dataflow analysis limitation
+	public void start() {
+		this.lock.lock();
+		try {
+			if (this.queueToConsumers.isEmpty()) {
+				for (String queue : this.queues) {
+					for (int i = 0; i < this.consumersPerQueue; i++) {
+						Consumer consumer =
+								this.connection.consumerBuilder()
+										.queue(queue)
+										.priority(this.priority)
+										.initialCredits(this.initialCredits)
+										.listeners(this.stateListeners)
+										.messageHandler(this::invokeListener)
+										.build();
+						this.queueToConsumers.add(queue, consumer);
+					}
+				}
+			}
+		}
+		finally {
+			this.lock.unlock();
+		}
+	}
+
+	private void invokeListener(Consumer.Context context, com.rabbitmq.client.amqp.Message amqpMessage) {
+		try {
+			doInvokeListener(context, amqpMessage);
+			if (this.autoSettle) {
+				context.accept();
+			}
+		}
+		catch (Exception ex) {
+			if (!handleSpecialErrors(ex, context)) {
+				try {
+					this.errorHandler.handleError(ex);
+					// If error handler does not re-throw an exception, treat it as a successful processing result.
+					context.accept();
+				}
+				catch (Exception rethrow) {
+					if (!handleSpecialErrors(rethrow, context)) {
+						if (this.defaultRequeue) {
+							context.requeue();
+						}
+						else {
+							context.discard();
+						}
+						LOG.error(rethrow, () ->
+								"The 'errorHandler' has thrown an exception. The '" + amqpMessage + "' is "
+										+ (this.defaultRequeue ? "re-queued." : "discarded."));
+					}
+				}
+			}
+		}
+	}
+
+	@SuppressWarnings("NullAway") // Dataflow analysis limitation
+	private void doInvokeListener(Consumer.Context context, com.rabbitmq.client.amqp.Message amqpMessage) {
+		Consumer.@Nullable Context contextToUse = this.autoSettle ? null : context;
+		if (this.messageListener instanceof RabbitAmqpMessageListener amqpMessageListener) {
+			amqpMessageListener.onAmqpMessage(amqpMessage, contextToUse);
+		}
+		else {
+			Message message = RabbitAmqpUtils.fromAmqpMessage(amqpMessage, contextToUse);
+			this.messageListener.onMessage(message);
+		}
+	}
+
+	private boolean handleSpecialErrors(Exception ex, Consumer.Context context) {
+		if (ContainerUtils.shouldRequeue(this.defaultRequeue, ex, LOG.getLog())) {
+			context.requeue();
+			return true;
+		}
+		if (ContainerUtils.isAmqpReject(ex)) {
+			context.discard();
+			return true;
+		}
+		if (ContainerUtils.isImmediateAcknowledge(ex)) {
+			context.accept();
+			return true;
+		}
+		return false;
+	}
+
+	@Override
+	public void stop() {
+		stop(() -> {
+		});
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	public void stop(Runnable callback) {
+		this.lock.lock();
+		try {
+			CompletableFuture<Void>[] completableFutures =
+					this.queueToConsumers.values().stream()
+							.flatMap(List::stream)
+							.peek(Consumer::pause)
+							.map((consumer) ->
+									CompletableFuture.supplyAsync(() -> {
+										try (consumer) {
+											while (consumer.unsettledMessageCount() > 0) {
+												Thread.sleep(100);
+											}
+										}
+										catch (InterruptedException ex) {
+											Thread.currentThread().interrupt();
+											throw new RuntimeException(ex);
+										}
+										return null;
+									}))
+							.toArray(CompletableFuture[]::new);
+
+			CompletableFuture.allOf(completableFutures)
+					.orTimeout(this.gracefulShutdownPeriod.toMillis(), TimeUnit.MILLISECONDS)
+					.whenComplete((unused, throwable) -> {
+						this.queueToConsumers.clear();
+						callback.run();
+					});
+		}
+		finally {
+			this.lock.unlock();
+		}
+	}
+
+	/**
+	 * Pause all the consumer for all queues.
+	 */
+	public void pause() {
+		this.queueToConsumers.values()
+				.stream()
+				.flatMap(List::stream)
+				.forEach(Consumer::pause);
+	}
+
+	/**
+	 * Resume all the consumer for all queues.
+	 */
+	public void resume() {
+		this.queueToConsumers.values()
+				.stream()
+				.flatMap(List::stream)
+				.forEach(Consumer::unpause);
+	}
+
+	/**
+	 * Pause all the consumer for specific queue.
+	 */
+	public void pause(String queueName) {
+		List<Consumer> consumers = this.queueToConsumers.get(queueName);
+		if (consumers != null) {
+			consumers.forEach(Consumer::pause);
+		}
+	}
+
+	/**
+	 * Resume all the consumer for specific queue.
+	 */
+	public void resume(String queueName) {
+		List<Consumer> consumers = this.queueToConsumers.get(queueName);
+		if (consumers != null) {
+			consumers.forEach(Consumer::unpause);
+		}
+	}
+
+}

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

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2021-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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.amqp.rabbitmq.client.listener;
+
+import com.rabbitmq.client.amqp.Consumer;
+import com.rabbitmq.client.amqp.Message;
+import org.jspecify.annotations.Nullable;
+
+import org.springframework.amqp.core.MessageListener;
+
+/**
+ * A message listener that receives native AMQP 1.0 messages from RabbitMQ.
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.0
+ */
+public interface RabbitAmqpMessageListener extends MessageListener {
+
+	/**
+	 * Process an AMQP message.
+	 * @param message the message to process.
+	 * @param context the consumer context to settle message.
+	 *                Null if container is configured for {@code autoSettle}.
+	 */
+	void onAmqpMessage(Message message, Consumer.@Nullable Context context);
+
+}

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

@@ -0,0 +1,72 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.amqp.rabbitmq.client.listener;
+
+import java.lang.reflect.Method;
+
+import com.rabbitmq.client.amqp.Consumer;
+import org.jspecify.annotations.Nullable;
+
+import org.springframework.amqp.rabbit.listener.adapter.InvocationResult;
+import org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter;
+import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
+import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException;
+import org.springframework.amqp.rabbitmq.client.RabbitAmqpUtils;
+
+/**
+ * A {@link MessagingMessageListenerAdapter} extension for the {@link RabbitAmqpMessageListener}.
+ * Provides these arguments for the {@link #getHandlerAdapter()} invocation:
+ * <ul>
+ * <li>{@link com.rabbitmq.client.amqp.Message} - the native AMQP 1.0 message without any conversions</li>
+ * <li>{@link org.springframework.amqp.core.Message} - Spring AMQP message abstraction as conversion result from the native AMQP 1.0 message</li>
+ * <li>{@link org.springframework.messaging.Message} - Spring Messaging abstraction as conversion result from the Spring AMQP message</li>
+ * <li>{@link Consumer.Context} - RabbitMQ AMQP client consumer settlement API.</li>
+ * <li>{@link org.springframework.amqp.core.AmqpAcknowledgment} - Spring AMQP acknowledgment abstraction: delegates to the {@link Consumer.Context}</li>
+ * </ul>
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.0
+ */
+public class RabbitAmqpMessageListenerAdapter extends MessagingMessageListenerAdapter
+		implements RabbitAmqpMessageListener {
+
+	public RabbitAmqpMessageListenerAdapter(@Nullable Object bean, @Nullable Method method, boolean returnExceptions,
+			@Nullable RabbitListenerErrorHandler errorHandler) {
+
+		super(bean, method, returnExceptions, errorHandler);
+	}
+
+	@Override
+	public void onAmqpMessage(com.rabbitmq.client.amqp.Message amqpMessage, Consumer.@Nullable Context context) {
+		try {
+			org.springframework.amqp.core.Message springMessage = RabbitAmqpUtils.fromAmqpMessage(amqpMessage, context);
+			org.springframework.messaging.Message<?> messagingMessage = toMessagingMessage(springMessage);
+			InvocationResult result = getHandlerAdapter()
+					.invoke(messagingMessage,
+							springMessage, springMessage.getMessageProperties().getAmqpAcknowledgment(),
+							amqpMessage, context);
+			if (result.getReturnValue() != null) {
+				logger.warn("Replies are not currently supported with RabbitMQ AMQP 1.0 listeners");
+			}
+		}
+		catch (Exception ex) {
+			throw new ListenerExecutionFailedException("Failed to invoke listener", ex);
+		}
+	}
+
+}

spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/package-info.java

@@ -0,0 +1,5 @@
+/**
+ * Provides Spring support for RabbitMQ AMQP 1.0 Consumer.
+ */
+@org.jspecify.annotations.NullMarked
+package org.springframework.amqp.rabbitmq.client.listener;