From 3154e17d21593661daee02180d5111ba9d8b3010 Mon Sep 17 00:00:00 2001 From: Jun Cho Date: Tue, 28 Jul 2026 21:12:08 +0900 Subject: [PATCH 1/5] GH-11194: Use simple confirms in `AmqpOutboundEndpoint` Fixes: https://github.com/spring-projects/spring-integration/issues/11194 The `waitForConfirm` option only took effect when the message carried `CorrelationData`, which requires a connection factory configured for correlated publisher confirms. With `ConfirmType.SIMPLE` there is no correlation data, so the option was silently ignored for single messages, even though the `multiSend` path already used Spring AMQP's scoped `waitForConfirmsOrDie()` for iterable payloads. * Send within a `RabbitTemplate.invoke()` scope and call `waitForConfirmsOrDie()` when `waitForConfirm` is true, there is no correlation data and the connection factory is configured for simple publisher confirms * Warn in `endpointInit()` when `waitForConfirm` is set but the connection factory supports no publisher confirms at all, instead of ignoring the option silently * Correct the `setWaitForConfirm()` Javadoc, the XSD and the reference manual, which stated that a `confirm-correlation-expression` is always required * Add `simpleConfirmsOk()` and `simpleConfirmsWithReject()` to `AmqpOutboundEndpointTests2` Signed-off-by: Jun Cho --- .../amqp/outbound/AmqpOutboundEndpoint.java | 37 ++++++++++++++ .../amqp/config/spring-integration-amqp.xsd | 3 ++ .../outbound/AmqpOutboundEndpointTests2.java | 51 +++++++++++++++++++ .../pages/amqp/outbound-channel-adapter.adoc | 5 +- 4 files changed, 95 insertions(+), 1 deletion(-) diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java index b81917096a5..bf812fc8bbf 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java @@ -48,6 +48,7 @@ * @author Gary Russell * @author Artem Bilan * @author Ngoc Nhan + * @author Jun Cho * * @since 2.1 */ @@ -95,6 +96,12 @@ public void setExpectReply(boolean expectReply) { * received within the confirm timeout or a negative acknowledgment or returned * message is received, an exception will be thrown. Does not apply to the gateway * since it blocks awaiting the reply. + *

