ARTEMIS-6189 Adding describe for MQTT Correlation

artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolManagerFactory.java

@@ -21,7 +21,9 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
 import org.apache.activemq.artemis.api.core.BaseInterceptor;
+import org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds;
 import org.apache.activemq.artemis.core.server.ActiveMQComponent;
 import org.apache.activemq.artemis.core.server.ActiveMQScheduledComponent;
 import org.apache.activemq.artemis.core.server.ActiveMQServer;
@@ -71,6 +73,16 @@ public void loadProtocolServices(ActiveMQServer server, List<ActiveMQComponent>
       services.add(new MQTTPeriodicTasks(server, server.getScheduledPool()));
    }
 
+   @Override
+   public Object describeJournalRecord(byte recordType, ActiveMQBuffer buffer) {
+      switch (recordType) {
+         case JournalRecordIds.MQTT_PACKET_ID_CORRELATION:
+            return PacketIdCorrelationKey.getPersister().decode(buffer, null, null);
+         default:
+            return null;
+      }
+   }
+
    public class MQTTPeriodicTasks extends ActiveMQScheduledComponent {
       final ActiveMQServer server;
       public MQTTPeriodicTasks(ActiveMQServer server, ScheduledExecutorService scheduledExecutorService) {

artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/PacketIdCorrelationKey.java

@@ -83,7 +83,7 @@ public String toString() {
       return "PacketIdCorrelation[" + "coreMessageId=" + coreMessageId + ", address=" + address + "]";
    }
 
-   private static class Persister extends AbstractHashMapPersister<String, PacketIdCorrelationKey, Integer> {
+   public static class Persister extends AbstractHashMapPersister<String, PacketIdCorrelationKey, Integer> {
       @Override
       protected int getCollectionIdSize(String collectionID) {
          return BufferHelper.sizeOfString(collectionID);

artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/DescribeJournal.java

@@ -69,6 +69,7 @@
 import org.apache.activemq.artemis.core.persistence.impl.journal.codec.ScheduledDeliveryEncoding;
 import org.apache.activemq.artemis.core.server.LargeServerMessage;
 import org.apache.activemq.artemis.spi.core.protocol.MessagePersister;
+import org.apache.activemq.artemis.spi.core.protocol.ProtocolManagerFactory;
 import org.apache.activemq.artemis.utils.Base64;
 import org.apache.activemq.artemis.utils.TableOut;
 import org.apache.activemq.artemis.utils.XMLUtil;
@@ -732,6 +733,17 @@ public static Object newObjectEncoding(RecordInfo info) {
       return newObjectEncoding(info, null);
    }
 
+   private static final List<ProtocolManagerFactory> protocolManagerFactories;
+
+   static {
+      List<ProtocolManagerFactory> factories = new LinkedList<>();
+      for (ProtocolManagerFactory factory : java.util.ServiceLoader.load(ProtocolManagerFactory.class, DescribeJournal.class.getClassLoader())) {
+         factories.add(factory);
+         logger.info("Loading factory {}", factory);
+      }
+      protocolManagerFactories = factories;
+   }
+
    public static Object newObjectEncoding(RecordInfo info, JournalStorageManager storageManager) {
       ActiveMQBuffer buffer = ActiveMQBuffers.wrappedBuffer(info.data);
       long id = info.id;
@@ -900,6 +912,12 @@ public static Object newObjectEncoding(RecordInfo info, JournalStorageManager st
             return AckRetry.getPersister().decode(buffer, null, null);
 
          default:
+            for (ProtocolManagerFactory factory : protocolManagerFactories) {
+               Object result = factory.describeJournalRecord(info.getUserRecordType(), buffer);
+               if (result != null) {
+                  return result;
+               }
+            }
             return null;
       }
    }
@@ -908,51 +926,9 @@ static String describeRecordType(byte recordType) {
       if (recordType == (byte)0) {
          return "";
       } else {
-         return recordType + " (" + recordTypeName(recordType) + ")";
+         return recordType + " (" + JournalRecordIds.recordTypeName(recordType) + ")";
       }
    }
-
-   private static String recordTypeName(byte recordType) {
-
-      switch (recordType) {
-         case JournalRecordIds.GROUP_RECORD: return "GROUP";
-         case JournalRecordIds.QUEUE_BINDING_RECORD: return "QUEUE_BINDING";
-         case JournalRecordIds.QUEUE_STATUS_RECORD: return "QUEUE_STATUS";
-         case JournalRecordIds.ID_COUNTER_RECORD: return "ID_COUNTER";
-         case JournalRecordIds.ADDRESS_SETTING_RECORD: return "ADDRESS_SETTING";
-         case JournalRecordIds.SECURITY_SETTING_RECORD: return "SECURITY_SETTING";
-         case JournalRecordIds.DIVERT_RECORD: return "DIVERT";
-         case JournalRecordIds.BRIDGE_RECORD: return "BRIDGE";
-         case JournalRecordIds.ADD_LARGE_MESSAGE_PENDING: return "ADD_LARGE_MESSAGE_PENDING";
-         case JournalRecordIds.ADD_LARGE_MESSAGE: return "ADD_LARGE_MESSAGE";
-         case JournalRecordIds.ADD_MESSAGE: return "ADD_MESSAGE";
-         case JournalRecordIds.ADD_REF: return "ADD_REF";
-         case JournalRecordIds.ACKNOWLEDGE_REF: return "ACKNOWLEDGE_REF";
-         case JournalRecordIds.UPDATE_DELIVERY_COUNT: return "UPDATE_DELIVERY_COUNT";
-         case JournalRecordIds.PAGE_TRANSACTION: return "PAGE_TRANSACTION";
-         case JournalRecordIds.SET_SCHEDULED_DELIVERY_TIME: return "SET_SCHEDULED_DELIVERY_TIME";
-         case JournalRecordIds.DUPLICATE_ID: return "DUPLICATE_ID";
-         case JournalRecordIds.HEURISTIC_COMPLETION: return "HEURISTIC_COMPLETION";
-         case JournalRecordIds.ACKNOWLEDGE_CURSOR: return "ACKNOWLEDGE_CURSOR";
-         case JournalRecordIds.PAGE_CURSOR_COUNTER_VALUE: return "PAGE_CURSOR_COUNTER_VALUE";
-         case JournalRecordIds.PAGE_CURSOR_COUNTER_INC: return "PAGE_CURSOR_COUNTER_INC";
-         case JournalRecordIds.PAGE_CURSOR_COMPLETE: return "PAGE_CURSOR_COMPLETE";
-         case JournalRecordIds.PAGE_CURSOR_PENDING_COUNTER: return "PAGE_CURSOR_PENDING_COUNTER";
-         case JournalRecordIds.ADDRESS_BINDING_RECORD: return "ADDRESS_BINDING";
-         case JournalRecordIds.ADD_MESSAGE_PROTOCOL: return "ADD_MESSAGE_PROTOCOL";
-         case JournalRecordIds.ADDRESS_STATUS_RECORD: return "ADDRESS_STATUS";
-         case JournalRecordIds.USER_RECORD: return "USER";
-         case JournalRecordIds.ROLE_RECORD: return "ROLE";
-         case JournalRecordIds.ADD_MESSAGE_BODY: return "ADD_MESSAGE_BODY";
-         case JournalRecordIds.KEY_VALUE_PAIR_RECORD: return "KEY_VALUE_PAIR";
-         case JournalRecordIds.CONNECTOR_RECORD: return "CONNECTOR";
-         case JournalRecordIds.ADDRESS_SETTING_RECORD_JSON: return "ADDRESS_SETTING_JSON";
-         case JournalRecordIds.ACK_RETRY: return "ACK_RETRY";
-         case JournalRecordIds.MQTT_PACKET_ID_CORRELATION: return "MQTT_PACKET_ID_CORRELATION";
-         default: return "UNKNOWN";
-      }
-   }
-
    private static final class PageCompleteCursorAckRecordEncoding extends CursorAckRecordEncoding {
 
       @Override

artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalRecordIds.java

@@ -107,4 +107,46 @@ public final class JournalRecordIds {
    public static final byte ACK_RETRY = 53;
 
    public static final byte MQTT_PACKET_ID_CORRELATION = 54;
+
+   public static String recordTypeName(byte recordType) {
+
+      switch (recordType) {
+         case JournalRecordIds.GROUP_RECORD: return "GROUP";
+         case JournalRecordIds.QUEUE_BINDING_RECORD: return "QUEUE_BINDING";
+         case JournalRecordIds.QUEUE_STATUS_RECORD: return "QUEUE_STATUS";
+         case JournalRecordIds.ID_COUNTER_RECORD: return "ID_COUNTER";
+         case JournalRecordIds.ADDRESS_SETTING_RECORD: return "ADDRESS_SETTING";
+         case JournalRecordIds.SECURITY_SETTING_RECORD: return "SECURITY_SETTING";
+         case JournalRecordIds.DIVERT_RECORD: return "DIVERT";
+         case JournalRecordIds.BRIDGE_RECORD: return "BRIDGE";
+         case JournalRecordIds.ADD_LARGE_MESSAGE_PENDING: return "ADD_LARGE_MESSAGE_PENDING";
+         case JournalRecordIds.ADD_LARGE_MESSAGE: return "ADD_LARGE_MESSAGE";
+         case JournalRecordIds.ADD_MESSAGE: return "ADD_MESSAGE";
+         case JournalRecordIds.ADD_REF: return "ADD_REF";
+         case JournalRecordIds.ACKNOWLEDGE_REF: return "ACKNOWLEDGE_REF";
+         case JournalRecordIds.UPDATE_DELIVERY_COUNT: return "UPDATE_DELIVERY_COUNT";
+         case JournalRecordIds.PAGE_TRANSACTION: return "PAGE_TRANSACTION";
+         case JournalRecordIds.SET_SCHEDULED_DELIVERY_TIME: return "SET_SCHEDULED_DELIVERY_TIME";
+         case JournalRecordIds.DUPLICATE_ID: return "DUPLICATE_ID";
+         case JournalRecordIds.HEURISTIC_COMPLETION: return "HEURISTIC_COMPLETION";
+         case JournalRecordIds.ACKNOWLEDGE_CURSOR: return "ACKNOWLEDGE_CURSOR";
+         case JournalRecordIds.PAGE_CURSOR_COUNTER_VALUE: return "PAGE_CURSOR_COUNTER_VALUE";
+         case JournalRecordIds.PAGE_CURSOR_COUNTER_INC: return "PAGE_CURSOR_COUNTER_INC";
+         case JournalRecordIds.PAGE_CURSOR_COMPLETE: return "PAGE_CURSOR_COMPLETE";
+         case JournalRecordIds.PAGE_CURSOR_PENDING_COUNTER: return "PAGE_CURSOR_PENDING_COUNTER";
+         case JournalRecordIds.ADDRESS_BINDING_RECORD: return "ADDRESS_BINDING";
+         case JournalRecordIds.ADD_MESSAGE_PROTOCOL: return "ADD_MESSAGE_PROTOCOL";
+         case JournalRecordIds.ADDRESS_STATUS_RECORD: return "ADDRESS_STATUS";
+         case JournalRecordIds.USER_RECORD: return "USER";
+         case JournalRecordIds.ROLE_RECORD: return "ROLE";
+         case JournalRecordIds.ADD_MESSAGE_BODY: return "ADD_MESSAGE_BODY";
+         case JournalRecordIds.KEY_VALUE_PAIR_RECORD: return "KEY_VALUE_PAIR";
+         case JournalRecordIds.CONNECTOR_RECORD: return "CONNECTOR";
+         case JournalRecordIds.ADDRESS_SETTING_RECORD_JSON: return "ADDRESS_SETTING_JSON";
+         case JournalRecordIds.ACK_RETRY: return "ACK_RETRY";
+         case JournalRecordIds.MQTT_PACKET_ID_CORRELATION: return "MQTT_CORRELATION";
+         default: return "UNKNOWN";
+      }
+   }
+
 }

artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManagerFactory.java

@@ -19,6 +19,7 @@
 import java.util.List;
 import java.util.Map;
 
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
 import org.apache.activemq.artemis.api.core.BaseInterceptor;
 import org.apache.activemq.artemis.api.core.Message;
 import org.apache.activemq.artemis.core.persistence.Persister;
@@ -63,4 +64,13 @@ ProtocolManager createProtocolManager(ActiveMQServer server,
     */
    void updateProtocolServices(ActiveMQServer server, List<ActiveMQComponent> services) throws Exception;
 
+   /**
+    * Return an object that describes a protocol-specific journal record.
+    * This is used by {@code ./artemis data print} to decode record types that the core journal does not recognize.
+    * Implementations must not read from the buffer if they do not recognize the recordType.
+    */
+   default Object describeJournalRecord(byte recordType, ActiveMQBuffer buffer) {
+      return null;
+   }
+
 }

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/MQTTPrintDataTest.java

@@ -0,0 +1,105 @@
+/*
+ * 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.artemis.tests.integration.mqtt;
+
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+import java.nio.charset.StandardCharsets;
+
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.cli.commands.tools.PrintData;
+import org.apache.activemq.artemis.core.config.FileDeploymentManager;
+import org.apache.activemq.artemis.core.config.impl.FileConfiguration;
+import org.apache.activemq.artemis.core.config.impl.SecurityConfiguration;
+import org.apache.activemq.artemis.core.protocol.mqtt.MQTTStateManager;
+import org.apache.activemq.artemis.core.protocol.mqtt.PacketIdCorrelationKey;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
+import org.apache.activemq.artemis.jms.server.config.impl.FileJMSConfiguration;
+import org.apache.activemq.artemis.spi.core.security.ActiveMQJAASSecurityManager;
+import org.apache.activemq.artemis.spi.core.security.jaas.InVMLoginModule;
+import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
+import org.apache.activemq.artemis.utils.RandomUtil;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class MQTTPrintDataTest extends ActiveMQTestBase {
+
+   @Test
+   @Timeout(30)
+   public void testPrintDataWithMQTTQoS2Correlation() throws Exception {
+      ActiveMQServer server = addServer(getActiveMQServer("dataprint/etc/broker.xml"));
+      try {
+         server.getConfiguration().setPersistenceEnabled(true);
+         server.start();
+         String clientID = RandomUtil.randomUUIDString();
+         SimpleString addressID = RandomUtil.randomUUIDSimpleString();
+         long messageID = 3000L;
+         MQTTStateManager.getInstance(server).putPacketIdCorrelation(clientID, PacketIdCorrelationKey.of(messageID, addressID), 1);
+         server.stop();
+
+         String previousInstance = System.getProperty("artemis.instance");
+         try {
+            System.setProperty("artemis.instance", this.getClass().getClassLoader().getResource("dataprint").getFile());
+
+            ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+            PrintStream printStream = new PrintStream(byteArrayOutputStream, true, StandardCharsets.UTF_8.name());
+            PrintData.printData(server.getConfiguration().getBindingsLocation().getAbsoluteFile(),
+                                server.getConfiguration().getJournalLocation().getAbsoluteFile(),
+                                server.getConfiguration().getPagingLocation().getAbsoluteFile(),
+                                printStream, false, false, false, false, -1, true);
+
+            String output = byteArrayOutputStream.toString();
+            System.out.println(output);
+            assertTrue(output.contains("PacketIdCorrelation"), "Print data output should contain the MQTT PacketIdCorrelation record.\nOutput:\n" + output);
+            assertTrue(output.contains(addressID.toString()), "Print data output should contain the MQTT PacketIdCorrelation record.\nOutput:\n" + output);
+            assertTrue(output.contains(clientID), "Print data output should contain the MQTT PacketIdCorrelation record.\nOutput:\n" + output);
+         } finally {
+            if (previousInstance != null) {
+               System.setProperty("artemis.instance", previousInstance);
+            } else {
+               System.clearProperty("artemis.instance");
+            }
+         }
+      } finally {
+         try {
+            server.stop();
+         } catch (Exception e) {
+         }
+      }
+   }
+
+   protected ActiveMQServer getActiveMQServer(String brokerConfig) throws Exception {
+      FileConfiguration fc = new FileConfiguration();
+      FileJMSConfiguration fileConfiguration = new FileJMSConfiguration();
+      FileDeploymentManager deploymentManager = new FileDeploymentManager(brokerConfig);
+      deploymentManager.addDeployable(fc);
+      deploymentManager.addDeployable(fileConfiguration);
+      deploymentManager.readConfiguration();
+
+      ActiveMQJAASSecurityManager sm = new ActiveMQJAASSecurityManager(InVMLoginModule.class.getName(), new SecurityConfiguration());
+
+      recreateDirectory(fc.getBindingsDirectory());
+      recreateDirectory(fc.getJournalDirectory());
+      recreateDirectory(fc.getPagingDirectory());
+      recreateDirectory(fc.getLargeMessagesDirectory());
+
+      return addServer(new ActiveMQServerImpl(fc, sm));
+   }
+}