Add option for deleting all scheduled messages on startup

Author: kenliao94

activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java

@@ -184,6 +184,7 @@ public class BrokerService implements Service {
     private JmsConnector[] jmsBridgeConnectors; // these are Jms to Jms bridges
     // to other jms messaging systems
     private boolean deleteAllMessagesOnStartup;
+    private boolean deleteAllScheduledMessagesOnStartup = false;
     private boolean advisorySupport = true;
     private boolean anonymousProducerAdvisorySupport = false;
     private URI vmConnectorURI;
@@ -1630,6 +1631,18 @@ public void setDeleteAllMessagesOnStartup(boolean deletePersistentMessagesOnStar
         this.deleteAllMessagesOnStartup = deletePersistentMessagesOnStartup;
     }
 
+    /**
+     * Sets whether all scheduled messages are deleted on startup
+     * @org.apache.xbean.Property propertyEditor="org.apache.activemq.util.BooleanEditor"
+     */
+    public void setDeleteAllScheduledMessagesOnStartup(boolean deleteAllScheduledMessagesOnStartup) {
+        this.deleteAllScheduledMessagesOnStartup = deleteAllScheduledMessagesOnStartup;
+    }
+
+    public boolean isDeleteAllScheduledMessagesOnStartup() {
+        return deleteAllScheduledMessagesOnStartup;
+    }
+
     public URI getVmConnectorURI() {
         if (vmConnectorURI == null) {
             try {
@@ -2440,6 +2453,7 @@ protected Broker addInterceptors(Broker broker) throws Exception {
         if (isSchedulerSupport()) {
             SchedulerBroker sb = new SchedulerBroker(this, broker, getJobSchedulerStore());
             sb.setMaxRepeatAllowed(maxSchedulerRepeatAllowed);
+            sb.setDeleteAllScheduledMessagesOnStartup(deleteAllScheduledMessagesOnStartup);
             if (isUseJmx()) {
                 JobSchedulerViewMBean view = new JobSchedulerView(sb.getJobScheduler());
                 try {

activemq-broker/src/main/java/org/apache/activemq/broker/scheduler/SchedulerBroker.java

@@ -72,6 +72,7 @@ public class SchedulerBroker extends BrokerFilter implements JobListener {
     private final JobSchedulerStore store;
     private JobScheduler scheduler;
     private int maxRepeatAllowed = MAX_REPEAT_ALLOWED;
+    private boolean deleteAllScheduledMessagesOnStartup;
 
     public SchedulerBroker(BrokerService brokerService, Broker next, JobSchedulerStore store) throws Exception {
         super(next);
@@ -212,6 +213,9 @@ public synchronized JobScheduler getJobScheduler() throws Exception {
     public void start() throws Exception {
         this.started.set(true);
         getInternalScheduler();
+        if (deleteAllScheduledMessagesOnStartup) {
+            deleteAllScheduledMessages();
+        }
         super.start();
     }
 
@@ -364,6 +368,11 @@ private void doSchedule(Message messageSend, Object cronValue, Object periodValu
                 new ByteSequence(packet.data, packet.offset, packet.length), cronEntry, delay, period, repeat);
     }
 
+    private void deleteAllScheduledMessages() throws Exception {
+        LOG.info("Deleting all scheduled messages on startup because deleteAllScheduledMessagesOnStartup configuration has been provided");
+        getInternalScheduler().removeAllJobs();
+    }
+
     @Override
     public void scheduledJob(String id, ByteSequence job) {
         org.apache.activemq.util.ByteSequence packet = new org.apache.activemq.util.ByteSequence(job.getData(), job.getOffset(), job.getLength());
@@ -487,4 +496,12 @@ public int getMaxRepeatAllowed() {
     public void setMaxRepeatAllowed(int maxRepeatAllowed) {
         this.maxRepeatAllowed = maxRepeatAllowed;
     }
+
+    public boolean getDeleteAllScheduledMessagesOnStartup() {
+        return deleteAllScheduledMessagesOnStartup;
+    }
+
+    public void setDeleteAllScheduledMessagesOnStartup(boolean deleteAllSchedulerdMessagesOnStartup) {
+        this.deleteAllScheduledMessagesOnStartup = deleteAllSchedulerdMessagesOnStartup;
+    }
 }

activemq-unit-tests/src/test/java/org/apache/activemq/broker/scheduler/JmsSchedulerDeleteAllMessageOnStartupOptionTest.java

@@ -0,0 +1,76 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.activemq.broker.scheduler;
+
+import jakarta.jms.Connection;
+import jakarta.jms.Message;
+import jakarta.jms.MessageConsumer;
+import jakarta.jms.MessageListener;
+import jakarta.jms.MessageProducer;
+import jakarta.jms.Session;
+import jakarta.jms.TextMessage;
+import org.apache.activemq.ScheduledMessage;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+
+public class JmsSchedulerDeleteAllMessageOnStartupOptionTest extends JobSchedulerTestSupport {
+
+    private static final transient Logger LOG = LoggerFactory.getLogger(JmsSchedulerDeleteAllMessageOnStartupOptionTest.class);
+
+    @Override
+    protected boolean shouldDeleteAllScheduledMessagesOnStartup() throws Exception {
+        return true;
+    }
+
+    @Test
+    public void testDeleteAllMessageOnRestart() throws Exception {
+        // Send a message delayed by 8 seconds
+        Connection connection = createConnection();
+        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        connection.start();
+        long time_ms = 10 * 1000;
+        MessageProducer producer = session.createProducer(destination);
+        TextMessage message = session.createTextMessage("test msg");
+        message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, time_ms);
+        producer.send(message);
+        producer.close();
+        // Shutdown broker
+        restartBroker(RestartType.NORMAL);
+        // Make sure the consumer won't get the message
+        connection = createConnection();
+        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        MessageConsumer consumer = session.createConsumer(destination);
+        final int COUNT = 1;
+        final CountDownLatch latch = new CountDownLatch(COUNT);
+        consumer.setMessageListener(new MessageListener() {
+            @Override
+            public void onMessage(Message message) {
+                latch.countDown();
+            }
+        });
+        latch.await(20, TimeUnit.SECONDS);
+        assertEquals(latch.getCount(), COUNT);
+        connection.close();
+    }
+}

activemq-unit-tests/src/test/java/org/apache/activemq/broker/scheduler/JobSchedulerTestSupport.java

@@ -116,6 +116,7 @@ protected BrokerService createBroker(boolean delete) throws Exception {
         answer.setSchedulerDirectoryFile(schedulerDirectory);
         answer.setSchedulerSupport(true);
         answer.setUseJmx(isUseJmx());
+        answer.setDeleteAllScheduledMessagesOnStartup(shouldDeleteAllScheduledMessagesOnStartup());
         return answer;
     }
 
@@ -136,4 +137,8 @@ protected void restartBroker(RestartType restartType) throws Exception {
         broker.start();
         broker.waitUntilStarted();
     }
+
+    protected boolean shouldDeleteAllScheduledMessagesOnStartup() throws Exception {
+        return false;
+    }
 }

activemq-unit-tests/src/test/java/org/apache/activemq/config/ConfigTest.java

@@ -170,6 +170,7 @@ public void testBrokerConfig() throws Exception {
             assertEquals("Broker Config Error (persistent)", false, broker.isPersistent());
             assertEquals("Broker Config Error (useShutdownHook)", false, broker.isUseShutdownHook());
             assertEquals("Broker Config Error (deleteAllMessagesOnStartup)", true, broker.isDeleteAllMessagesOnStartup());
+            assertEquals("Broker Config Error (deleteAllScheduledMessagesOnStartup)", true, broker.isDeleteAllScheduledMessagesOnStartup());
             LOG.info("Success");
 
             // Check specific vm transport

activemq-unit-tests/src/test/resources/org/apache/activemq/config/example.xml

@@ -26,7 +26,7 @@
     <amq:broker brokerName="brokerConfigTest" populateJMSXUserID="false"
         useLoggingForShutdownErrors="true" useJmx="true"
         persistent="false" vmConnectorURI="vm://javacoola"
-        useShutdownHook="false" deleteAllMessagesOnStartup="true">
+        useShutdownHook="false" deleteAllMessagesOnStartup="true" deleteAllScheduledMessagesOnStartup="true">
 
         <!--
         || NOTE this config file is used for unit testing the configuration mechanism