activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jbert...@apache.org
Subject [3/4] activemq-6 git commit: ACTIVEMQ6-94: Using proper flow control on very large messages over the bridge
Date Thu, 23 Apr 2015 20:03:14 GMT
ACTIVEMQ6-94: Using proper flow control on very large messages over the bridge

This will remove some of the verifications written by Howard on his commit.
I did this to simplify the flow control

This closes #197


Project: http://git-wip-us.apache.org/repos/asf/activemq-6/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-6/commit/ada112a6
Tree: http://git-wip-us.apache.org/repos/asf/activemq-6/tree/ada112a6
Diff: http://git-wip-us.apache.org/repos/asf/activemq-6/diff/ada112a6

Branch: refs/heads/master
Commit: ada112a6a37dce8ddf48e2238904421b2ca8e0dc
Parents: c1111cc
Author: Clebert Suconic <clebertsuconic@apache.org>
Authored: Wed Apr 22 21:44:28 2015 -0400
Committer: Clebert Suconic <clebertsuconic@apache.org>
Committed: Thu Apr 23 15:48:20 2015 -0400

----------------------------------------------------------------------
 .../config/ActiveMQDefaultConfiguration.java    | 14 ++++
 .../activemq/core/protocol/core/Channel.java    | 11 ----
 .../core/impl/ActiveMQSessionContext.java       |  8 ---
 .../core/protocol/core/impl/ChannelImpl.java    | 68 +++++---------------
 .../core/config/BridgeConfiguration.java        | 15 +++++
 .../config/ClusterConnectionConfiguration.java  | 14 ++++
 .../deployers/impl/FileConfigurationParser.java | 12 +++-
 .../core/server/cluster/ClusterManager.java     |  5 +-
 .../cluster/impl/ClusterConnectionImpl.java     | 11 +++-
 .../resources/schema/activemq-configuration.xsd | 20 +++++-
 .../core/config/impl/FileConfigurationTest.java |  4 ++
 .../resources/ConfigurationTest-full-config.xml | 38 ++++++-----
 docs/user-manual/en/clusters.md                 |  4 ++
 docs/user-manual/en/configuration-index.md      |  2 +
 docs/user-manual/en/core-bridges.md             |  6 ++
 .../integration/cluster/bridge/BridgeTest.java  | 36 +++--------
 .../cluster/util/BackupSyncDelay.java           |  6 --
 17 files changed, 145 insertions(+), 129 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/activemq-6/blob/ada112a6/activemq-core-client/src/main/java/org/apache/activemq/api/config/ActiveMQDefaultConfiguration.java
----------------------------------------------------------------------
diff --git a/activemq-core-client/src/main/java/org/apache/activemq/api/config/ActiveMQDefaultConfiguration.java
b/activemq-core-client/src/main/java/org/apache/activemq/api/config/ActiveMQDefaultConfiguration.java
index 4049fb4..cea7633 100644
--- a/activemq-core-client/src/main/java/org/apache/activemq/api/config/ActiveMQDefaultConfiguration.java
+++ b/activemq-core-client/src/main/java/org/apache/activemq/api/config/ActiveMQDefaultConfiguration.java
@@ -309,6 +309,10 @@ public final class ActiveMQDefaultConfiguration
    // Once the bridge has received this many bytes, it sends a confirmation
    private static int DEFAULT_BRIDGE_CONFIRMATION_WINDOW_SIZE = 1048576;
 
+   // Producer flow control is disabled by default on the bridge
+   // You probably need to enable this if you use lots of huge messages
+   private static int DEFAULT_BRIDGE_PRODUCER_WINDOW_SIZE = -1;
+
    // Upon reconnection this configures the number of time the same node on the topology
will be retried before reseting the server locator and using the initial connectors
    private static int DEFAULT_BRIDGE_CONNECT_SAME_NODE = 10;
 
@@ -870,6 +874,16 @@ public final class ActiveMQDefaultConfiguration
       return DEFAULT_BRIDGE_CONFIRMATION_WINDOW_SIZE;
    }
 
+
+   /**
+    * This default is used for both bridge and cluster connections (since they both translate
to bridges) *
+    * @return
+    */
+   public static int getDefaultBridgeProducerWindowSize()
+   {
+      return DEFAULT_BRIDGE_PRODUCER_WINDOW_SIZE;
+   }
+
    /**
     * Upon reconnection this configures the number of time the same node on the topology
will be retried before reseting the server locator and using the initial connectors
     */

http://git-wip-us.apache.org/repos/asf/activemq-6/blob/ada112a6/activemq-core-client/src/main/java/org/apache/activemq/core/protocol/core/Channel.java
----------------------------------------------------------------------
diff --git a/activemq-core-client/src/main/java/org/apache/activemq/core/protocol/core/Channel.java
b/activemq-core-client/src/main/java/org/apache/activemq/core/protocol/core/Channel.java
index c876419..7546952 100644
--- a/activemq-core-client/src/main/java/org/apache/activemq/core/protocol/core/Channel.java
+++ b/activemq-core-client/src/main/java/org/apache/activemq/core/protocol/core/Channel.java
@@ -201,15 +201,4 @@ public interface Channel
     * @param transferring whether the channel is transferring
     */
    void setTransferring(boolean transferring);
