[activemq-artemis] branch main updated: ARTEMIS-3501 Added exception handling on #handleAddMessage to not stop broker from starting with currupted messages

[jbertram]

This is an automated email from the ASF dual-hosted git repository.

jbertram pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git


The following commit(s) were added to refs/heads/main by this push:
     new 919245b  ARTEMIS-3501 Added exception handling on #handleAddMessage to 
not stop broker from starting with currupted messages
     new 98a3af8  This closes #3776
919245b is described below

commit 919245b3b3cedd64a6dd6dacd2ee7bb67a677873
Author: AntonRoskvist <anton.roskv...@volvo.com>
AuthorDate: Tue Sep 28 12:20:05 2021 +0200

    ARTEMIS-3501 Added exception handling on #handleAddMessage to not stop 
broker from starting with currupted messages
---
 .../artemis/core/server/ActiveMQServerLogger.java  |  4 +++
 .../core/server/impl/PostOfficeJournalLoader.java  | 29 +++++++++++++---------
 2 files changed, 21 insertions(+), 12 deletions(-)

diff --git 
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
 
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
index 0cfae2a..a6a7187 100644
--- 
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
+++ 
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
@@ -1759,6 +1759,10 @@ public interface ActiveMQServerLogger extends 
BasicLogger {
    @Message(id = 222303, value = "Redistribution by {0} of messageID = {1} 
failed", format = Message.Format.MESSAGE_FORMAT)
    void errorRedistributing(@Cause Throwable t, String queueName, long m);
 
+   @LogMessage(level = Logger.Level.WARN)
+   @Message(id = 222304, value = "Unable to load message from journal", format 
= Message.Format.MESSAGE_FORMAT)
+   void unableToLoadMessageFromJournal(@Cause Throwable t);
+
    @LogMessage(level = Logger.Level.ERROR)
    @Message(id = 224000, value = "Failure in initialisation", format = 
Message.Format.MESSAGE_FORMAT)
    void initializationError(@Cause Throwable e);
diff --git 
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java
 
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java
index f679188..bdb4443 100644
--- 
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java
+++ 
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java
@@ -221,23 +221,28 @@ public class PostOfficeJournalLoader implements 
JournalLoader {
          long currentTime = System.currentTimeMillis();
 
          for (AddMessageRecord record : valueRecords) {
-            long scheduledDeliveryTime = record.getScheduledDeliveryTime();
+            try {
+               long scheduledDeliveryTime = record.getScheduledDeliveryTime();
 
-            if (scheduledDeliveryTime != 0 && scheduledDeliveryTime <= 
currentTime) {
-               scheduledDeliveryTime = 0;
-               record.getMessage().setScheduledDeliveryTime(0L);
-            }
+               if (scheduledDeliveryTime != 0 && scheduledDeliveryTime <= 
currentTime) {
+                  scheduledDeliveryTime = 0;
+                  record.getMessage().setScheduledDeliveryTime(0L);
+               }
 
-            if (scheduledDeliveryTime != 0) {
-               
record.getMessage().setScheduledDeliveryTime(scheduledDeliveryTime);
-            }
+               if (scheduledDeliveryTime != 0) {
+                  
record.getMessage().setScheduledDeliveryTime(scheduledDeliveryTime);
+               }
 
-            MessageReference ref = postOffice.reload(record.getMessage(), 
queue, null);
+               MessageReference ref = postOffice.reload(record.getMessage(), 
queue, null);
 
-            ref.setDeliveryCount(record.getDeliveryCount());
+               ref.setDeliveryCount(record.getDeliveryCount());
 
-            if (scheduledDeliveryTime != 0) {
-               record.getMessage().setScheduledDeliveryTime(0L);
+               if (scheduledDeliveryTime != 0) {
+                  record.getMessage().setScheduledDeliveryTime(0L);
+               }
+            } catch (Throwable t) {
+               ActiveMQServerLogger.LOGGER.unableToLoadMessageFromJournal(t);
+               continue;
             }
          }
       }