QPID-6207 : [Java Broker] Flow uncommitted messages to disk if combined size greater than threshold

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1636617 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 46 insertions, 5 deletions

diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/logging/messages/ChannelMessages.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/logging/messages/ChannelMessages.java
index 6ae1ac4f02..c112dd4292 100644
--- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/logging/messages/ChannelMessages.java
+++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/logging/messages/ChannelMessages.java
@@ -22,15 +22,14 @@ package org.apache.qpid.server.logging.messages;
 
 import static org.apache.qpid.server.logging.AbstractMessageLogger.DEFAULT_LOG_HIERARCHY_PREFIX;
 
-import java.text.MessageFormat;
-import java.util.Locale;
-import java.util.ResourceBundle;
-
 import org.apache.log4j.Logger;
-
 import org.apache.qpid.server.configuration.BrokerProperties;
 import org.apache.qpid.server.logging.LogMessage;
 
+import java.text.MessageFormat;
+import java.util.Locale;
+import java.util.ResourceBundle;
+
 /**
  * DO NOT EDIT DIRECTLY, THIS FILE WAS GENERATED.
  *
@@ -51,6 +50,7 @@ public class ChannelMessages
     public static final String CLOSE_LOG_HIERARCHY = DEFAULT_LOG_HIERARCHY_PREFIX + "channel.close";
     public static final String PREFETCH_SIZE_LOG_HIERARCHY = DEFAULT_LOG_HIERARCHY_PREFIX + "channel.prefetch_size";
     public static final String CLOSE_FORCED_LOG_HIERARCHY = DEFAULT_LOG_HIERARCHY_PREFIX + "channel.close_forced";
+    public static final String LARGE_TRANSACTION_WARN_LOG_HIERARCHY = DEFAULT_LOG_HIERARCHY_PREFIX + "channel.large_transaction_warn";
     public static final String DEADLETTERMSG_LOG_HIERARCHY = DEFAULT_LOG_HIERARCHY_PREFIX + "channel.deadlettermsg";
     public static final String DISCARDMSG_NOALTEXCH_LOG_HIERARCHY = DEFAULT_LOG_HIERARCHY_PREFIX + "channel.discardmsg_noaltexch";
     public static final String IDLE_TXN_LOG_HIERARCHY = DEFAULT_LOG_HIERARCHY_PREFIX + "channel.idle_txn";
@@ -68,6 +68,7 @@ public class ChannelMessages
         Logger.getLogger(CLOSE_LOG_HIERARCHY);
         Logger.getLogger(PREFETCH_SIZE_LOG_HIERARCHY);
         Logger.getLogger(CLOSE_FORCED_LOG_HIERARCHY);
+        Logger.getLogger(LARGE_TRANSACTION_WARN_LOG_HIERARCHY);
         Logger.getLogger(DEADLETTERMSG_LOG_HIERARCHY);
         Logger.getLogger(DISCARDMSG_NOALTEXCH_LOG_HIERARCHY);
         Logger.getLogger(IDLE_TXN_LOG_HIERARCHY);
@@ -263,6 +264,38 @@ public class ChannelMessages
 
     /**
      * Log a Channel message of the Format:
+     * <pre>CHN-1013 : Uncommitted transaction contains {0,number} bytes of incoming message data.</pre>
+     * Optional values are contained in [square brackets] and are numbered
+     * sequentially in the method call.
+     *
+     */
+    public static LogMessage LARGE_TRANSACTION_WARN(Number param1)
+    {
+        String rawMessage = _messages.getString("LARGE_TRANSACTION_WARN");
+
+        final Object[] messageArguments = {param1};
+        // Create a new MessageFormat to ensure thread safety.
+        // Sharing a MessageFormat and using applyPattern is not thread safe
+        MessageFormat formatter = new MessageFormat(rawMessage, _currentLocale);
+
+        final String message = formatter.format(messageArguments);
+
+        return new LogMessage()
+        {
+            public String toString()
+            {
+                return message;
+            }
+
+            public String getLogHierarchy()
+            {
+                return LARGE_TRANSACTION_WARN_LOG_HIERARCHY;
+            }
+        };
+    }
+
+    /**
+     * Log a Channel message of the Format:
      * <pre>CHN-1011 : Message : {0,number} moved to dead letter queue : {1}</pre>
      * Optional values are contained in [square brackets] and are numbered
      * sequentially in the method call.
diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/logging/messages/Channel_logmessages.properties b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/logging/messages/Channel_logmessages.properties
index 5c6e066541..4ae8e39d25 100644
--- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/logging/messages/Channel_logmessages.properties
+++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/logging/messages/Channel_logmessages.properties
@@ -40,3 +40,6 @@ DISCARDMSG_NOROUTE = CHN-1010 : Discarded message : {0,number} as no binding on
 DEADLETTERMSG = CHN-1011 : Message : {0,number} moved to dead letter queue : {1}
 
 FLOW_CONTROL_IGNORED = CHN-1012 : Flow Control Ignored. Channel will be closed.
+
+LARGE_TRANSACTION_WARN = CHN-1013 : Uncommitted transaction contains {0,number} bytes of incoming message data.
+
diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/Connection.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/Connection.java
index 5b3965904e..b28441438d 100644
--- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/Connection.java
+++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/Connection.java
@@ -44,6 +44,11 @@ public interface Connection<X extends Connection<X>> extends ConfiguredObject<X>
     String PORT = "port";
 
 
+    String MAX_UNCOMMITTED_IN_MEMORY_SIZE = "connection.maxUncommittedInMemorySize";
+
+    @ManagedContextDefault(name = MAX_UNCOMMITTED_IN_MEMORY_SIZE)
+    long DEFAULT_MAX_UNCOMMITTED_IN_MEMORY_SIZE = 10l * 1024l * 1024l;
+
     @DerivedAttribute
     String getClientId();