-
-   /**
-    * for large message server send, each entry in resend cache will hold a reference to
-    * a chunk of bytes which can cause OOM if the cache quickly build up. This method
-    * make sure the resent cache size can't be more than one by blocking the call.
-    *
-    * @param timeout max waiting time for the resend cache
-    *
-    * @return true if the resend cache gets cleared
-    */
-   boolean largeServerCheck(long timeout);
 }

http://git-wip-us.apache.org/repos/asf/activemq-6/blob/ada112a6/activemq-core-client/src/main/java/org/apache/activemq/core/protocol/core/impl/ActiveMQSessionContext.java
----------------------------------------------------------------------
diff --git a/activemq-core-client/src/main/java/org/apache/activemq/core/protocol/core/impl/ActiveMQSessionContext.java
b/activemq-core-client/src/main/java/org/apache/activemq/core/protocol/core/impl/ActiveMQSessionContext.java
index a3d532d..7097bd2 100644
--- a/activemq-core-client/src/main/java/org/apache/activemq/core/protocol/core/impl/ActiveMQSessionContext.java
+++ b/activemq-core-client/src/main/java/org/apache/activemq/core/protocol/core/impl/ActiveMQSessionContext.java
@@ -461,14 +461,6 @@ public class ActiveMQSessionContext extends SessionContext
       else
       {
          sessionChannel.send(chunkPacket);
-         if (!sessionChannel.largeServerCheck(MAX_RESENDCACHE_WAITING_TIME))
-         {
-            ActiveMQClientLogger.LOGGER.warn("Bridge detected that the target server is slow
to " +
-                    " send back chunk confirmations. It 's possible the bridge may take more
memory" +
-                    " during sending of a large message. It may be a temporary situation
if this warning" +
-                    " occasionally shows up.");
-         }
-
       }
 
       return chunkPacket.getPacketSize();

http://git-wip-us.apache.org/repos/asf/activemq-6/blob/ada112a6/activemq-core-client/src/main/java/org/apache/activemq/core/protocol/core/impl/ChannelImpl.java
----------------------------------------------------------------------
diff --git a/activemq-core-client/src/main/java/org/apache/activemq/core/protocol/core/impl/ChannelImpl.java
b/activemq-core-client/src/main/java/org/apache/activemq/core/protocol/core/impl/ChannelImpl.java
index 25f6f81..a71aed5 100644
--- a/activemq-core-client/src/main/java/org/apache/activemq/core/protocol/core/impl/ChannelImpl.java
+++ b/activemq-core-client/src/main/java/org/apache/activemq/core/protocol/core/impl/ChannelImpl.java
@@ -226,27 +226,6 @@ public final class ChannelImpl implements Channel
       this.transferring = transferring;
    }
 
-   @Override
-   public boolean largeServerCheck(long timeout)
-   {
-      if (resendCache == null) return true;
-
-      synchronized (resendCache)
-      {
-         if (resendCache.size() >= 1)
-         {
-            try
-            {
-               resendCache.wait(timeout);
-            }
-            catch (InterruptedException e)
-            {
-            }
-         }
-      }
-      return resendCache.size() == 0;
-   }
-
    // This must never called by more than one thread concurrently
    public boolean send(final Packet packet, final boolean flush, final boolean batch)
    {
@@ -628,12 +607,7 @@ public final class ChannelImpl implements Channel
 
          firstStoredCommandID = 0;
 
-         synchronized (resendCache)
-         {
-            resendCache.clear();
-            resendCache.notifyAll();
-         }
-
+         resendCache.clear();
       }
    }
 
@@ -698,38 +672,28 @@ public final class ChannelImpl implements Channel
 
       int sizeToFree = 0;
 
-      try
+      for (int i = 0; i < numberToClear; i++)
       {
-         for (int i = 0; i < numberToClear; i++)
-         {
-            final Packet packet = resendCache.poll();
-
-            if (packet == null)
-            {
-               if (lastReceivedCommandID > 0)
-               {
-                  ActiveMQClientLogger.LOGGER.cannotFindPacketToClear(lastReceivedCommandID,
firstStoredCommandID);
-               }
-               firstStoredCommandID = lastReceivedCommandID + 1;
-               return;
-            }
+         final Packet packet = resendCache.poll();
 
-            if (packet.getType() != PacketImpl.PACKETS_CONFIRMED)
+         if (packet == null)
+         {
+            if (lastReceivedCommandID > 0)
             {
-               sizeToFree += packet.getPacketSize();
+               ActiveMQClientLogger.LOGGER.cannotFindPacketToClear(lastReceivedCommandID,
firstStoredCommandID);
             }
+            firstStoredCommandID = lastReceivedCommandID + 1;
+            return;
+         }
 
-            if (commandConfirmationHandler != null)
-            {
-               commandConfirmationHandler.commandConfirmed(packet);
-            }
+         if (packet.getType() != PacketImpl.PACKETS_CONFIRMED)
+         {
+            sizeToFree += packet.getPacketSize();
          }
-      }
-      finally
-      {
-         synchronized (resendCache)
+
+         if (commandConfirmationHandler != null)
          {
-            resendCache.notifyAll();
+            commandConfirmationHandler.commandConfirmed(packet);
          }
       }
 

http://git-wip-us.apache.org/repos/asf/activemq-6/blob/ada112a6/activemq-server/src/main/java/org/apache/activemq/core/config/BridgeConfiguration.java
----------------------------------------------------------------------
diff --git a/activemq-server/src/main/java/org/apache/activemq/core/config/BridgeConfiguration.java
b/activemq-server/src/main/java/org/apache/activemq/core/config/BridgeConfiguration.java
index 87fae42..65cb2a4 100644
--- a/activemq-server/src/main/java/org/apache/activemq/core/config/BridgeConfiguration.java
+++ b/activemq-server/src/main/java/org/apache/activemq/core/config/BridgeConfiguration.java
@@ -56,6 +56,8 @@ public final class BridgeConfiguration implements Serializable
 
    private int confirmationWindowSize = ActiveMQClient.DEFAULT_CONFIRMATION_WINDOW_SIZE;
 
+   private int producerWindowSize = ActiveMQDefaultConfiguration.getDefaultBridgeProducerWindowSize();
+
    private long clientFailureCheckPeriod = ActiveMQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD;
 
    private String user = ActiveMQDefaultConfiguration.getDefaultClusterUser();
@@ -302,6 +304,19 @@ public final class BridgeConfiguration implements Serializable
       return this;
    }
 
+   /** The producer flow control on the birdge */
+   public BridgeConfiguration setProducerWindowSize(final int producerWindowSize)
+   {
+      this.producerWindowSize = producerWindowSize;
+      return this;
+   }
+
+   public int getProducerWindowSize()
+   {
+      return producerWindowSize;
+
+   }
+
    public long getClientFailureCheckPeriod()
    {
       return clientFailureCheckPeriod;

http://git-wip-us.apache.org/repos/asf/activemq-6/blob/ada112a6/activemq-server/src/main/java/org/apache/activemq/core/config/ClusterConnectionConfiguration.java
----------------------------------------------------------------------
diff --git a/activemq-server/src/main/java/org/apache/activemq/core/config/ClusterConnectionConfiguration.java
b/activemq-server/src/main/java/org/apache/activemq/core/config/ClusterConnectionConfiguration.java
index 3a6c315..ea95dd8 100644
--- a/activemq-server/src/main/java/org/apache/activemq/core/config/ClusterConnectionConfiguration.java
+++ b/activemq-server/src/main/java/org/apache/activemq/core/config/ClusterConnectionConfiguration.java
@@ -63,6 +63,8 @@ public final class ClusterConnectionConfiguration implements Serializable
 
    private int confirmationWindowSize = ActiveMQDefaultConfiguration.getDefaultClusterConfirmationWindowSize();
 
+   private int producerWindowSize = ActiveMQDefaultConfiguration.getDefaultBridgeProducerWindowSize();
+
    private boolean allowDirectConnectionsOnly = false;
 
    private int minLargeMessageSize = ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE;
@@ -198,6 +200,18 @@ public final class ClusterConnectionConfiguration implements Serializable
       return this;
    }
 
