[AMQ-8354] Fix deinitialization of ReplicaBroker.

amazon-mq · Feb 28, 2024 · 153ae2a · 153ae2a
1 parent 90ee78a
commit 153ae2a
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 5 deletions.
diff --git a/activemq-broker/src/main/java/org/apache/activemq/replica/ReplicaBroker.java b/activemq-broker/src/main/java/org/apache/activemq/replica/ReplicaBroker.java
@@ -86,7 +86,7 @@ public void brokerServiceStarted(ReplicaRole role) {
 
     @Override
     public void stop() throws Exception {
-        logger.info("Stopping Source broker");
+        logger.info("Stopping broker replication.");
         deinitialize();
         super.stop();
     }
@@ -109,9 +109,8 @@ public void startAfterRoleChange() throws Exception {
     }
 
     void completeBeforeRoleChange() throws Exception {
-        messageListener.deinitialize();
-        removeReplicationQueues();
         deinitialize();
+        removeReplicationQueues();
         onStopSuccess();
     }
 
@@ -131,7 +130,7 @@ private void init(ReplicaRole role) {
         messageListener = new ReplicaBrokerEventListener(this, queueProvider, periodAcknowledgeCallBack, replicaPolicy, replicaStatistics);
     }
 
-    private void deinitialize() throws JMSException {
+    private void deinitialize() throws Exception {
         if (replicationScheduledFuture != null) {
             replicationScheduledFuture.cancel(true);
         }
@@ -142,13 +141,19 @@ private void deinitialize() throws JMSException {
         ActiveMQMessageConsumer consumer = eventConsumer.get();
         ActiveMQSession session = connectionSession.get();
         ActiveMQConnection brokerConnection = connection.get();
+        if (consumer != null) {
+            consumer.setMessageListener(null);
+        }
         if (messageListener != null) {
             messageListener.close();
         }
         if (consumer != null) {
             consumer.stop();
             consumer.close();
         }
+        if (messageListener != null) {
+            messageListener.deinitialize();
+        }
         if (session != null) {
             session.close();
         }
@@ -162,6 +167,7 @@ private void deinitialize() throws JMSException {
         connection.set(null);
         replicationScheduledFuture = null;
         ackPollerScheduledFuture = null;
+        messageListener = null;
     }
 
     @Override

diff --git a/activemq-broker/src/main/java/org/apache/activemq/replica/ReplicaBrokerEventListener.java b/activemq-broker/src/main/java/org/apache/activemq/replica/ReplicaBrokerEventListener.java
@@ -199,11 +199,13 @@ private synchronized void processMessageWithRetries(ActiveMQMessage message, Tra
             }
             return null;
         });
+
+        ReplicaEventRetrier outerRetrier = replicaEventRetrier.get();
         replicaEventRetrier.set(retrier);
         try {
             retrier.process();
         } finally {
-            replicaEventRetrier.set(null);
+            replicaEventRetrier.set(outerRetrier);
         }
     }
 

diff --git a/activemq-broker/src/main/java/org/apache/activemq/replica/ReplicaEventRetrier.java b/activemq-broker/src/main/java/org/apache/activemq/replica/ReplicaEventRetrier.java
@@ -54,6 +54,9 @@ public void process() throws InterruptedException {
                 Thread.sleep(sleepInterval);
             }
         }
+        if (!running.get()) {
+            throw new InterruptedException("Retried was stopped");
+        }
     }
 
     public void stop() {