Add ClusterListener Support for Cluster Node Failure Detection

quartz/src/main/java/org/quartz/ClusterListener.java

@@ -0,0 +1,70 @@
+
+/*
+ * All content copyright Terracotta, Inc., unless otherwise indicated. All rights reserved.
+ * Copyright IBM Corp. 2024, 2025, 2026
+ *
+ * Licensed 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.quartz;
+
+/**
+ * The interface to be implemented by classes that want to be informed when a
+ * cluster node fails in a clustered Quartz environment.
+ *
+ * <p>
+ * In cluster mode, when a node fails (detected by missing heartbeats in the
+ * database), registered ClusterListeners will be notified with the ID of the
+ * failed instance.
+ * </p>
+ *
+ * @see ListenerManager#addClusterListener(ClusterListener)
+ * @see ListenerManager#removeClusterListener(String)
+ */
+public interface ClusterListener {
+
+    /*
+     * ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+     *
+     * Interface.
+     *
+     * ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+     */
+
+    /**
+     * <p>
+     * Get the name of the <code>ClusterListener</code>.
+     * </p>
+     */
+    String getName();
+
+    /**
+     * <p>
+     * Called by the <code>{@link Scheduler}</code> when a cluster node has been
+     * detected as failed. This method is only invoked in clustered environments
+     * when the ClusterManager detects that another scheduler instance in the
+     * cluster has missed its check-in deadline.
+     * </p>
+     *
+     * <p>
+     * The <code>failedInstanceId</code> parameter contains the scheduler instance
+     * ID of the node that has failed. This can be used to identify which node
+     * in the cluster is no longer active.
+     * </p>
+     *
+     * @param failedInstanceId the scheduler instance ID of the failed cluster node
+     */
+    void clusterNodeFailed(String failedInstanceId);
+
+}

quartz/src/main/java/org/quartz/ListenerManager.java

@@ -275,4 +275,30 @@ public interface ListenerManager {
      */
     List<SchedulerListener> getSchedulerListeners();
 
+    /**
+     * Add the given <code>{@link ClusterListener}</code> to the <code>Scheduler</code>.
+     *
+     * @see ClusterListener
+     */
+    void addClusterListener(ClusterListener clusterListener);
+
+    /**
+     * Remove the identified <code>{@link ClusterListener}</code> from the <code>Scheduler</code>.
+     *
+     * @return true if the identified listener was found in the list, and
+     *         removed.
+     */
+    boolean removeClusterListener(String name);
+
+    /**
+     * Get a List containing all of the <code>{@link ClusterListener}</code>s
+     * in the <code>Scheduler</code>, in the order in which they were registered.
+     */
+    List<ClusterListener> getClusterListeners();
+
+    /**
+     * Get the <code>{@link ClusterListener}</code> that has the given name.
+     */
+    ClusterListener getClusterListener(String name);
+
 }

quartz/src/main/java/org/quartz/core/ListenerManagerImpl.java

@@ -8,6 +8,7 @@
 import java.util.List;
 import java.util.Map;
 
+import org.quartz.ClusterListener;
 import org.quartz.JobKey;
 import org.quartz.JobListener;
 import org.quartz.ListenerManager;
@@ -27,6 +28,8 @@ public class ListenerManagerImpl implements ListenerManager {
 
     private final Map<String, List<Matcher<TriggerKey>>> globalTriggerListenersMatchers = new LinkedHashMap<>(10);
 
+    private final Map<String, ClusterListener> globalClusterListeners = new LinkedHashMap<>(10);
+
     private final ArrayList<SchedulerListener> schedulerListeners = new ArrayList<>(10);
 
 
@@ -272,4 +275,34 @@ public List<SchedulerListener> getSchedulerListeners() {
             return java.util.Collections.unmodifiableList(new ArrayList<>(schedulerListeners));
         }
     }