+
+   public int getProducerWindowSize()
+   {
+      return producerWindowSize;
+   }
+
+   public ClusterConnectionConfiguration setProducerindowSize(int producerWindowSize)
+   {
+      this.producerWindowSize = producerWindowSize;
+      return this;
+   }
+
    public List<String> getStaticConnectors()
    {
       return staticConnectors;

http://git-wip-us.apache.org/repos/asf/activemq-6/blob/ada112a6/activemq-server/src/main/java/org/apache/activemq/core/deployers/impl/FileConfigurationParser.java
----------------------------------------------------------------------
diff --git a/activemq-server/src/main/java/org/apache/activemq/core/deployers/impl/FileConfigurationParser.java
b/activemq-server/src/main/java/org/apache/activemq/core/deployers/impl/FileConfigurationParser.java
index 775e60c..3f7c637 100644
--- a/activemq-server/src/main/java/org/apache/activemq/core/deployers/impl/FileConfigurationParser.java
+++ b/activemq-server/src/main/java/org/apache/activemq/core/deployers/impl/FileConfigurationParser.java
@@ -1417,6 +1417,10 @@ public final class FileConfigurationParser extends XMLConfigurationUtil
          getInteger(e, "confirmation-window-size", ActiveMQDefaultConfiguration.getDefaultClusterConfirmationWindowSize(),
                     Validators.GT_ZERO);
 
+      int producerWindowSize =
+         getInteger(e, "producer-window-size", ActiveMQDefaultConfiguration.getDefaultBridgeConfirmationWindowSize(),
+                    Validators.MINUS_ONE_OR_GT_ZERO);
+
       long clusterNotificationInterval = getLong(e, "notification-interval", ActiveMQDefaultConfiguration.getDefaultClusterNotificationInterval(),
Validators.GT_ZERO);
 
       int clusterNotificationAttempts = getInteger(e, "notification-attempts", ActiveMQDefaultConfiguration.getDefaultClusterNotificationAttempts(),
Validators.GT_ZERO);
@@ -1468,6 +1472,7 @@ public final class FileConfigurationParser extends XMLConfigurationUtil
          .setForwardWhenNoConsumers(forwardWhenNoConsumers)
          .setMaxHops(maxHops)
          .setConfirmationWindowSize(confirmationWindowSize)
+         .setProducerindowSize(producerWindowSize)
          .setAllowDirectConnectionsOnly(allowDirectConnectionsOnly)
          .setClusterNotificationInterval(clusterNotificationInterval)
          .setClusterNotificationAttempts(clusterNotificationAttempts);
@@ -1549,6 +1554,10 @@ public final class FileConfigurationParser extends XMLConfigurationUtil
          getInteger(brNode, "reconnect-attempts-same-node", ActiveMQDefaultConfiguration.getDefaultBridgeConnectSameNode(),
                     Validators.MINUS_ONE_OR_GE_ZERO);
 
+      int producerWindowSize =
+         getInteger(brNode, "producer-window-size", ActiveMQDefaultConfiguration.getDefaultBridgeProducerWindowSize(),
+                    Validators.MINUS_ONE_OR_GE_ZERO);
+
       boolean useDuplicateDetection = getBoolean(brNode,
                                                  "use-duplicate-detection",
                                                  ActiveMQDefaultConfiguration.isDefaultBridgeDuplicateDetection());
@@ -1630,7 +1639,8 @@ public final class FileConfigurationParser extends XMLConfigurationUtil
          .setConfirmationWindowSize(confirmationWindowSize)
          .setHA(ha)
          .setUser(user)
-         .setPassword(password);
+         .setPassword(password)
+         .setProducerWindowSize(producerWindowSize);
 
       if (!staticConnectorNames.isEmpty())
       {

http://git-wip-us.apache.org/repos/asf/activemq-6/blob/ada112a6/activemq-server/src/main/java/org/apache/activemq/core/server/cluster/ClusterManager.java
----------------------------------------------------------------------
diff --git a/activemq-server/src/main/java/org/apache/activemq/core/server/cluster/ClusterManager.java
b/activemq-server/src/main/java/org/apache/activemq/core/server/cluster/ClusterManager.java
index 24ed7c0..a289e42 100644
--- a/activemq-server/src/main/java/org/apache/activemq/core/server/cluster/ClusterManager.java
+++ b/activemq-server/src/main/java/org/apache/activemq/core/server/cluster/ClusterManager.java
@@ -529,8 +529,7 @@ public final class ClusterManager implements ActiveMQComponent
       serverLocator.setBlockOnDurableSend(!config.isUseDuplicateDetection());
       serverLocator.setBlockOnNonDurableSend(!config.isUseDuplicateDetection());
       serverLocator.setMinLargeMessageSize(config.getMinLargeMessageSize());
-      //disable flow control
-      serverLocator.setProducerWindowSize(-1);
+      serverLocator.setProducerWindowSize(config.getProducerWindowSize());
 
       // This will be set to 30s unless it's changed from embedded / testing
       // there is no reason to exception the config for this timeout
@@ -735,6 +734,7 @@ public final class ClusterManager implements ActiveMQComponent
                                                        config.isDuplicateDetection(),
                                                        config.isForwardWhenNoConsumers(),
                                                        config.getConfirmationWindowSize(),
+                                                       config.getProducerWindowSize(),
                                                        executorFactory,
                                                        server,
                                                        postOffice,
@@ -777,6 +777,7 @@ public final class ClusterManager implements ActiveMQComponent
                                                        config.isDuplicateDetection(),
                                                        config.isForwardWhenNoConsumers(),
                                                        config.getConfirmationWindowSize(),
+                                                       config.getProducerWindowSize(),
                                                        executorFactory,
                                                        server,
                                                        postOffice,

http://git-wip-us.apache.org/repos/asf/activemq-6/blob/ada112a6/activemq-server/src/main/java/org/apache/activemq/core/server/cluster/impl/ClusterConnectionImpl.java
----------------------------------------------------------------------
diff --git a/activemq-server/src/main/java/org/apache/activemq/core/server/cluster/impl/ClusterConnectionImpl.java
b/activemq-server/src/main/java/org/apache/activemq/core/server/cluster/impl/ClusterConnectionImpl.java
index 9ff9948..c02ed57 100644
--- a/activemq-server/src/main/java/org/apache/activemq/core/server/cluster/impl/ClusterConnectionImpl.java
+++ b/activemq-server/src/main/java/org/apache/activemq/core/server/cluster/impl/ClusterConnectionImpl.java
@@ -113,6 +113,8 @@ public final class ClusterConnectionImpl implements ClusterConnection,
AfterConn
 
    private final int confirmationWindowSize;
 
+   private final int producerWindowSize;
+
    /**
     * Guard for the field {@link #records}. Note that the field is {@link ConcurrentHashMap},
     * however we need the guard to synchronize multiple step operations during topology updates.
@@ -179,6 +181,7 @@ public final class ClusterConnectionImpl implements ClusterConnection,
AfterConn
                                 final boolean useDuplicateDetection,
                                 final boolean routeWhenNoConsumers,
                                 final int confirmationWindowSize,
+                                final int producerWindowSize,
                                 final ExecutorFactory executorFactory,
                                 final ActiveMQServer server,
                                 final PostOffice postOffice,
@@ -220,6 +223,8 @@ public final class ClusterConnectionImpl implements ClusterConnection,
AfterConn
 
       this.confirmationWindowSize = confirmationWindowSize;
 
+      this.producerWindowSize = producerWindowSize;
+
       this.executorFactory = executorFactory;
 
       this.clusterNotificationInterval = clusterNotificationInterval;
@@ -286,6 +291,7 @@ public final class ClusterConnectionImpl implements ClusterConnection,
AfterConn
                                 final boolean useDuplicateDetection,
                                 final boolean routeWhenNoConsumers,
                                 final int confirmationWindowSize,
+                                final int producerWindowSize,
                                 final ExecutorFactory executorFactory,
                                 final ActiveMQServer server,
                                 final PostOffice postOffice,
@@ -333,6 +339,8 @@ public final class ClusterConnectionImpl implements ClusterConnection,
AfterConn
 
       this.confirmationWindowSize = confirmationWindowSize;
 
+      this.producerWindowSize = producerWindowSize;
+
       this.executorFactory = executorFactory;
 
       this.clusterNotificationInterval = clusterNotificationInterval;
@@ -637,8 +645,7 @@ public final class ClusterConnectionImpl implements ClusterConnection,
AfterConn
          serverLocator.setBlockOnNonDurableSend(!useDuplicateDetection);
          serverLocator.setCallTimeout(callTimeout);
          serverLocator.setCallFailoverTimeout(callFailoverTimeout);
-         // No producer flow control on the bridges, as we don't want to lock the queues
-         serverLocator.setProducerWindowSize(-1);
+         serverLocator.setProducerWindowSize(producerWindowSize);
 
          if (retryInterval > 0)
          {

http://git-wip-us.apache.org/repos/asf/activemq-6/blob/ada112a6/activemq-server/src/main/resources/schema/activemq-configuration.xsd
----------------------------------------------------------------------
diff --git a/activemq-server/src/main/resources/schema/activemq-configuration.xsd b/activemq-server/src/main/resources/schema/activemq-configuration.xsd
index 10ab978..31b840c 100644
--- a/activemq-server/src/main/resources/schema/activemq-configuration.xsd
+++ b/activemq-server/src/main/resources/schema/activemq-configuration.xsd
@@ -1,4 +1,4 @@
-<?xml version="1.0" encoding="UTF-8"?>   
+<?xml version="1.0" encoding="UTF-8"?>
 <!--
   Licensed to the Apache Software Foundation (ASF) under one or more
   contributor license agreements.  See the NOTICE file distributed with
@@ -287,7 +287,7 @@
                </xsd:documentation>
             </xsd:annotation>
          </xsd:element>
-         
+
          <xsd:element name="remoting-incoming-interceptors" type="class-name-sequenceType"
maxOccurs="1" minOccurs="0">
             <xsd:annotation>
                <xsd:documentation>
@@ -1049,6 +1049,14 @@
             </xsd:annotation>
          </xsd:element>
 
+         <xsd:element name="producer-window-size" type="xsd:int" maxOccurs="1" minOccurs="0"
default="1048576">
+            <xsd:annotation>
+               <xsd:documentation>
+                  Producer flow control on the bridge (default disabled = -1)
+               </xsd:documentation>
+            </xsd:annotation>
+         </xsd:element>
+
          <xsd:element name="user" type="xsd:string" maxOccurs="1" minOccurs="0">
             <xsd:annotation>
                <xsd:documentation>
@@ -1231,6 +1239,14 @@
             </xsd:annotation>
          </xsd:element>
 
+         <xsd:element name="producer-window-size" type="xsd:int" maxOccurs="1" minOccurs="0"
default="1048576">
+            <xsd:annotation>
+               <xsd:documentation>
+                  Producer flow control on the bridge (default disabled = -1)
+               </xsd:documentation>
+            </xsd:annotation>
+         </xsd:element>
+
          <xsd:element name="call-failover-timeout" type="xsd:long" default="-1" maxOccurs="1"
minOccurs="0">
             <xsd:annotation>
                <xsd:documentation>

http://git-wip-us.apache.org/repos/asf/activemq-6/blob/ada112a6/activemq-server/src/test/java/org/apache/activemq/core/config/impl/FileConfigurationTest.java
----------------------------------------------------------------------
diff --git a/activemq-server/src/test/java/org/apache/activemq/core/config/impl/FileConfigurationTest.java
b/activemq-server/src/test/java/org/apache/activemq/core/config/impl/FileConfigurationTest.java
index 6f2a689..eb2731e 100644
--- a/activemq-server/src/test/java/org/apache/activemq/core/config/impl/FileConfigurationTest.java
+++ b/activemq-server/src/test/java/org/apache/activemq/core/config/impl/FileConfigurationTest.java
@@ -223,6 +223,7 @@ public class FileConfigurationTest extends ConfigurationImplTest
             Assert.assertEquals(true, bc.isUseDuplicateDetection());
             Assert.assertEquals("connector1", bc.getStaticConnectors().get(0));
             Assert.assertEquals(null, bc.getDiscoveryGroupName());
+            Assert.assertEquals(444, bc.getProducerWindowSize());
          }
          else
          {
@@ -233,6 +234,7 @@ public class FileConfigurationTest extends ConfigurationImplTest
             Assert.assertEquals(null, bc.getTransformerClassName());
             Assert.assertEquals(null, bc.getStaticConnectors());
             Assert.assertEquals("dg1", bc.getDiscoveryGroupName());
+            Assert.assertEquals(555, bc.getProducerWindowSize());
          }
       }
 
@@ -267,6 +269,7 @@ public class FileConfigurationTest extends ConfigurationImplTest
             Assert.assertEquals("connector1", ccc.getStaticConnectors().get(0));
             Assert.assertEquals("connector2", ccc.getStaticConnectors().get(1));
             Assert.assertEquals(null, ccc.getDiscoveryGroupName());
+            Assert.assertEquals(222, ccc.getProducerWindowSize());
          }
          else
          {
@@ -280,6 +283,7 @@ public class FileConfigurationTest extends ConfigurationImplTest
             Assert.assertEquals(2, ccc.getMaxHops());
             Assert.assertEquals(Collections.emptyList(), ccc.getStaticConnectors());
             Assert.assertEquals("dg1", ccc.getDiscoveryGroupName());
+            Assert.assertEquals(333, ccc.getProducerWindowSize());
          }
       }
 

http://git-wip-us.apache.org/repos/asf/activemq-6/blob/ada112a6/activemq-server/src/test/resources/ConfigurationTest-full-config.xml
----------------------------------------------------------------------
diff --git a/activemq-server/src/test/resources/ConfigurationTest-full-config.xml b/activemq-server/src/test/resources/ConfigurationTest-full-config.xml
index b16e11a..e55b522 100644
--- a/activemq-server/src/test/resources/ConfigurationTest-full-config.xml
+++ b/activemq-server/src/test/resources/ConfigurationTest-full-config.xml
@@ -142,6 +142,7 @@
              <reconnect-attempts>2</reconnect-attempts>
              <failover-on-server-shutdown>false</failover-on-server-shutdown>
              <use-duplicate-detection>true</use-duplicate-detection>
+            <producer-window-size>444</producer-window-size>
              <static-connectors>
                <connector-ref>connector1</connector-ref>
              </static-connectors>
@@ -149,6 +150,7 @@
          <bridge name="bridge2">
              <queue-name>queue2</queue-name>
              <forwarding-address>bridge-forwarding-address2</forwarding-address>
+            <producer-window-size>555</producer-window-size>
              <discovery-group-ref discovery-group-name="dg1"/>
          </bridge>
       </bridges>
@@ -167,24 +169,25 @@
       </ha-policy>
       <cluster-connections>
          <cluster-connection name="cluster-connection1">
-         <address>queues1</address>
-         <connector-ref>connector1</connector-ref>
-         <check-period>331</check-period>
-         <connection-ttl>3370</connection-ttl>
-         <min-large-message-size>321</min-large-message-size>
-         <call-timeout>123</call-timeout>
-         <retry-interval>3</retry-interval>
-         <retry-interval-multiplier>0.25</retry-interval-multiplier>
-         <max-retry-interval>10000</max-retry-interval>
-         <reconnect-attempts>72</reconnect-attempts>
-         <use-duplicate-detection>true</use-duplicate-detection>
-         <forward-when-no-consumers>false</forward-when-no-consumers>
-         <max-hops>1</max-hops>
-         <call-failover-timeout>123</call-failover-timeout>
-         <static-connectors>
+            <address>queues1</address>
             <connector-ref>connector1</connector-ref>
-            <connector-ref>connector2</connector-ref>
-         </static-connectors>
+            <check-period>331</check-period>
+            <connection-ttl>3370</connection-ttl>
+            <min-large-message-size>321</min-large-message-size>
+            <call-timeout>123</call-timeout>
+            <retry-interval>3</retry-interval>
+            <retry-interval-multiplier>0.25</retry-interval-multiplier>
+            <max-retry-interval>10000</max-retry-interval>
+            <reconnect-attempts>72</reconnect-attempts>
+            <use-duplicate-detection>true</use-duplicate-detection>
+            <forward-when-no-consumers>false</forward-when-no-consumers>
+            <max-hops>1</max-hops>
+            <producer-window-size>222</producer-window-size>
+            <call-failover-timeout>123</call-failover-timeout>
+            <static-connectors>
+               <connector-ref>connector1</connector-ref>
+               <connector-ref>connector2</connector-ref>
+            </static-connectors>
          </cluster-connection>
          <cluster-connection name="cluster-connection2">
              <address>queues2</address>
@@ -194,6 +197,7 @@
              <use-duplicate-detection>false</use-duplicate-detection>
              <forward-when-no-consumers>true</forward-when-no-consumers>
              <max-hops>2</max-hops>
+            <producer-window-size>333</producer-window-size>
              <call-failover-timeout>456</call-failover-timeout>
              <discovery-group-ref discovery-group-name="dg1"/>
          </cluster-connection>

http://git-wip-us.apache.org/repos/asf/activemq-6/blob/ada112a6/docs/user-manual/en/clusters.md
----------------------------------------------------------------------
diff --git a/docs/user-manual/en/clusters.md b/docs/user-manual/en/clusters.md
index 6dc8f15..baf9690 100644
--- a/docs/user-manual/en/clusters.md
+++ b/docs/user-manual/en/clusters.md
@@ -736,6 +736,10 @@ specified. The following shows all the available configuration options
     server has received `confirmation-window-size` bytes it notifies its
     client, default is 1048576. A value of -1 means no window.
 
+-   `producer-window-size`. The size for producer flow control over cluster connection.
+     it's by default disabled through the cluster connection bridge but you may want
+     to set a value if you are using really large messages in cluster. A value of -1 means
no window.
+
 -   `call-failover-timeout`. Similar to `call-timeout` but used when a
     call is made during a failover attempt. Default is -1 (no timeout).
 

http://git-wip-us.apache.org/repos/asf/activemq-6/blob/ada112a6/docs/user-manual/en/configuration-index.md
----------------------------------------------------------------------
diff --git a/docs/user-manual/en/configuration-index.md b/docs/user-manual/en/configuration-index.md
index 5991446..994c402 100644
--- a/docs/user-manual/en/configuration-index.md
+++ b/docs/user-manual/en/configuration-index.md
@@ -127,6 +127,7 @@ Name | Description
 [reconnect-attempts](core-bridges.md "Chapter 36. Core Bridges")                 |  maximum
number of retry attempts, -1 means 'no limits'. default -1
 [use-duplicate-detection](core-bridges.md "Chapter 36. Core Bridges")            |  forward
duplicate detection headers?. default true
 [confirmation-window-size](core-bridges.md "Chapter 36. Core Bridges")           |  number
of bytes before confirmations are sent. default 1MB
+[producer-window-size](core-bridges.md "Chapter 36. Core Bridges")               |  Producer
flow control size on the bridge. Default -1 (disabled)
 [user](core-bridges.md "Chapter 36. Core Bridges")                               |  Username
for the bridge, the default is the cluster username
 [password](core-bridges.md "Chapter 36. Core Bridges")                           |  Password
for the bridge, default is the cluster password
 [reconnect-attempts-same-node](core-bridges.md "Chapter 36. Core Bridges")       |  Number
of retries before trying another node. default 10
@@ -165,6 +166,7 @@ Name | Description
 [forward-when-no-consumers](clusters.md "Chapter 38. Clusters")                         
                    |   should messages be load balanced if there are no matching consumers
on target? Default=false
 [max-hops](clusters.md "Chapter 38. Clusters")                                          
                    |   maximum number of hops cluster topology is propagated. Default=1
 [confirmation-window-size](client-reconnection.md "Chapter 34. Client Reconnection and Session
Reattachment")|   The size (in bytes) of the window used for confirming data from the server
connected to. Default 1048576
+[producer-window-size](clusters.md "Chapter 38. Clusters")                              
                    |   Flow Control for the Cluster connection bridge. Default -1 (disabled)
 [call-failover-timeout](clusters.md "38.3.1. Configuring Cluster Connections")          
                    |   How long to wait for a reply if in the middle of a fail-over. -1 means
wait forever. Default -1
 [notification-interval](clusters.md "Chapter 38. Clusters")                             
                    |   how often the cluster connection will notify the cluster of its existence
right after joining the cluster. Default 1000
 [notification-attempts](clusters.md "Chapter 38. Clusters")                             
                    |   how many times this cluster connection will notify the cluster of
its existence right after joining the cluster Default 2

http://git-wip-us.apache.org/repos/asf/activemq-6/blob/ada112a6/docs/user-manual/en/core-bridges.md
----------------------------------------------------------------------
diff --git a/docs/user-manual/en/core-bridges.md b/docs/user-manual/en/core-bridges.md
index 2965d3f..3bf4f9d 100644
--- a/docs/user-manual/en/core-bridges.md
+++ b/docs/user-manual/en/core-bridges.md
@@ -195,6 +195,12 @@ Let's take a look at all the parameters in turn:
     > `confirmation-window-size` is less than or equal to
     > `max-size-bytes` to prevent the flow of messages from ceasing.
 
+-   `producer-window-size`. This optional parameter determines the
+    producer flow control through the bridge. You usually leave this off
+    unless you are dealing with huge large messages. 
+    
+    Default=-1 (disabled)
+
 -   `user`. This optional parameter determines the user name to use when
     creating the bridge connection to the remote server. If it is not
     specified the default cluster user specified by `cluster-user` in

http://git-wip-us.apache.org/repos/asf/activemq-6/blob/ada112a6/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/cluster/bridge/BridgeTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/cluster/bridge/BridgeTest.java
b/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/cluster/bridge/BridgeTest.java
index ff4adf3..b466c3a 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/cluster/bridge/BridgeTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/cluster/bridge/BridgeTest.java
@@ -1795,7 +1795,7 @@ public class BridgeTest extends ServiceTestBase
          ArrayList<String> staticConnectors = new ArrayList<String>();
          staticConnectors.add(server1tc.getName());
 
-         int minLargeMessageSize = 50 * 1024 * 1024; //50M
+         int minLargeMessageSize = 1024 * 1024;
 
          BridgeConfiguration bridgeConfiguration = new BridgeConfiguration()
                  .setName("bridge1")
@@ -1806,7 +1806,8 @@ public class BridgeTest extends ServiceTestBase
                  .setUseDuplicateDetection(false)
                  .setConfirmationWindowSize(1024)
                  .setStaticConnectors(staticConnectors)
-                 .setMinLargeMessageSize(minLargeMessageSize);
+                 .setMinLargeMessageSize(minLargeMessageSize)
+                 .setProducerWindowSize(minLargeMessageSize / 2);
 
          List<BridgeConfiguration> bridgeConfigs = new ArrayList<BridgeConfiguration>();
          bridgeConfigs.add(bridgeConfiguration);
@@ -1847,10 +1848,9 @@ public class BridgeTest extends ServiceTestBase
          session1.start();
 
          //create a large message bigger than Integer.MAX_VALUE
-         final long largeMessageSize = 3L * 1024L * 1024L * 1024L;
+         final long largeMessageSize = Integer.MAX_VALUE + 1000L;
 
-         File destDir = createDestDir("testBridgeWithVeryLargeMessage");
-         ClientMessage largeMessage = createLargeMessage(session0, largeMessageSize, destDir);
+         ClientMessage largeMessage = createLargeMessage(session0, largeMessageSize);
 
          producer0.send(largeMessage);
 
@@ -1878,7 +1878,7 @@ public class BridgeTest extends ServiceTestBase
          ClientMessage message = consumer1.receive(5000);
          message.acknowledge();
 
-         File outputFile = new File(destDir, "huge_message_received.dat");
+         File outputFile = new File(getTemporaryDir(), "huge_message_received.dat");
 
          System.out.println("-----message save to: " + outputFile.getAbsolutePath());
          FileOutputStream fileOutputStream = new FileOutputStream(outputFile);
@@ -1930,30 +1930,10 @@ public class BridgeTest extends ServiceTestBase
       assertEquals(0, loadQueues(server0).size());
    }
 
-   private File createDestDir(String dirName)
-   {
-      File clientDir = new File(getClientLargeMessagesDir());
-      if (!clientDir.exists())
-      {
-         if (!clientDir.mkdirs())
-         {
-            throw new IllegalStateException("Can't create dir " + clientDir.getAbsolutePath());
-         }
-      }
-
-      File destDir = new File(clientDir, dirName);
-      if (!destDir.mkdir())
-      {
-         throw new IllegalStateException("Can't create dir " + destDir.getAbsolutePath());
-      }
-      return destDir;
-   }
-
-
-   private ClientMessage createLargeMessage(ClientSession session, long largeMessageSize,
File destDir) throws Exception
+   private ClientMessage createLargeMessage(ClientSession session, long largeMessageSize)
throws Exception
    {
 
-      File fileInput = new File(destDir, "huge_message_to_send.dat");
+      File fileInput = new File(getTemporaryDir(), "huge_message_to_send.dat");
 
       createFile(fileInput, largeMessageSize);
 

http://git-wip-us.apache.org/repos/asf/activemq-6/blob/ada112a6/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/cluster/util/BackupSyncDelay.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/cluster/util/BackupSyncDelay.java
b/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/cluster/util/BackupSyncDelay.java
index 95506d5..710c4d3 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/cluster/util/BackupSyncDelay.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/tests/integration/cluster/util/BackupSyncDelay.java
@@ -376,12 +376,6 @@ public class BackupSyncDelay implements Interceptor
       }
 
       @Override
-      public boolean largeServerCheck(long timeout)
-      {
-         return true;
-      }
-
-      @Override
       public boolean supports(byte packetID)
       {
          return true;


Mime
View raw message