Validate subscription key length in KahaDB

activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java

@@ -103,6 +103,7 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter,
     public static final String PROPERTY_ASYNC_EXECUTOR_MAX_THREADS = "org.apache.activemq.store.kahadb.ASYNC_EXECUTOR_MAX_THREADS";
     private static final int asyncExecutorMaxThreads = Integer.parseInt(System.getProperty(
             PROPERTY_ASYNC_EXECUTOR_MAX_THREADS, "1"), 10);
+    static final int MAX_SUBSCRIPTION_KEY_LENGTH = 65535;
 
     protected ExecutorService queueExecutor;
     protected ExecutorService topicExecutor;
@@ -1403,7 +1404,11 @@ public MessageStoreSubscriptionStatistics getMessageStoreSubStatistics() {
     }
 
     String subscriptionKey(String clientId, String subscriptionName) {
-        return clientId + ":" + subscriptionName;
+        String key = clientId + ":" + subscriptionName;
+        if (key.length() > MAX_SUBSCRIPTION_KEY_LENGTH) {
+            throw new IllegalArgumentException("Subscription key exceeds max length " + MAX_SUBSCRIPTION_KEY_LENGTH);
+        }
+        return key;
     }
 
     @Override

activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBStoreTest.java

@@ -16,6 +16,7 @@
  */
 package org.apache.activemq.store.kahadb;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Vector;
 import java.util.concurrent.ExecutorService;
@@ -24,16 +25,20 @@
 import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.command.ActiveMQMessage;
 import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.command.ActiveMQTopic;
 import org.apache.activemq.command.Message;
 import org.apache.activemq.command.MessageAck;
 import org.apache.activemq.command.MessageId;
 import org.apache.activemq.command.ProducerId;
+import org.apache.activemq.command.SubscriptionInfo;
+import org.apache.activemq.store.TopicMessageStore;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
 public class KahaDBStoreTest {
@@ -139,4 +144,31 @@ public void testTrackAndForgetRecoveredAcks() throws Exception {
         underTest.forgetRecoveredAcks(new ArrayList<>(acks.subList(5, 10)), false);
         assertTrue(underTest.ackedAndPreparedMap.isEmpty());
     }
+
+    @Test
+    public void testSubscriptionKeyValidation() {
+        assertEquals("client1:sub1", store.subscriptionKey("client1", "sub1"));
+        Throwable error = assertThrows(IllegalArgumentException.class,
+                () -> store.subscriptionKey("client1", "a".repeat(KahaDBStore.MAX_SUBSCRIPTION_KEY_LENGTH)));
+        assertTrue(error.getMessage().contains("Subscription key exceeds max length"));
+        error = assertThrows(IllegalArgumentException.class,
+                () -> store.subscriptionKey("a".repeat(KahaDBStore.MAX_SUBSCRIPTION_KEY_LENGTH), "subname"));
+        assertTrue(error.getMessage().contains("Subscription key exceeds max length"));
+    }
+
+    @Test
+    public void testSubscriptionCommandValidation() throws IOException {
+        TopicMessageStore topicStore = store.createTopicMessageStore(new ActiveMQTopic("test"));
+
+        // check addSubscription, which creates KahaSubscriptionCommand, validates
+        Throwable error = assertThrows(IllegalArgumentException.class,
+                () -> topicStore.addSubscription(new SubscriptionInfo("client1",
+                        "a".repeat(KahaDBStore.MAX_SUBSCRIPTION_KEY_LENGTH)), false));
+        assertTrue(error.getMessage().contains("Subscription key exceeds max length"));
+
+        // check deleteSubscription, which also creates KahaSubscriptionCommand, validates
+        error = assertThrows(IllegalArgumentException.class, () -> topicStore.deleteSubscription(
+                "client1", "a".repeat(KahaDBStore.MAX_SUBSCRIPTION_KEY_LENGTH)));
+        assertTrue(error.getMessage().contains("Subscription key exceeds max length"));
+    }
 }

activemq-unit-tests/src/test/java/org/apache/activemq/usecases/DurableSubscriptionTestSupport.java

@@ -40,9 +40,9 @@
 public abstract class DurableSubscriptionTestSupport extends TestSupport {
 
     protected Connection connection;
-    private Session session;
-    private TopicSubscriber consumer;
-    private MessageProducer producer;
+    protected Session session;
+    protected TopicSubscriber consumer;
+    protected MessageProducer producer;
     protected BrokerService broker;
 
     @Override

activemq-unit-tests/src/test/java/org/apache/activemq/usecases/KahaDBDurableSubscriptionTest.java

@@ -16,6 +16,11 @@
  */
 package org.apache.activemq.usecases;
 
+import static org.junit.Assert.assertThrows;
+
+import jakarta.jms.DeliveryMode;
+import jakarta.jms.JMSException;
+import jakarta.jms.Topic;
 import java.io.IOException;
 import org.apache.activemq.store.PersistenceAdapter;
 import org.junit.experimental.categories.Category;
@@ -27,4 +32,17 @@ public class KahaDBDurableSubscriptionTest extends DurableSubscriptionTestSuppor
     protected PersistenceAdapter createPersistenceAdapter() throws IOException {
         return null; // use default
     }
+
+    public void testDurableSubscriptionKeyValidation() throws Exception {
+        connection.start();
+        session = connection.createSession(false, jakarta.jms.Session.AUTO_ACKNOWLEDGE);
+
+        Topic topic = session.createTopic("TestTopic");
+        // This should throw an exception back to the client because the sub key
+        // value is too large and validation should catch it
+        assertThrows(JMSException.class, () -> session.createDurableSubscriber(topic, "s".repeat(65535)));
+
+        // Ensure we can restart the broker without error
+        restartBroker();
+    }
 }