ARTEMIS-5573 and ARTEMIS-5975 Improve AMQP Size estimation

Also making it immutable to avoid races after updates like we had in the
past

Author: clebertsuconic

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-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,17 @@ private void genericParseLargeMessage() {
       }
    }
 
+   @Override
+   protected ApplicationProperties readApplicationProperties(ReadableBuffer data, int position) {
+      applicationProperties = super.readApplicationProperties(data, position);
+      if (applicationProperties != null) {
+         this.applicationPropertiesSize = data.position() - position;
+         this.applicationPropertiesCount = applicationProperties.getValue().size();
+      }
+      return applicationProperties;
+   }
+
+
    protected void parseLargeMessage(ReadableBuffer data) {
       MessageDataScanningStatus status = getDataScanningStatus();
       if (status == MessageDataScanningStatus.NOT_SCANNED) {
@@ -604,8 +626,10 @@ public long getWholeMessageSize() {
    @Override
    public synchronized int getMemoryEstimate() {
       if (memoryEstimate == -1) {
-         memoryEstimate = memoryOffset * 2 + (extraProperties != null ? extraProperties.getEncodeSize() : 0);
-         originalEstimate = memoryEstimate;
+         // the applicationPropertiesSize is * 4 to give us some room to estimate the size of the applicationProperties after decoding.
+         // A lot of testing was done in AMQPGlobalMaxTest to determine this factor. It can always be further adjusted
+         // Also it's better to overestimate than to underestimate as that could lead to OutOfMemoryException
+         memoryEstimate = MINIMUM_ESTIMATE + (extraProperties != null ? extraProperties.getEncodeSize() : 0) + applicationPropertiesSize * 2 + applicationPropertiesCount * DataConstants.SIZE_INT;
       }
       return memoryEstimate;
    }