activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From clebertsuco...@apache.org
Subject [2/3] activemq-artemis git commit: ARTEMIS-1791 Large message files are not removed after redistribution across a cluster
Date Mon, 09 Apr 2018 15:11:39 GMT
ARTEMIS-1791 Large message files are not removed after redistribution across a cluster


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

Branch: refs/heads/master
Commit: de5c0d51b976a2a3c60235da56cfd854418007a7
Parents: c69d6b0
Author: Howard Gao <howard.gao@gmail.com>
Authored: Mon Apr 9 11:07:49 2018 +0800
Committer: Clebert Suconic <clebertsuconic@apache.org>
Committed: Mon Apr 9 11:06:27 2018 -0400

----------------------------------------------------------------------
 .../artemis/core/postoffice/Bindings.java       |  2 ++
 .../core/postoffice/impl/BindingsImpl.java      |  5 ++++
 .../core/postoffice/impl/PostOfficeImpl.java    | 25 ++++++++++++++++----
 .../core/server/cluster/impl/Redistributor.java |  1 +
 .../impl/WildcardAddressManagerUnitTest.java    |  5 ++++
 5 files changed, 33 insertions(+), 5 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/de5c0d51/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/Bindings.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/Bindings.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/Bindings.java
index f3592c4..30a2680 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/Bindings.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/Bindings.java
@@ -39,4 +39,6 @@ public interface Bindings extends UnproposalListener {
    boolean redistribute(Message message, Queue originatingQueue, RoutingContext context)
throws Exception;
 
    void route(Message message, RoutingContext context) throws Exception;
+
+   boolean allowRedistribute();
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/de5c0d51/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/BindingsImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/BindingsImpl.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/BindingsImpl.java
index 2e2b31c..478c700 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/BindingsImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/BindingsImpl.java
@@ -155,6 +155,11 @@ public final class BindingsImpl implements Bindings {
    }
 
    @Override
+   public boolean allowRedistribute() {
+      return messageLoadBalancingType.equals(MessageLoadBalancingType.ON_DEMAND);
+   }
+
+   @Override
    public boolean redistribute(final Message message,
                                final Queue originatingQueue,
                                final RoutingContext context) throws Exception {

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/de5c0d51/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
index b2bfe37..f1f7a38 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
@@ -970,14 +970,29 @@ public class PostOfficeImpl implements PostOffice, NotificationListener,
Binding
    public Pair<RoutingContext, Message> redistribute(final Message message,
                                                      final Queue originatingQueue,
                                                      final Transaction tx) throws Exception
{
-      // We have to copy the message and store it separately, otherwise we may lose remote
bindings in case of restart before the message
-      // arrived the target node
-      // as described on https://issues.jboss.org/browse/JBPAPP-6130
-      Message copyRedistribute = message.copy(storageManager.generateID());
 
       Bindings bindings = addressManager.getBindingsForRoutingAddress(originatingQueue.getAddress());
 
-      if (bindings != null) {
+      if (bindings != null && bindings.allowRedistribute()) {
+         // We have to copy the message and store it separately, otherwise we may lose remote
bindings in case of restart before the message
+         // arrived the target node
+         // as described on https://issues.jboss.org/browse/JBPAPP-6130
+         Message copyRedistribute = message.copy(storageManager.generateID());
+         if (tx != null) {
+            tx.addOperation(new TransactionOperationAbstract() {
+               @Override
+               public void afterRollback(Transaction tx) {
+                  try {
+                     //this will cause large message file to be
+                     //cleaned up
+                     copyRedistribute.decrementRefCount();
+                  } catch (Exception e) {
+                     logger.warn("Failed to clean up message: " + copyRedistribute);
+                  }
+               }
+            });
+         }
+
          RoutingContext context = new RoutingContextImpl(tx);
 
          boolean routed = bindings.redistribute(copyRedistribute, originatingQueue, context);

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/de5c0d51/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/Redistributor.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/Redistributor.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/Redistributor.java
index cfb9eee..7982018 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/Redistributor.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/Redistributor.java
@@ -150,6 +150,7 @@ public class Redistributor implements Consumer {
       final Pair<RoutingContext, Message> routingInfo = postOffice.redistribute(reference.getMessage(),
queue, tx);
 
       if (routingInfo == null) {
+         tx.rollback();
          return HandleStatus.BUSY;
       }
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/de5c0d51/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/WildcardAddressManagerUnitTest.java
----------------------------------------------------------------------
diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/WildcardAddressManagerUnitTest.java
b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/WildcardAddressManagerUnitTest.java
index 1c13cbd..40fadf9 100644
--- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/WildcardAddressManagerUnitTest.java
+++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/WildcardAddressManagerUnitTest.java
@@ -345,6 +345,11 @@ public class WildcardAddressManagerUnitTest extends ActiveMQTestBase
{
       public void route(Message message, RoutingContext context) throws Exception {
          System.out.println("routing message: " + message);
       }
+
+      @Override
+      public boolean allowRedistribute() {
+         return false;
+      }
    }
 
 }


Mime
View raw message