ARTEMIS-5573 and ARTEMIS-5975 Improve AMQP Size estimation

artemis-cli/src/main/java/org/apache/activemq/artemis/util/ServerUtil.java

@@ -24,6 +24,7 @@
 import java.io.InputStreamReader;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.function.Consumer;
 
 import org.apache.activemq.artemis.api.core.TransportConfiguration;
 import org.apache.activemq.artemis.api.core.client.ClientSession;
@@ -51,7 +52,11 @@ public static Process startServer(String artemisInstance, String serverName, int
    }
 
    public static Process startServer(String artemisInstance, String serverName, int id, int timeout, File brokerProperties) throws Exception {
-      final Process process = internalStartServer(artemisInstance, serverName, brokerProperties);
+      return startServer(artemisInstance, serverName, id, timeout, brokerProperties, null);
+   }
+
+   public static Process startServer(String artemisInstance, String serverName, int id, int timeout, File brokerProperties, Consumer<String> logCallback) throws Exception {
+      final Process process = internalStartServer(artemisInstance, serverName, brokerProperties, logCallback);
 
       // wait for start
       if (timeout > 0) {
@@ -66,7 +71,11 @@ public static Process startServer(String artemisInstance, String serverName, Str
    }
 
    public static Process startServer(String artemisInstance, String serverName, String uri, int timeout, File propertiesFile) throws Exception {
-      final Process process = internalStartServer(artemisInstance, serverName, propertiesFile);
+      return startServer(artemisInstance, serverName, uri, timeout, propertiesFile, null);
+   }
+
+   public static Process startServer(String artemisInstance, String serverName, String uri, int timeout, File propertiesFile, Consumer<String> logCallback) throws Exception {
+      final Process process = internalStartServer(artemisInstance, serverName, propertiesFile, logCallback);
 
       // wait for start
       if (timeout != 0) {
@@ -78,20 +87,30 @@ public static Process startServer(String artemisInstance, String serverName, Str
 
    private static Process internalStartServer(String artemisInstance,
                                               String serverName) throws IOException, ClassNotFoundException {
-      return internalStartServer(artemisInstance, serverName, null);
+      return internalStartServer(artemisInstance, serverName, null, null);
    }
    private static Process internalStartServer(String artemisInstance,
                                               String serverName,
                                               File propertiesFile) throws IOException, ClassNotFoundException {
+      return internalStartServer(artemisInstance, serverName, propertiesFile, null);
+   }
+   private static Process internalStartServer(String artemisInstance,
+                                              String serverName,
+                                              File propertiesFile,
+                                              Consumer<String> logCallback) throws IOException, ClassNotFoundException {
 
       if (propertiesFile != null) {
-         return execute(artemisInstance, serverName, "run", "--properties", propertiesFile.getAbsolutePath());
+         return execute(artemisInstance, serverName, logCallback, "run", "--properties", propertiesFile.getAbsolutePath());
       } else {
-         return execute(artemisInstance, serverName, "run");
+         return execute(artemisInstance, serverName, logCallback, "run");
       }
    }
 
    public static Process execute(String artemisInstance, String jobName, String...args) throws IOException, ClassNotFoundException {
+      return execute(artemisInstance, jobName, null, args);
+   }
+
+   public static Process execute(String artemisInstance, String jobName, Consumer<String> logCallback, String...args) throws IOException, ClassNotFoundException {
       try {
          boolean IS_WINDOWS = System.getProperty("os.name").toLowerCase().trim().startsWith("win");
 
@@ -117,11 +136,11 @@ public static Process execute(String artemisInstance, String jobName, String...a
          final Process process = builder.start();
          Runtime.getRuntime().addShutdownHook(new Thread(() -> process.destroy()));
 
-         ProcessLogger outputLogger = new ProcessLogger(true, process.getInputStream(), jobName, false);
+         ProcessLogger outputLogger = new ProcessLogger(logCallback == null, process.getInputStream(), jobName, false, logCallback);
          outputLogger.start();
 
          // Adding a reader to System.err, so the VM won't hang on a System.err.println
-         ProcessLogger errorLogger = new ProcessLogger(true, process.getErrorStream(), jobName, true);
+         ProcessLogger errorLogger = new ProcessLogger(logCallback == null, process.getErrorStream(), jobName, true, logCallback);
          errorLogger.start();
          return process;
       } catch (IOException e) {
@@ -215,14 +234,18 @@ static class ProcessLogger extends Thread {
 
       private final boolean sendToErr;
 
+      private final Consumer<String> logCallback;
+
       ProcessLogger(final boolean print,
                     final InputStream is,
                     final String logName,
-                    final boolean sendToErr) throws ClassNotFoundException {
+                    final boolean sendToErr,
+                    final Consumer<String> logCallback) throws ClassNotFoundException {
          this.is = is;
          this.print = print;
          this.logName = logName;
          this.sendToErr = sendToErr;
+         this.logCallback = logCallback;
          setDaemon(false);
       }
 
@@ -240,6 +263,9 @@ public void run() {
                      System.out.println(logName + "-out:" + line);
                   }
                }
+               if (logCallback != null) {
+                  logCallback.accept((sendToErr ? logName + "-err:" : logName + "-out:") + line);
+               }
             }
          } catch (IOException e) {
             // ok, stream closed

artemis-commons/src/main/java/org/apache/activemq/artemis/core/persistence/PersisterIDs.java

@@ -23,7 +23,7 @@
  */
 public class PersisterIDs {
 
-   public static final int MAX_PERSISTERS = 5;
+   public static final int MAX_PERSISTERS = 7;
 
    public static final byte CoreLargeMessagePersister_ID = (byte)0;
 
@@ -37,4 +37,8 @@ public class PersisterIDs {
 
    public static final byte AMQPMessagePersisterV3_ID = (byte)5;
 
+   public static final byte AMQPMessagePersisterV4_ID = (byte)6;
+
+   public static final byte AMQPLargeMessagePersisterV2_ID = (byte)7;
+
 }

artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/Message.java

@@ -876,14 +876,6 @@ default CompositeData toCompositeData(int fieldsLimit, int deliveryCount) throws
 
    int getMemoryEstimate();
 
-   /**
-    * The first estimate that's been calculated without any updates.
-    */
-   default int getOriginalEstimate() {
-      // For Core Protocol we always use the same estimate
-      return getMemoryEstimate();
-   }
-
    /**
     * This is the size of the message when persisted on disk which is used for metrics tracking Note that even if the
     * message itself is not persisted on disk (ie non-durable) this value is still used for metrics tracking If a normal

artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/SimpleManagement.java

@@ -109,6 +109,18 @@ public void rebuildPageCounters() throws Exception {
       simpleManagementVoid("broker", "rebuildPageCounters");
    }
 
+   public long getAddressSize(String address) throws Exception {
+      return simpleManagementLong(ResourceNames.ADDRESS + address, "getAddressSize");
+   }
+
+   public long getMessageCountOnAddress(String address) throws Exception {
+      return simpleManagementLong(ResourceNames.ADDRESS + address, "getMessageCount");
+   }
+
+   public int removeMessagesOnQueue(String queue, String filter) throws Exception {
+      return simpleManagementInt(ResourceNames.QUEUE + queue, "removeMessages", filter);
+   }
+
    /**
     * Simple helper for management returning a string.
     */

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessage.java

@@ -40,6 +40,7 @@
 import org.apache.activemq.artemis.protocol.amqp.util.NettyReadable;
 import org.apache.activemq.artemis.protocol.amqp.util.NettyWritable;
 import org.apache.activemq.artemis.protocol.amqp.util.TLSEncode;
+import org.apache.activemq.artemis.utils.DataConstants;
 import org.apache.activemq.artemis.utils.collections.TypedProperties;
 import org.apache.qpid.proton.amqp.messaging.ApplicationProperties;
 import org.apache.qpid.proton.amqp.messaging.DeliveryAnnotations;
@@ -80,6 +81,8 @@ public Message getMessage() {
 
    private boolean reencoded = false;
 
+   private int applicationPropertiesSize;
+
    /**
     * AMQPLargeMessagePersister will save the buffer here.
     */
@@ -264,7 +267,9 @@ protected void readSavedEncoding(ByteBuf buf) {
          applicationPropertiesPosition = buf.readInt();
          remainingBodyPosition = buf.readInt();
 
+         int applicationPropertiesInitialPosition = buf.readerIndex();
          applicationProperties = (ApplicationProperties)TLSEncode.getDecoder().readObject();
+         this.applicationPropertiesSize = buf.readerIndex() - applicationPropertiesInitialPosition;
 
          if (properties != null && properties.getAbsoluteExpiryTime() != null && properties.getAbsoluteExpiryTime().getTime() > 0) {
             if (!expirationReload) {
@@ -400,6 +405,12 @@ protected void parseLargeMessage(byte[] data, boolean initialHeader) {
       }
    }
 
+   @Override
+   protected synchronized void resetMessageData() {
+      super.resetMessageData();
+      applicationPropertiesSize = 0;
+   }
+
    private void genericParseLargeMessage() {
       try {
          parsingBuffer.position(0);
@@ -412,6 +423,16 @@ private void genericParseLargeMessage() {
       }
    }
 
+   @Override
+   protected ApplicationProperties readApplicationProperties(ReadableBuffer data, int position) {
+      ApplicationProperties localAP = super.readApplicationProperties(data, position);
+      if (localAP != null) {
+         this.applicationPropertiesSize = data.position() - position;
+         this.applicationPropertiesCount = localAP.getValue().size();
+      }
+      return localAP;
+   }
+
    protected void parseLargeMessage(ReadableBuffer data) {
       MessageDataScanningStatus status = getDataScanningStatus();
       if (status == MessageDataScanningStatus.NOT_SCANNED) {
@@ -596,16 +617,16 @@ public long getWholeMessageSize() {
          return largeBody.getBodySize();
       } catch (Exception e) {
          logger.warn(e.getMessage());
-         return -1;
+         return VALUE_NOT_PRESENT;
       }
    }
 
 
    @Override
    public synchronized int getMemoryEstimate() {
-      if (memoryEstimate == -1) {
-         memoryEstimate = memoryOffset * 2 + (extraProperties != null ? extraProperties.getEncodeSize() : 0);
-         originalEstimate = memoryEstimate;
+      if (memoryEstimate == VALUE_NOT_PRESENT) {
+         // This estimation was tested and validated through AMQPGlobalMaxTest on soak-tests
+         memoryEstimate = MINIMUM_ESTIMATE + (extraProperties != null ? extraProperties.getEncodeSize() : 0) + applicationPropertiesSize * 2 + applicationPropertiesCount * DataConstants.SIZE_INT;
       }
       return memoryEstimate;
    }
@@ -637,7 +658,7 @@ public long getPersistentSize() {
 
    @Override
    public Persister<Message> getPersister() {
-      return AMQPLargeMessagePersister.getInstance();
+      return AMQPLargeMessagePersisterV2.getInstance();
    }
 
    @Override

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessagePersisterV2.java

@@ -0,0 +1,100 @@
+/*
+ * 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.protocol.amqp.broker;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.core.persistence.CoreMessageObjectPools;
+import org.apache.activemq.artemis.utils.DataConstants;
+
+import static org.apache.activemq.artemis.core.persistence.PersisterIDs.AMQPLargeMessagePersisterV2_ID;
+
+public class AMQPLargeMessagePersisterV2 extends AMQPLargeMessagePersister {
+
+   public static final byte ID = AMQPLargeMessagePersisterV2_ID;
+
+   public static AMQPLargeMessagePersisterV2 theInstance;
+
+   public static AMQPLargeMessagePersisterV2 getInstance() {
+      if (theInstance == null) {
+         theInstance = new AMQPLargeMessagePersisterV2();
+      }
+      return theInstance;
+   }
+
+   @Override
+   public byte getID() {
+      return ID;
+   }
+
+   public AMQPLargeMessagePersisterV2() {
+      super();
+   }
+
+
+   protected static final int PERSISTER_SIZE = DataConstants.SIZE_INT + // meemory estimate
+                                             DataConstants.SIZE_BYTE + // message priority
+                                             DataConstants.SIZE_BOOLEAN; // durable
+
+   @Override
+   public int getEncodeSize(Message record) {
+      return super.getEncodeSize(record) +
+         DataConstants.SIZE_INT + // size delimiter for future use to keep compatibility better
+         PERSISTER_SIZE;
+   }
+
+   /**
+    * Sub classes must add the first short as the protocol-id
+    */
+   @Override
+   public void encode(ActiveMQBuffer buffer, Message record) {
+      super.encode(buffer, record);
+
+      AMQPLargeMessage msgEncode = (AMQPLargeMessage) record;
+      writeSizeDelimiter(buffer);
+      buffer.writeInt(msgEncode.getMemoryEstimate());
+      buffer.writeByte(msgEncode.getPriority());
+      buffer.writeBoolean(msgEncode.isDurable());
+   }
+
+   protected void writeSizeDelimiter(ActiveMQBuffer buffer) {
+      buffer.writeInt(PERSISTER_SIZE); // how many bytes this persister is using
+   }
+
+   @Override
+   public Message decode(ActiveMQBuffer buffer, Message record, CoreMessageObjectPools pools) {
+      AMQPLargeMessage message = (AMQPLargeMessage) super.decode(buffer, record, pools);
+
+
+      int sizePersister = buffer.readInt();
+      int lastPosition = buffer.readerIndex() + sizePersister;
+
+      {
+         message.setMemoryEstimate(buffer.readInt());
+         message.setPriority(buffer.readByte());
+         message.directSetDurable(buffer.readBoolean());
+
+         assert buffer.readerIndex() <= lastPosition;
+      }
+
+      // if a future version of this persister wrote more bytes than what we expected now, this will make sure we skip them.
+      buffer.readerIndex(lastPosition);
+
+      return message;
+   }
+
+}