This is an automated email from the ASF dual-hosted git repository.
mattrpav pushed a commit to branch activemq-6.3.x
in repository https://gitbox.apache.org/repos/asf/activemq.git
The following commit(s) were added to refs/heads/activemq-6.3.x by this push:
new cbf6a36d2b [#] Update TopicRegion removeSubscription to use connection
clientId (#2478)
cbf6a36d2b is described below
commit cbf6a36d2b05bf9f95771dc98dfa32d50576d63d
Author: Matt Pavlovich <[email protected]>
AuthorDate: Thu Aug 20 14:45:32 2026 -0500
[#] Update TopicRegion removeSubscription to use connection clientId (#2478)
(cherry picked from commit dc713c16f53a9af0027532cd7911cda7fa7bb773)
---
.../apache/activemq/broker/region/TopicRegion.java | 6 +-
.../activemq/command/RemoveSubscriptionInfo.java | 1 +
.../usecases/RemoveSubscriptionClientIdTest.java | 140 +++++++++++++++++++++
3 files changed, 145 insertions(+), 2 deletions(-)
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/TopicRegion.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/TopicRegion.java
index 7ec37f4f7e..b02783eb6e 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/TopicRegion.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/TopicRegion.java
@@ -209,11 +209,13 @@ public class TopicRegion extends AbstractRegion {
@Override
public void removeSubscription(ConnectionContext context,
RemoveSubscriptionInfo info) throws Exception {
- SubscriptionKey key = new SubscriptionKey(info.getClientId(),
info.getSubscriptionName());
+ // clientId from connection context
+ final String clientId = context.getClientId();
+ SubscriptionKey key = new SubscriptionKey(clientId,
info.getSubscriptionName());
DurableTopicSubscription sub = durableSubscriptions.get(key);
if (sub == null) {
throw new InvalidDestinationException("No durable subscription
exists for clientID: " +
- info.getClientId() + " and
subscriptionName: " +
+ clientId + " and
subscriptionName: " +
info.getSubscriptionName());
}
if (sub.isActive()) {
diff --git
a/activemq-client/src/main/java/org/apache/activemq/command/RemoveSubscriptionInfo.java
b/activemq-client/src/main/java/org/apache/activemq/command/RemoveSubscriptionInfo.java
index d805b788db..c449c070f4 100644
---
a/activemq-client/src/main/java/org/apache/activemq/command/RemoveSubscriptionInfo.java
+++
b/activemq-client/src/main/java/org/apache/activemq/command/RemoveSubscriptionInfo.java
@@ -27,6 +27,7 @@ public class RemoveSubscriptionInfo extends BaseCommand {
public static final byte DATA_STRUCTURE_TYPE =
CommandTypes.REMOVE_SUBSCRIPTION_INFO;
protected ConnectionId connectionId;
+ @Deprecated
protected String clientId;
protected String subscriptionName;
diff --git
a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/RemoveSubscriptionClientIdTest.java
b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/RemoveSubscriptionClientIdTest.java
new file mode 100644
index 0000000000..c4450e25ec
--- /dev/null
+++
b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/RemoveSubscriptionClientIdTest.java
@@ -0,0 +1,140 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.activemq.usecases;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import jakarta.jms.JMSException;
+import jakarta.jms.Session;
+
+import org.apache.activemq.ActiveMQConnection;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.command.ActiveMQTopic;
+import org.apache.activemq.command.RemoveSubscriptionInfo;
+import org.apache.activemq.util.Wait;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.apache.activemq.test.annotations.ParallelTest;
+
+/**
+ * These tests exercise the broker's remove-subscription command handling
directly with a
+ * crafted {@link RemoveSubscriptionInfo} to confirm that (a) a connection may
remove its
+ * own durable subscription and (b) a connection may not remove another
clientID's durable
+ * subscription.
+ */
+@Category(ParallelTest.class)
+public class RemoveSubscriptionClientIdTest {
+
+ private static final String FIRST_CLIENT_ID = "first-client";
+ private static final String SECOND_CLIENT_ID = "second-client";
+ private static final String SUBSCRIPTION_NAME = "durable-sub";
+
+ private BrokerService broker;
+ private String connectionUri;
+ private final ActiveMQTopic topic = new ActiveMQTopic("SomeTopic");
+
+ @Before
+ public void setUp() throws Exception {
+ broker = new BrokerService();
+ broker.setUseJmx(true);
+ broker.setPersistent(false);
+ broker.setDeleteAllMessagesOnStartup(true);
+ connectionUri =
broker.addConnector("tcp://localhost:0").getPublishableConnectString();
+ broker.start();
+ broker.waitUntilStarted();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (broker != null) {
+ broker.stop();
+ broker.waitUntilStopped();
+ }
+ }
+
+ private ActiveMQConnection createConnection(String clientId) throws
Exception {
+ var factory = new ActiveMQConnectionFactory(connectionUri);
+ var connection = (ActiveMQConnection) factory.createConnection();
+ connection.setClientID(clientId);
+ connection.start();
+ return connection;
+ }
+
+ /**
+ * Register a durable subscription for the first clientID, then take it
offline so it
+ * exists as an inactive durable subscription in the broker.
+ */
+ private void createInactiveFirstSubscription() throws Exception {
+ try (var first = createConnection(FIRST_CLIENT_ID);
+ var session = first.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ var consumer = session.createDurableSubscriber(topic,
SUBSCRIPTION_NAME)) {
+ // register the durable subscription; closing the resources below
leaves it inactive
+ }
+
+ assertTrue("first durable subscription should exist and be inactive",
+ Wait.waitFor(() ->
broker.getAdminView().getInactiveDurableTopicSubscribers().length == 1,
+ 5000, 10));
+ }
+
+ /**
+ * A connection removing its own durable subscription must succeed
+ */
+ @Test(timeout = 60 * 1000)
+ public void testRemoveOwnSubscriptionSucceeds() throws Exception {
+ createInactiveFirstSubscription();
+
+ try (var first = createConnection(FIRST_CLIENT_ID)) {
+ var rsi = new RemoveSubscriptionInfo();
+ rsi.setConnectionId(first.getConnectionInfo().getConnectionId());
+ rsi.setClientId(FIRST_CLIENT_ID);
+ rsi.setSubscriptionName(SUBSCRIPTION_NAME);
+ first.syncSendPacket(rsi);
+ }
+
+ assertEquals("owner should be able to remove its own durable
subscription",
+ 0,
broker.getAdminView().getInactiveDurableTopicSubscribers().length);
+ }
+
+ /**
+ * A foreign clientID must not be able to delete another clientID's
durable subscription.
+ */
+ @Test(timeout = 60 * 1000)
+ public void testRemoveSubscriptionForForeignClientIdIsIgnored() throws
Exception {
+ createInactiveFirstSubscription();
+
+ try (var second = createConnection(SECOND_CLIENT_ID)) {
+ var rsi = new RemoveSubscriptionInfo();
+ rsi.setConnectionId(second.getConnectionInfo().getConnectionId());
+ // but names the first subscription clientId
+ rsi.setClientId(FIRST_CLIENT_ID);
+ rsi.setSubscriptionName(SUBSCRIPTION_NAME);
+
+ try {
+ second.syncSendPacket(rsi);
+ } catch (JMSException rejected) {
+ // A hardened broker may reject the mismatched request
outright - acceptable.
+ }
+ }
+
+ assertEquals("a foreign clientID must not be able to remove the
another's durable subscription",
+ 1,
broker.getAdminView().getInactiveDurableTopicSubscribers().length);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
For further information, visit: https://activemq.apache.org/contact
