diff --git a/src/main/java/com/rabbitmq/client/impl/AMQConnection.java b/src/main/java/com/rabbitmq/client/impl/AMQConnection.java index 8f7c50c83..382d32499 100644 --- a/src/main/java/com/rabbitmq/client/impl/AMQConnection.java +++ b/src/main/java/com/rabbitmq/client/impl/AMQConnection.java @@ -215,6 +215,22 @@ public Map getServerProperties() { return _serverProperties; } + boolean doesServerHaveCapability(ServerCapability capability) { + return doesServerHaveCapability(_serverProperties, capability); + } + + public boolean doesBrokerAcceptClientSentBasicCancelOk() { + return doesServerHaveCapability(ServerCapability.ACCEPT_CONSUMER_CANCEL_OK); + } + + @SuppressWarnings("unchecked") + static boolean doesServerHaveCapability(Map serverProperties, + ServerCapability capability) { + Map capabilities = (Map) + serverProperties.getOrDefault("capabilities", Collections.emptyMap()); + return Boolean.TRUE.equals(capabilities.get(capability.getName())); + } + public AMQConnection(ConnectionParams params, FrameHandler frameHandler) { this(params, frameHandler, new NoOpMetricsCollector(), ObservationCollector.NO_OP); } diff --git a/src/main/java/com/rabbitmq/client/impl/ChannelN.java b/src/main/java/com/rabbitmq/client/impl/ChannelN.java index ea2df2d71..e6856f768 100644 --- a/src/main/java/com/rabbitmq/client/impl/ChannelN.java +++ b/src/main/java/com/rabbitmq/client/impl/ChannelN.java @@ -394,6 +394,10 @@ private void releaseChannel() { Basic.Cancel m = (Basic.Cancel)method; String consumerTag = m.getConsumerTag(); Consumer callback = _consumers.remove(consumerTag); + if (getConnection().doesBrokerAcceptClientSentBasicCancelOk()) { + // The server will remove the consumer after a short timeout in any case + transmit(new Basic.CancelOk(consumerTag)); + } // Not finding any matching consumer isn't necessarily an indication of an issue anywhere. // Sometimes there's a natural race condition between consumer management on the server and client ends. // E.g. Channel#basicCancel called just before a basic.cancel for the same consumer tag is received. diff --git a/src/main/java/com/rabbitmq/client/impl/ServerCapability.java b/src/main/java/com/rabbitmq/client/impl/ServerCapability.java new file mode 100644 index 000000000..a35b43060 --- /dev/null +++ b/src/main/java/com/rabbitmq/client/impl/ServerCapability.java @@ -0,0 +1,44 @@ +// Copyright (c) 2026 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries. +// +// This software, the RabbitMQ Java client library, is triple-licensed under the +// Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2 +// ("GPL") and the Apache License version 2 ("ASL"). For the MPL, please see +// LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2. For the ASL, +// please see LICENSE-APACHE2. +// +// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND, +// either express or implied. See the LICENSE file for specific language governing +// rights and limitations of this software. +// +// If you have any questions regarding licensing, please contact us at +// info@rabbitmq.com. + +package com.rabbitmq.client.impl; + +/** + * Capabilities a RabbitMQ node advertises in the capabilities table + * of the connection.start server properties. + */ +enum ServerCapability { + + PUBLISHER_CONFIRMS("publisher_confirms"), + EXCHANGE_EXCHANGE_BINDINGS("exchange_exchange_bindings"), + BASIC_NACK("basic.nack"), + CONSUMER_CANCEL_NOTIFY("consumer_cancel_notify"), + CONNECTION_BLOCKED("connection.blocked"), + CONSUMER_PRIORITIES("consumer_priorities"), + AUTHENTICATION_FAILURE_CLOSE("authentication_failure_close"), + PER_CONSUMER_QOS("per_consumer_qos"), + DIRECT_REPLY_TO("direct_reply_to"), + ACCEPT_CONSUMER_CANCEL_OK("accept_consumer_cancel_ok"); + + private final String name; + + ServerCapability(String name) { + this.name = name; + } + + String getName() { + return name; + } +} diff --git a/src/test/java/com/rabbitmq/client/impl/AMQConnectionCapabilitiesTest.java b/src/test/java/com/rabbitmq/client/impl/AMQConnectionCapabilitiesTest.java new file mode 100644 index 000000000..51a1ba464 --- /dev/null +++ b/src/test/java/com/rabbitmq/client/impl/AMQConnectionCapabilitiesTest.java @@ -0,0 +1,52 @@ +// Copyright (c) 2026 Broadcom. All Rights Reserved. +// The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries. +// +// This software, the RabbitMQ Java client library, is triple-licensed under the +// Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2 +// ("GPL") and the Apache License version 2 ("ASL"). For the MPL, please see +// LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2. For the ASL, +// please see LICENSE-APACHE2. +// +// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND, +// either express or implied. See the LICENSE file for specific language governing +// rights and limitations of this software. +// +// If you have any questions regarding licensing, please contact us at +// info@rabbitmq.com. + +package com.rabbitmq.client.impl; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.Collections; +import org.junit.jupiter.api.Test; + +public class AMQConnectionCapabilitiesTest { + + @Test + void capabilitySetToTrueIsPresent() { + assertThat(AMQConnection.doesServerHaveCapability( + Collections.singletonMap("capabilities", Collections.singletonMap("accept_consumer_cancel_ok", true)), + ServerCapability.ACCEPT_CONSUMER_CANCEL_OK)).isTrue(); + } + + @Test + void capabilitySetToFalseIsAbsent() { + assertThat(AMQConnection.doesServerHaveCapability( + Collections.singletonMap("capabilities", Collections.singletonMap("accept_consumer_cancel_ok", false)), + ServerCapability.ACCEPT_CONSUMER_CANCEL_OK)).isFalse(); + } + + @Test + void missingCapabilityIsAbsent() { + assertThat(AMQConnection.doesServerHaveCapability( + Collections.singletonMap("capabilities", Collections.singletonMap("consumer_cancel_notify", true)), + ServerCapability.ACCEPT_CONSUMER_CANCEL_OK)).isFalse(); + } + + @Test + void capabilityIsAbsentWithoutCapabilitiesTable() { + assertThat(AMQConnection.doesServerHaveCapability(Collections.emptyMap(), ServerCapability.ACCEPT_CONSUMER_CANCEL_OK)) + .isFalse(); + } +} diff --git a/src/test/java/com/rabbitmq/client/test/ChannelNTest.java b/src/test/java/com/rabbitmq/client/test/ChannelNTest.java index 1ec24e1c5..2a690f14c 100644 --- a/src/test/java/com/rabbitmq/client/test/ChannelNTest.java +++ b/src/test/java/com/rabbitmq/client/test/ChannelNTest.java @@ -15,7 +15,9 @@ package com.rabbitmq.client.test; +import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Command; +import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.Method; import com.rabbitmq.client.TrafficListener; import com.rabbitmq.client.impl.*; @@ -55,6 +57,53 @@ public void serverBasicCancelForUnknownConsumerDoesNotThrowException() throws Ex channel.processAsync(new AMQCommand(method)); } + @Test + public void serverBasicCancelIsAnsweredWithCancelOkWhenBrokerAcceptsIt() throws Exception { + TrafficListener trafficListener = Mockito.mock(TrafficListener.class); + AMQConnection connection = connectionAcceptingConsumerCancelOk(true, trafficListener); + ChannelN channel = channelWithConsumer(connection, "ctag"); + + channel.processAsync(new AMQCommand(new AMQImpl.Basic.Cancel.Builder().consumerTag("ctag").build())); + + Mockito.verify(trafficListener, Mockito.times(1)).write( + Mockito.argThat(c -> c.getMethod() instanceof AMQP.Basic.CancelOk)); + } + + @Test + public void serverBasicCancelIsNotAnsweredWhenBrokerDoesNotAcceptCancelOk() throws Exception { + TrafficListener trafficListener = Mockito.mock(TrafficListener.class); + AMQConnection connection = connectionAcceptingConsumerCancelOk(false, trafficListener); + ChannelN channel = channelWithConsumer(connection, "ctag"); + + channel.processAsync(new AMQCommand(new AMQImpl.Basic.Cancel.Builder().consumerTag("ctag").build())); + + Mockito.verify(trafficListener, Mockito.never()).write( + Mockito.argThat(c -> c.getMethod() instanceof AMQP.Basic.CancelOk)); + } + + @Test + public void serverBasicCancelForUnknownConsumerIsAnsweredWithCancelOkWhenBrokerAcceptsIt() throws Exception { + TrafficListener trafficListener = Mockito.mock(TrafficListener.class); + AMQConnection connection = connectionAcceptingConsumerCancelOk(true, trafficListener); + ChannelN channel = new ChannelN(connection, 1, consumerWorkService); + + channel.processAsync(new AMQCommand(new AMQImpl.Basic.Cancel.Builder().consumerTag("does-not-exist").build())); + + Mockito.verify(trafficListener, Mockito.times(1)).write( + Mockito.argThat(c -> c.getMethod() instanceof AMQP.Basic.CancelOk)); + } + + @Test + public void serverBasicCancelForUnknownConsumerIsNotAnsweredWhenBrokerDoesNotAcceptCancelOk() throws Exception { + TrafficListener trafficListener = Mockito.mock(TrafficListener.class); + AMQConnection connection = connectionAcceptingConsumerCancelOk(false, trafficListener); + ChannelN channel = new ChannelN(connection, 1, consumerWorkService); + + channel.processAsync(new AMQCommand(new AMQImpl.Basic.Cancel.Builder().consumerTag("does-not-exist").build())); + + Mockito.verify(trafficListener, Mockito.never()).write(Mockito.any(Command.class)); + } + @Test public void callingBasicCancelForUnknownConsumerDoesNotThrowException() throws Exception { AMQConnection connection = Mockito.mock(AMQConnection.class); @@ -107,6 +156,30 @@ public void confirmSelectOnlySendsRPCCallOnce() throws Exception { Mockito.verify(trafficListener, Mockito.times(1)).write(Mockito.any(Command.class)); } + private AMQConnection connectionAcceptingConsumerCancelOk( + boolean accepts, TrafficListener trafficListener) { + AMQConnection connection = Mockito.mock(AMQConnection.class); + Mockito.when(connection.getTrafficListener()).thenReturn(trafficListener); + Mockito.when(connection.doesBrokerAcceptClientSentBasicCancelOk()).thenReturn(accepts); + return connection; + } + + private ChannelN channelWithConsumer(AMQConnection connection, String consumerTag) throws Exception { + ChannelN channel = new ChannelN(connection, 1, consumerWorkService); + // the RPC is enqueued before the request is written, so replying from the write is race-free + TrafficListener trafficListener = connection.getTrafficListener(); + Mockito.doAnswer(invocation -> { + Command command = invocation.getArgument(0); + if (command.getMethod() instanceof AMQP.Basic.Consume) { + channel.handleCompleteInboundCommand( + new AMQCommand(new AMQImpl.Basic.ConsumeOk(consumerTag))); + } + return null; + }).when(trafficListener).write(Mockito.any(Command.class)); + channel.basicConsume("q", false, consumerTag, new DefaultConsumer(channel)); + return channel; + } + interface Consumer { void apply(int value) throws Exception;