+
+    public void addClusterListener(ClusterListener clusterListener) {
+        if (clusterListener.getName() == null
+                || clusterListener.getName().isEmpty()) {
+            throw new IllegalArgumentException(
+                    "ClusterListener name cannot be empty.");
+        }
+
+        synchronized (globalClusterListeners) {
+            globalClusterListeners.put(clusterListener.getName(), clusterListener);
+        }
+    }
+
+    public boolean removeClusterListener(String name) {
+        synchronized (globalClusterListeners) {
+            return (globalClusterListeners.remove(name) != null);
+        }
+    }
+
+    public List<ClusterListener> getClusterListeners() {
+        synchronized (globalClusterListeners) {
+            return java.util.Collections.unmodifiableList(new LinkedList<>(globalClusterListeners.values()));
+        }
+    }
+
+    public ClusterListener getClusterListener(String name) {
+        synchronized (globalClusterListeners) {
+            return globalClusterListeners.get(name);
+        }
+    }
 }

quartz/src/main/java/org/quartz/core/QuartzScheduler.java

@@ -42,6 +42,7 @@
 import javax.management.ObjectName;
 
 import org.quartz.Calendar;
+import org.quartz.ClusterListener;
 import org.quartz.InterruptableJob;
 import org.quartz.Job;
 import org.quartz.JobDataMap;
@@ -156,6 +157,8 @@ public class QuartzScheduler implements RemotableQuartzScheduler {
 
     private final ArrayList<SchedulerListener> internalSchedulerListeners = new ArrayList<>(10);
 
+    private final ArrayList<ClusterListener> internalClusterListeners = new ArrayList<>(10);
+
     private JobFactory jobFactory = new PropertySettingJobFactory();
 
     ExecutingJobsManager jobMgr = null;
@@ -1778,6 +1781,59 @@ public List<SchedulerListener> getInternalSchedulerListeners() {
         }
     }
 
+    /**
+     * <p>
+     * Register the given <code>{@link ClusterListener}</code> with the
+     * <code>Scheduler</code>'s list of internal listeners.
+     * </p>
+     */
+    public void addInternalClusterListener(ClusterListener clusterListener) {
+        synchronized (internalClusterListeners) {
+            internalClusterListeners.add(clusterListener);
+        }
+    }
+
+    /**
+     * <p>
+     * Remove the given <code>{@link ClusterListener}</code> from the
+     * <code>Scheduler</code>'s list of internal listeners.
+     * </p>
+     *
+     * @return true if the identified listener was found in the list, and
+     *         removed.
+     */
+    public boolean removeInternalClusterListener(ClusterListener clusterListener) {
+        synchronized (internalClusterListeners) {
+            return internalClusterListeners.remove(clusterListener);
+        }
+    }
+
+    /**
+     * <p>
+     * Get a List containing all of the <i>internal</i> <code>{@link ClusterListener}</code>s
+     * registered with the <code>Scheduler</code>.
+     * </p>
+     */
+    public List<ClusterListener> getInternalClusterListeners() {
+        synchronized (internalClusterListeners) {
+            return java.util.Collections.unmodifiableList(new ArrayList<>(internalClusterListeners));
+        }
+    }
+
+    public void notifyClusterListenersNodeFailed(String failedInstanceId) {
+        // build a list of all cluster listeners that are to be notified...
+        List<ClusterListener> clusterListeners = buildClusterListenerList();
+
+        // notify all cluster listeners
+        for(ClusterListener cl: clusterListeners) {
+            try {
+                cl.clusterNodeFailed(failedInstanceId);
+            } catch (Exception e) {
+                getLog().error("Error while notifying ClusterListener of failed node: {}", failedInstanceId, e);
+            }
+        }
+    }
+
     protected void notifyJobStoreJobComplete(OperableTrigger trigger, JobDetail detail, CompletedExecutionInstruction instCode) {
         resources.getJobStore().triggeredJobComplete(trigger, detail, instCode);
     }
@@ -1817,7 +1873,15 @@ private List<SchedulerListener> buildSchedulerListenerList() {
 
         return allListeners;
     }
