Apache
/
activemq-artemis
mirror of https://github.com/apache/activemq-artemis.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

ARTEMIS-3501 Added exception handling on #handleAddMessage to not stop broker from starting with currupted messages

This commit is contained in:
AntonRoskvist 2021-09-28 12:20:05 +02:00 committed by Justin Bertram
parent 20511375de
commit 919245b3b3
2 changed files with 21 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
View File

@ -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);

29
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java
View File

@ -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;
}
}
}