You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cs...@apache.org on 2025/07/23 21:10:36 UTC
(activemq) branch main updated: AMQ-9747 - Handle IOExceptionHandler thrown exceptions in KahaDB (#1474)
This is an automated email from the ASF dual-hosted git repository.
cshannon pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq.git
The following commit(s) were added to refs/heads/main by this push:
new 2eb0d66750 AMQ-9747 - Handle IOExceptionHandler thrown exceptions in KahaDB (#1474)
2eb0d66750 is described below
commit 2eb0d6675008a5e7fd6cbec4a177ab13fde3b8f2
Author: Christopher L. Shannon <cs...@apache.org>
AuthorDate: Wed Jul 23 17:10:31 2025 -0400
AMQ-9747 - Handle IOExceptionHandler thrown exceptions in KahaDB (#1474)
Update the KahaDB scheduled tasks to catch any runtime exceptions thrown
by the configured IOExceptionHandler. This will prevent the tasks from
being killed and no longer running if the IOExceptionHandler does
not shut down the broker.
---
.../activemq/store/kahadb/MessageDatabase.java | 35 +++++++++--
.../apache/activemq/store/kahadb/KahaDBTest.java | 69 +++++++++++++++++++++-
2 files changed, 97 insertions(+), 7 deletions(-)
diff --git a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
index dbab306fc7..088d9f297d 100644
--- a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
+++ b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
@@ -448,10 +448,10 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
}
} catch (IOException ioe) {
LOG.error("Checkpoint failed", ioe);
- brokerService.handleIOException(ioe);
+ handleIOException("CheckpointRunner", ioe);
} catch (Throwable e) {
LOG.error("Checkpoint failed", e);
- brokerService.handleIOException(IOExceptionSupport.create(e));
+ handleIOException("CheckpointRunner", IOExceptionSupport.create(e));
}
}
}
@@ -2120,10 +2120,10 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
forwarded = true;
} catch (IOException ioe) {
LOG.error("Forwarding of acks failed", ioe);
- brokerService.handleIOException(ioe);
+ handleIOException("AckCompactionRunner", ioe);
} catch (Throwable e) {
LOG.error("Forwarding of acks failed", e);
- brokerService.handleIOException(IOExceptionSupport.create(e));
+ handleIOException("AckCompactionRunner", IOExceptionSupport.create(e));
}
} finally {
checkpointLock.readLock().unlock();
@@ -2136,10 +2136,10 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
}
} catch (IOException ioe) {
LOG.error("Checkpoint failed", ioe);
- brokerService.handleIOException(ioe);
+ handleIOException("AckCompactionRunner", ioe);
} catch (Throwable e) {
LOG.error("Checkpoint failed", e);
- brokerService.handleIOException(IOExceptionSupport.create(e));
+ handleIOException("AckCompactionRunner", IOExceptionSupport.create(e));
}
}
}
@@ -4274,4 +4274,27 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
}
}
+
+ /*
+ * Execute the configured IOExceptionHandler when an IOException is thrown during
+ * task execution and handle any runtime exceptions that the handler itself might throw.
+ *
+ * By default, the DefaultIOExceptionHandler will stop the broker when handling an IOException,
+ * however, if DefaultIOExceptionHandler is configured with startStopConnectors to be true
+ * it will throw a SuppressReplyException and not stop the broker. It's also possible another
+ * custom implementation of IOExceptionHandler could throw a runtime exception.
+ *
+ * This method will now handle and log those runtime exceptions so that the task will not
+ * die and will continue to execute future iterations if the broker is not shut down.
+ */
+ private void handleIOException(String taskName, IOException ioe) {
+ try {
+ brokerService.handleIOException(ioe);
+ } catch (RuntimeException e) {
+ LOG.warn("IOException handler threw exception in task {} with "
+ + "error: {}, continuing.", taskName,
+ e.getMessage());
+ LOG.debug(e.getMessage(), e);
+ }
+ }
}
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBTest.java
index e1fc1d7781..d06544f673 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBTest.java
@@ -27,11 +27,15 @@ import jakarta.jms.MessageConsumer;
import jakarta.jms.MessageProducer;
import jakarta.jms.Session;
+import java.util.concurrent.atomic.AtomicInteger;
import junit.framework.TestCase;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.util.DefaultIOExceptionHandler;
+import org.apache.activemq.util.IOExceptionHandler;
+import org.apache.activemq.util.Wait;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.core.LogEvent;
@@ -236,6 +240,69 @@ public class KahaDBTest extends TestCase {
assertFalse("Did not replay any records from the journal", didSomeRecovery.get());
}
+
+ /**
+ * Test the checkpoint runner task continues to run if the configured
+ * IOExceptionHandler throws a runtime exception while processing
+ * the IOException it is handling and the broker is not shut down. This
+ * could happen if using the DefaultIOExceptionHandler and startStopConnectors
+ * is set to true, or if a user provides their own IOExceptionHandler that
+ * throws an exception.
+ */
+ public void testCheckpointExceptionKeepRunning() throws Exception {
+ testCheckpointIOException(true);
+ }
+
+ /**
+ * Test the broker shuts down by when DefaultIOExceptionHandler
+ * handles an IOException thrown by the checkpoint runner task. This is the
+ * default behavior of the broker if not configured with a custom
+ * IOExceptionHandler and startStopConnectors is false
+ */
+ public void testCheckpointExceptionShutdown() throws Exception {
+ testCheckpointIOException(false);
+ }
+
+ private void testCheckpointIOException(boolean startStopConnectors) throws Exception {
+ final AtomicInteger iterations = new AtomicInteger();
+ // Create a store that always throws an IOException when checkpoint is called
+ final KahaDBStore kaha = new KahaDBStore() {
+ @Override
+ protected void checkpointCleanup(boolean cleanup) throws IOException {
+ iterations.incrementAndGet();
+ throw new IOException("fail");
+ }
+ };
+ kaha.setDirectory(new File("target/activemq-data/kahadb"));
+ kaha.deleteAllMessages();
+ // Set the checkpoint interval to be very short so we can quickly
+ // check number of iterations
+ kaha.setCheckpointInterval(100);
+
+ BrokerService broker = createBroker(kaha);
+ DefaultIOExceptionHandler ioExceptionHandler = new DefaultIOExceptionHandler();
+ ioExceptionHandler.setStopStartConnectors(startStopConnectors);
+ broker.setIoExceptionHandler(ioExceptionHandler);
+ broker.start();
+
+ try {
+ if (startStopConnectors) {
+ // If startStopConnectors is true, the task should continue with future iterations
+ // as the SuppressReplyException that will be thrown is now handled so just verify
+ // we see 10 iterations which should happen quickly
+ assertTrue(Wait.waitFor(() -> iterations.get() == 10, 2000, 100));
+ // broker should not be stopped
+ assertFalse(broker.isStopped());
+ } else {
+ // If startStopConnectors is false, an IOException should shut down the broker
+ // which is the normal behavior
+ assertTrue(Wait.waitFor(broker::isStopped, 2000, 100));
+ }
+ } finally {
+ broker.stop();
+ }
+ }
+
private void assertExistsAndDelete(File file) {
assertTrue(file.exists());
file.delete();
@@ -281,4 +348,4 @@ public class KahaDBTest extends TestCase {
return sb.toString();
}
-}
\ No newline at end of file
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@activemq.apache.org
For additional commands, e-mail: commits-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact