Respond with basic.cancel-ok to the servers that support it

src/main/java/com/rabbitmq/client/impl/AMQConnection.java

@@ -215,6 +215,22 @@ public Map<String, Object> 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<String, Object> serverProperties,
+                                            ServerCapability capability) {
+        Map<String, Object> capabilities = (Map<String, Object>)
+            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);
     }

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.

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 <code>capabilities</code> table
+ * of the <code>connection.start</code> 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;
+    }
+}

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.<String, Object>emptyMap(), ServerCapability.ACCEPT_CONSUMER_CANCEL_OK))
+            .isFalse();
+    }
+}

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;