Add option to clean 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,14 @@ public void setDeleteAllMessagesOnStartup(boolean deletePersistentMessagesOnStar
         this.deleteAllMessagesOnStartup = deletePersistentMessagesOnStartup;
     }
 
+    /**
+     * Sets whether or not 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 URI getVmConnectorURI() {
         if (vmConnectorURI == null) {
             try {
@@ -2440,6 +2449,7 @@ protected Broker addInterceptors(Broker broker) throws Exception {
         if (isSchedulerSupport()) {
             SchedulerBroker sb = new SchedulerBroker(this, broker, getJobSchedulerStore());
             sb.setMaxRepeatAllowed(maxSchedulerRepeatAllowed);
+            sb.setDeleteAllSchedulerdMessagesOnStartup(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 deleteAllSchedulerdMessagesOnStartup;
 
     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 (deleteAllSchedulerdMessagesOnStartup) {
+            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 getDeleteAllSchedulerdMessagesOnStartup() {
+        return deleteAllSchedulerdMessagesOnStartup;
+    }
+
+    public void setDeleteAllSchedulerdMessagesOnStartup(boolean deleteAllSchedulerdMessagesOnStartup) {
+        this.deleteAllSchedulerdMessagesOnStartup = deleteAllSchedulerdMessagesOnStartup;
+    }
 }