-
+
+    private List<ClusterListener> buildClusterListenerList() {
+        List<ClusterListener> allListeners = new LinkedList<>();
+        allListeners.addAll(getListenerManager().getClusterListeners());
+        allListeners.addAll(getInternalClusterListeners());
+
+        return allListeners;
+    }
+
     private boolean matchJobListener(JobListener listener, JobKey key) {
         List<Matcher<JobKey>> matchers = getListenerManager().getJobListenerMatchers(listener.getName());
         if(matchers == null)

quartz/src/main/java/org/quartz/core/SchedulerSignalerImpl.java

@@ -96,4 +96,8 @@ public void notifySchedulerListenersJobDeleted(JobKey jobKey) {
     public void notifySchedulerListenersError(String string, SchedulerException jpe) {
         sched.notifySchedulerListenersError(string, jpe);
     }
+
+    public void notifyClusterListenersNodeFailed(String failedInstanceId) {
+        sched.notifyClusterListenersNodeFailed(failedInstanceId);
+    }
 }

quartz/src/main/java/org/quartz/impl/jdbcjobstore/JobStoreSupport.java

@@ -3421,6 +3421,14 @@ protected void clusterRecover(Connection conn, List<SchedulerStateRecord> failed
             logWarnIfNonZero(failedInstances.size(),
                     "ClusterManager: detected " + failedInstances.size()
                             + " failed or restarted instances.");
+
+            // Notify ClusterListeners of failed nodes
+            for (SchedulerStateRecord rec : failedInstances) {
+                if (!rec.getSchedulerInstanceId().equals(getInstanceId())) {
+                    schedSignaler.notifyClusterListenersNodeFailed(rec.getSchedulerInstanceId());
+                }
+            }
+
             try {
                 for (SchedulerStateRecord rec : failedInstances) {
                     getLog().info("ClusterManager: Scanning for instance \"{}\"'s failed in-progress jobs.", rec.getSchedulerInstanceId());

quartz/src/main/java/org/quartz/listeners/ClusterListenerSupport.java

@@ -0,0 +1,68 @@
+
+/*
+ * All content copyright Terracotta, Inc., unless otherwise indicated. All rights reserved.
+ * Copyright IBM Corp. 2024, 2025, 2026
+ *
+ * Licensed 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.quartz.listeners;
+
+import org.quartz.ClusterListener;
+
+/**
+ * A helpful abstract base class for implementors of
+ * <code>{@link ClusterListener}</code>.
+ *
+ * <p>
+ * The implementations of this class only override methods they are interested
+ * in receiving, while adhering to the contract of
+ * <code>{@link ClusterListener}</code>.
+ * </p>
+ *
+ * @see ClusterListener
+ */
+public abstract class ClusterListenerSupport implements ClusterListener {
+
+    private final String name;
+
+    /**
+     * Constructor with a default name.
+     */
+    protected ClusterListenerSupport() {
+        this.name = getClass().getName() + "_" + System.currentTimeMillis();
+    }
+
+    /**
+     * Constructor with a specified name.
+     *
+     * @param name the name of the listener
+     */
+    protected ClusterListenerSupport(String name) {
+        if (name == null || name.isEmpty()) {
+            throw new IllegalArgumentException("Listener name cannot be null or empty");
+        }
+        this.name = name;
+    }
+
+    @Override
+    public String getName() {
+        return name;
+    }
+
+    @Override
+    public void clusterNodeFailed(String failedInstanceId) {
+        // do nothing by default
+    }
+}

quartz/src/main/java/org/quartz/spi/SchedulerSignaler.java

@@ -47,4 +47,12 @@ public interface SchedulerSignaler {
     void signalSchedulingChange(long candidateNewNextFireTime);
 
     void notifySchedulerListenersError(String string, SchedulerException jpe);
+
+    /**
+     * Notify all registered <code>{@link org.quartz.ClusterListener}</code>s that a
+     * cluster node has failed.
+     *
+     * @param failedInstanceId the scheduler instance ID of the failed cluster node
+     */
+    void notifyClusterListenersNodeFailed(String failedInstanceId);
 }

quartz/src/test/java/org/quartz/AbstractJobStoreTest.java

@@ -802,6 +802,9 @@ public void notifySchedulerListenersJobDeleted(JobKey jobKey) {
 
         public void notifySchedulerListenersError(String string, SchedulerException jpe) {
         }
+
+        public void notifyClusterListenersNodeFailed(String failedInstanceId) {
+        }
     }
 
     /** An empty job for testing purpose. */