+ * When there is no correlation data for the message and the connection factory is + * configured for simple publisher confirms, the message is sent within the scope of a + * {@code RabbitTemplate.invoke()} operation and {@code waitForConfirmsOrDie()} is used + * instead. A returned message does not fail the send in that case; it is published to + * the return channel, if one is configured. * @param waitForConfirm true to block until the confirmation or timeout is received. * @since 5.2 * @see #setConfirmTimeout(long) @@ -156,6 +163,10 @@ protected void endpointInit() { + "does not support returned messages; none will be received"); } } + if (this.waitForConfirm && !usesSimplePublisherConfirms() && !usesCorrelatedPublisherConfirms()) { + this.logger.warn("The 'waitForConfirm' is set to true but the underlying connection factory " + + "does not support publisher confirms; no confirmations will be awaited"); + } Duration confirmTimeout = getConfirmTimeout(); if (confirmTimeout != null) { this.waitForConfirmTimeout = confirmTimeout; @@ -181,6 +192,10 @@ protected void doStop() { multiSend(requestMessage, exchangeName, routingKey); return null; } + else if (this.waitForConfirm && correlationData == null && usesSimplePublisherConfirms()) { + sendAndWaitForConfirms(exchangeName, routingKey, requestMessage); + return null; + } else { send(exchangeName, routingKey, requestMessage, correlationData); if (this.waitForConfirm && correlationData != null) { @@ -208,6 +223,28 @@ private void multiSend(Message requestMessage, @Nullable String exchangeName, }); } + private void sendAndWaitForConfirms(@Nullable String exchangeName, @Nullable String routingKey, + Message requestMessage) { + + RabbitTemplate rabbitTemplateToUse = this.rabbitTemplate; + Assert.notNull(rabbitTemplateToUse, "The 'RabbitTemplate' must be provided to wait for confirms."); + rabbitTemplateToUse.<@Nullable Object>invoke(template -> { + doRabbitSend(exchangeName, routingKey, requestMessage, null, rabbitTemplateToUse); + template.waitForConfirmsOrDie(this.waitForConfirmTimeout.toMillis()); + return null; + }); + } + + private boolean usesSimplePublisherConfirms() { + return this.rabbitTemplate != null + && this.rabbitTemplate.getConnectionFactory().isSimplePublisherConfirms(); + } + + private boolean usesCorrelatedPublisherConfirms() { + return this.rabbitTemplate != null + && this.rabbitTemplate.getConnectionFactory().isPublisherConfirms(); + } + private void waitForConfirm(Message requestMessage, CorrelationData correlationData) { try { Confirm confirm = correlationData.getFuture().get(this.waitForConfirmTimeout.toMillis(), diff --git a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp.xsd b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp.xsd index 14163860e77..31313470bdb 100644 --- a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp.xsd +++ b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp.xsd @@ -70,6 +70,9 @@ been received. Requires a template configured for returns. If a confirm is not received within the confirm timeout or a negative acknowledgment or returned message is received, an exception will be thrown. + When there is no correlation data for the message and the connection factory is + configured for simple publisher confirms, the message is sent within the scope of + a RabbitTemplate.invoke() operation and waitForConfirmsOrDie() is used instead. diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests2.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests2.java index 7f74a3a8942..27ef9911f2e 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests2.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests2.java @@ -53,6 +53,7 @@ /** * @author Gary Russell * @author Artem Bilan + * @author Jun Cho * * @since 5.2 * @@ -63,14 +64,18 @@ public class AmqpOutboundEndpointTests2 implements RabbitTestContainer { static final String QUEUE_TEST_CONFIRM_OK = "testConfirmOk"; + static final String QUEUE_TEST_SIMPLE_CONFIRMS = "testSimpleConfirms"; + @BeforeAll static void initQueue() throws IOException, InterruptedException { RABBITMQ.execInContainer("rabbitmqadmin", "declare", "queue", "name=" + QUEUE_TEST_CONFIRM_OK); + RABBITMQ.execInContainer("rabbitmqadmin", "declare", "queue", "name=" + QUEUE_TEST_SIMPLE_CONFIRMS); } @AfterAll static void deleteQueue() throws IOException, InterruptedException { RABBITMQ.execInContainer("rabbitmqadmin", "delete", "queue", "name=" + QUEUE_TEST_CONFIRM_OK); + RABBITMQ.execInContainer("rabbitmqadmin", "delete", "queue", "name=" + QUEUE_TEST_SIMPLE_CONFIRMS); } @Test @@ -118,6 +123,32 @@ void testWithReject(@Autowired IntegrationFlow flow, @Autowired RabbitAdmin admi admin.deleteQueue(queue.getName()); } + @Test + void simpleConfirmsOk(@Autowired IntegrationFlow simpleConfirmsFlow, @Autowired RabbitTemplate template) { + simpleConfirmsFlow.getInputChannel() + .send(new GenericMessage<>("test", Collections.singletonMap("rk", QUEUE_TEST_SIMPLE_CONFIRMS))); + assertThat(template.receive(QUEUE_TEST_SIMPLE_CONFIRMS)).isNotNull(); + } + + @Test + void simpleConfirmsWithReject(@Autowired IntegrationFlow simpleConfirmsFlow, @Autowired RabbitAdmin admin, + @Autowired RabbitTemplate template) { + + Queue queue = QueueBuilder.nonDurable().autoDelete().maxLength(1L).overflow(Overflow.rejectPublish).build(); + admin.declareQueue(queue); + try { + simpleConfirmsFlow.getInputChannel() + .send(new GenericMessage<>("test", Collections.singletonMap("rk", queue.getName()))); + assertThatThrownBy(() -> simpleConfirmsFlow.getInputChannel() + .send(new GenericMessage<>("test", Collections.singletonMap("rk", queue.getName())))) + .hasCauseInstanceOf(AmqpException.class); + assertThat(template.receive(queue.getName())).isNotNull(); + } + finally { + admin.deleteQueue(queue.getName()); + } + } + @Configuration(proxyBeanMethods = false) @EnableIntegration public static class Config { @@ -138,6 +169,14 @@ public IntegrationFlow flow2(RabbitTemplate template) { .routingKeyFunction(msg -> msg.getHeaders().get("rk", String.class))); } + @Bean + public IntegrationFlow simpleConfirmsFlow(RabbitTemplate simpleConfirmsTemplate) { + return f -> f.handle(Amqp.outboundAdapter(simpleConfirmsTemplate) + .exchangeName("") + .routingKeyFunction(msg -> msg.getHeaders().get("rk", String.class)) + .waitForConfirm(true)); + } + @Bean public CachingConnectionFactory cf() { CachingConnectionFactory ccf = new CachingConnectionFactory(RabbitTestContainer.amqpPort()); @@ -154,6 +193,18 @@ public RabbitTemplate template(ConnectionFactory cf) { return rabbitTemplate; } + @Bean + public CachingConnectionFactory simpleConfirmsCf() { + CachingConnectionFactory ccf = new CachingConnectionFactory(RabbitTestContainer.amqpPort()); + ccf.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.SIMPLE); + return ccf; + } + + @Bean + public RabbitTemplate simpleConfirmsTemplate(ConnectionFactory simpleConfirmsCf) { + return new RabbitTemplate(simpleConfirmsCf); + } + @Bean public RabbitAdmin admin(ConnectionFactory cf) { return new RabbitAdmin(cf); diff --git a/src/reference/antora/modules/ROOT/pages/amqp/outbound-channel-adapter.adoc b/src/reference/antora/modules/ROOT/pages/amqp/outbound-channel-adapter.adoc index 48fa652acc4..95fd31c505c 100644 --- a/src/reference/antora/modules/ROOT/pages/amqp/outbound-channel-adapter.adoc +++ b/src/reference/antora/modules/ROOT/pages/amqp/outbound-channel-adapter.adoc @@ -123,10 +123,13 @@ Pending confirms are checked every 50% of this value, so the actual time a nack Also see xref:amqp/alternative-confirms-returns.adoc[Alternative Mechanism for Publisher Confirms and Returns]. Default none (nacks will not be generated). <14> When set to true, the calling thread will block, waiting for a publisher confirmation. -This requires a `RabbitTemplate` configured for `confirms` as well as a `confirm-correlation-expression`. +This requires a `RabbitTemplate` configured for `confirms`. +With correlated publisher confirms, the message also needs correlation data, which is usually provided by a `confirm-correlation-expression`. The thread will block for up to `confirm-timeout` (or 5 seconds by default). If a timeout occurs, a `MessageTimeoutException` will be thrown. If returns are enabled and a message is returned, or any other exception occurs while awaiting the confirmation, a `MessageHandlingException` will be thrown, with an appropriate message. +When there is no correlation data for the message and the connection factory is configured for simple publisher confirms, the message is sent within the scope of a `RabbitTemplate.invoke()` operation and `waitForConfirmsOrDie()` is used instead. +A returned message does not fail the send in that case; it is published to the `return-channel`, if one is configured. <15> The channel to which returned messages are sent. When provided, the underlying AMQP template is configured to return undeliverable messages to the adapter. When there is no `ErrorMessageStrategy` configured, the message is constructed from the data received from AMQP, with the following additional headers: `amqp_returnReplyCode`, `amqp_returnReplyText`, `amqp_returnExchange`, `amqp_returnRoutingKey`. From 8626f64b1987099933001e984ac8ff555e85755a Mon Sep 17 00:00:00 2001 From: Jun Cho Date: Thu, 6 Aug 2026 13:22:55 +0900 Subject: [PATCH 2/5] Add what's new entry for the simple publisher confirms support Signed-off-by: Jun Cho --- src/reference/antora/modules/ROOT/pages/whats-new.adoc | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/src/reference/antora/modules/ROOT/pages/whats-new.adoc b/src/reference/antora/modules/ROOT/pages/whats-new.adoc index 5a5ef019e10..46d2c5b2b32 100644 --- a/src/reference/antora/modules/ROOT/pages/whats-new.adoc +++ b/src/reference/antora/modules/ROOT/pages/whats-new.adoc @@ -22,3 +22,11 @@ Java 17 is still the baseline, but Java 25 is supported. The `DefaultErrorMessageStrategy` now preserves the original headers when building the `ErrorMessage`. See xref:error-handling.adoc[] for more information. + +[[x7.2-amqp-changes]] +=== AMQP Support Changes + +The `wait-for-confirm` option on the outbound channel adapter now also works with simple publisher confirms. +When the message has no correlation data, the send is performed within the scope of a `RabbitTemplate.invoke()` operation and `waitForConfirmsOrDie()` is used. +A warning is now logged when the option is set but the connection factory supports no publisher confirms at all. +See xref:amqp/outbound-channel-adapter.adoc[] for more information. From 3d1846d63beed55ee8d19eba9ef2a4ebb4a1c4d5 Mon Sep 17 00:00:00 2001 From: Jun Cho Date: Thu, 6 Aug 2026 13:23:22 +0900 Subject: [PATCH 3/5] Clean up the simpleConfirmsWithReject test Extract the queue name and the message, drop the `receive()` assertion that the first send already proves, and assert the `AmqpIOException` that the broker nack surfaces instead of the generic `AmqpException`. Signed-off-by: Jun Cho --- .../outbound/AmqpOutboundEndpointTests2.java | 18 ++++++++---------- 1 file changed, 8 insertions(+), 10 deletions(-) diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests2.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests2.java index 27ef9911f2e..ca09c76e3f8 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests2.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests2.java @@ -25,6 +25,7 @@ import org.junit.jupiter.api.Test; import org.springframework.amqp.AmqpException; +import org.springframework.amqp.AmqpIOException; import org.springframework.amqp.core.Queue; import org.springframework.amqp.core.QueueBuilder; import org.springframework.amqp.core.QueueBuilder.Overflow; @@ -131,21 +132,18 @@ void simpleConfirmsOk(@Autowired IntegrationFlow simpleConfirmsFlow, @Autowired } @Test - void simpleConfirmsWithReject(@Autowired IntegrationFlow simpleConfirmsFlow, @Autowired RabbitAdmin admin, - @Autowired RabbitTemplate template) { - + void simpleConfirmsWithReject(@Autowired IntegrationFlow simpleConfirmsFlow, @Autowired RabbitAdmin admin) { Queue queue = QueueBuilder.nonDurable().autoDelete().maxLength(1L).overflow(Overflow.rejectPublish).build(); + String queueName = queue.getName(); admin.declareQueue(queue); + GenericMessage message = new GenericMessage<>("test", Collections.singletonMap("rk", queueName)); try { - simpleConfirmsFlow.getInputChannel() - .send(new GenericMessage<>("test", Collections.singletonMap("rk", queue.getName()))); - assertThatThrownBy(() -> simpleConfirmsFlow.getInputChannel() - .send(new GenericMessage<>("test", Collections.singletonMap("rk", queue.getName())))) - .hasCauseInstanceOf(AmqpException.class); - assertThat(template.receive(queue.getName())).isNotNull(); + simpleConfirmsFlow.getInputChannel().send(message); + assertThatThrownBy(() -> simpleConfirmsFlow.getInputChannel().send(message)) + .hasCauseInstanceOf(AmqpIOException.class); } finally { - admin.deleteQueue(queue.getName()); + admin.deleteQueue(queueName); } } From f0fece4f3d005836c8d2d6a95e9732cdb94a7e1e Mon Sep 17 00:00:00 2001 From: Jun Cho Date: Thu, 6 Aug 2026 13:24:15 +0900 Subject: [PATCH 4/5] Document the scoped send conflict with BoundRabbitChannelAdvice The adapter's scoped send and the advice both open a scoped operation on the same template, and nested scopes are not supported. Also mirror the returned message note from the Javadoc in the XSD. Signed-off-by: Jun Cho --- .../integration/amqp/config/spring-integration-amqp.xsd | 2 ++ .../modules/ROOT/pages/amqp/outbound-channel-adapter.adoc | 6 ++++-- .../antora/modules/ROOT/pages/amqp/strict-ordering.adoc | 4 ++++ 3 files changed, 10 insertions(+), 2 deletions(-) diff --git a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp.xsd b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp.xsd index 31313470bdb..7d736888c2d 100644 --- a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp.xsd +++ b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp.xsd @@ -73,6 +73,8 @@ When there is no correlation data for the message and the connection factory is configured for simple publisher confirms, the message is sent within the scope of a RabbitTemplate.invoke() operation and waitForConfirmsOrDie() is used instead. + A returned message does not fail the send in that case; it is published to the + return channel, if one is configured. diff --git a/src/reference/antora/modules/ROOT/pages/amqp/outbound-channel-adapter.adoc b/src/reference/antora/modules/ROOT/pages/amqp/outbound-channel-adapter.adoc index 95fd31c505c..7050ca042f5 100644 --- a/src/reference/antora/modules/ROOT/pages/amqp/outbound-channel-adapter.adoc +++ b/src/reference/antora/modules/ROOT/pages/amqp/outbound-channel-adapter.adoc @@ -128,8 +128,10 @@ With correlated publisher confirms, the message also needs correlation data, whi The thread will block for up to `confirm-timeout` (or 5 seconds by default). If a timeout occurs, a `MessageTimeoutException` will be thrown. If returns are enabled and a message is returned, or any other exception occurs while awaiting the confirmation, a `MessageHandlingException` will be thrown, with an appropriate message. -When there is no correlation data for the message and the connection factory is configured for simple publisher confirms, the message is sent within the scope of a `RabbitTemplate.invoke()` operation and `waitForConfirmsOrDie()` is used instead. -A returned message does not fail the send in that case; it is published to the `return-channel`, if one is configured. +With simple publisher confirms and no correlation data, the adapter sends the message within the scope of a `RabbitTemplate.invoke()` operation and calls `waitForConfirmsOrDie()` instead. +A returned message then does not fail the send; it is published to the `return-channel`, if one is configured. +This option cannot be combined with a `BoundRabbitChannelAdvice` that shares the same `RabbitTemplate`, since both open a scoped operation. +Also see xref:amqp/strict-ordering.adoc[Strict Message Ordering]. <15> The channel to which returned messages are sent. When provided, the underlying AMQP template is configured to return undeliverable messages to the adapter. When there is no `ErrorMessageStrategy` configured, the message is constructed from the data received from AMQP, with the following additional headers: `amqp_returnReplyCode`, `amqp_returnReplyText`, `amqp_returnExchange`, `amqp_returnRoutingKey`. diff --git a/src/reference/antora/modules/ROOT/pages/amqp/strict-ordering.adoc b/src/reference/antora/modules/ROOT/pages/amqp/strict-ordering.adoc index 0fc1d3dc821..7b370a2f35a 100644 --- a/src/reference/antora/modules/ROOT/pages/amqp/strict-ordering.adoc +++ b/src/reference/antora/modules/ROOT/pages/amqp/strict-ordering.adoc @@ -55,5 +55,9 @@ Notice that the same `RabbitTemplate` (which implements `RabbitOperations`) is u The advice runs the downstream flow within the template's `invoke` method so that all operations run on the same channel. If the optional timeout is provided, when the flow completes, the advice calls the `waitForConfirmsOrDie` method, which throws an exception if the confirmations are not received within the specified time. +NOTE: The outbound adapter opens a scoped operation of its own when `multi-send` is used, and also when `wait-for-confirm` is set with simple publisher confirms. +Do not use those options on an adapter whose flow is already wrapped by a `BoundRabbitChannelAdvice` that shares the same `RabbitTemplate`, because nested scopes are not supported. +The advice already waits for the confirmations of all the messages sent within its scope. + IMPORTANT: There must be no thread hands-off in the downstream flow (`QueueChannel`, `ExecutorChannel`, and others). From 0d56c7f494fecc3fba54640122354826610fa3cf Mon Sep 17 00:00:00 2001 From: Jun Cho Date: Sun, 9 Aug 2026 23:43:17 +0900 Subject: [PATCH 5/5] Declare the reject queue on the publishing connection factory `simpleConfirmsWithReject` declared the queue through the `admin` bean, which is bound to the correlated connection factory, while the flow publishes through `simpleConfirmsCf`. An unrouted first publish leaves the queue below its limit, so the second publish is confirmed instead of nacked and the test fails. Signed-off-by: Jun Cho --- .../amqp/outbound/AmqpOutboundEndpointTests2.java | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests2.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests2.java index ca09c76e3f8..9d3e52f1077 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests2.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests2.java @@ -132,10 +132,12 @@ void simpleConfirmsOk(@Autowired IntegrationFlow simpleConfirmsFlow, @Autowired } @Test - void simpleConfirmsWithReject(@Autowired IntegrationFlow simpleConfirmsFlow, @Autowired RabbitAdmin admin) { + void simpleConfirmsWithReject(@Autowired IntegrationFlow simpleConfirmsFlow, + @Autowired RabbitAdmin simpleConfirmsAdmin) { + Queue queue = QueueBuilder.nonDurable().autoDelete().maxLength(1L).overflow(Overflow.rejectPublish).build(); String queueName = queue.getName(); - admin.declareQueue(queue); + simpleConfirmsAdmin.declareQueue(queue); GenericMessage message = new GenericMessage<>("test", Collections.singletonMap("rk", queueName)); try { simpleConfirmsFlow.getInputChannel().send(message); @@ -143,7 +145,7 @@ void simpleConfirmsWithReject(@Autowired IntegrationFlow simpleConfirmsFlow, @Au .hasCauseInstanceOf(AmqpIOException.class); } finally { - admin.deleteQueue(queueName); + simpleConfirmsAdmin.deleteQueue(queueName); } } @@ -203,6 +205,11 @@ public RabbitTemplate simpleConfirmsTemplate(ConnectionFactory simpleConfirmsCf) return new RabbitTemplate(simpleConfirmsCf); } + @Bean + public RabbitAdmin simpleConfirmsAdmin(ConnectionFactory simpleConfirmsCf) { + return new RabbitAdmin(simpleConfirmsCf); + } + @Bean public RabbitAdmin admin(ConnectionFactory cf) { return new RabbitAdmin(cf);