diff --git a/quartz/src/main/java/org/quartz/ClusterListener.java b/quartz/src/main/java/org/quartz/ClusterListener.java new file mode 100644 index 000000000..f30af5ca2 --- /dev/null +++ b/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. + * + *

+ * 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. + *

+ * + * @see ListenerManager#addClusterListener(ClusterListener) + * @see ListenerManager#removeClusterListener(String) + */ +public interface ClusterListener { + + /* + * ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + * + * Interface. + * + * ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + */ + + /** + *

+ * Get the name of the ClusterListener. + *

+ */ + String getName(); + + /** + *

+ * Called by the {@link Scheduler} 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. + *

+ * + *

+ * The failedInstanceId 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. + *

+ * + * @param failedInstanceId the scheduler instance ID of the failed cluster node + */ + void clusterNodeFailed(String failedInstanceId); + +} diff --git a/quartz/src/main/java/org/quartz/ListenerManager.java b/quartz/src/main/java/org/quartz/ListenerManager.java index bdf780f86..7075b25a6 100644 --- a/quartz/src/main/java/org/quartz/ListenerManager.java +++ b/quartz/src/main/java/org/quartz/ListenerManager.java @@ -275,4 +275,30 @@ public interface ListenerManager { */ List getSchedulerListeners(); + /** + * Add the given {@link ClusterListener} to the Scheduler. + * + * @see ClusterListener + */ + void addClusterListener(ClusterListener clusterListener); + + /** + * Remove the identified {@link ClusterListener} from the Scheduler. + * + * @return true if the identified listener was found in the list, and + * removed. + */ + boolean removeClusterListener(String name); + + /** + * Get a List containing all of the {@link ClusterListener}s + * in the Scheduler, in the order in which they were registered. + */ + List getClusterListeners(); + + /** + * Get the {@link ClusterListener} that has the given name. + */ + ClusterListener getClusterListener(String name); + } \ No newline at end of file diff --git a/quartz/src/main/java/org/quartz/core/ListenerManagerImpl.java b/quartz/src/main/java/org/quartz/core/ListenerManagerImpl.java index 8e43800d8..055dcfceb 100644 --- a/quartz/src/main/java/org/quartz/core/ListenerManagerImpl.java +++ b/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>> globalTriggerListenersMatchers = new LinkedHashMap<>(10); + private final Map globalClusterListeners = new LinkedHashMap<>(10); + private final ArrayList schedulerListeners = new ArrayList<>(10); @@ -272,4 +275,34 @@ public List 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 getClusterListeners() { + synchronized (globalClusterListeners) { + return java.util.Collections.unmodifiableList(new LinkedList<>(globalClusterListeners.values())); + } + } + + public ClusterListener getClusterListener(String name) { + synchronized (globalClusterListeners) { + return globalClusterListeners.get(name); + } + } } diff --git a/quartz/src/main/java/org/quartz/core/QuartzScheduler.java b/quartz/src/main/java/org/quartz/core/QuartzScheduler.java index ca52753f9..20f583002 100644 --- a/quartz/src/main/java/org/quartz/core/QuartzScheduler.java +++ b/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 internalSchedulerListeners = new ArrayList<>(10); + private final ArrayList internalClusterListeners = new ArrayList<>(10); + private JobFactory jobFactory = new PropertySettingJobFactory(); ExecutingJobsManager jobMgr = null; @@ -1778,6 +1781,59 @@ public List getInternalSchedulerListeners() { } } + /** + *

+ * Register the given {@link ClusterListener} with the + * Scheduler's list of internal listeners. + *

+ */ + public void addInternalClusterListener(ClusterListener clusterListener) { + synchronized (internalClusterListeners) { + internalClusterListeners.add(clusterListener); + } + } + + /** + *

+ * Remove the given {@link ClusterListener} from the + * Scheduler's list of internal listeners. + *

+ * + * @return true if the identified listener was found in the list, and + * removed. + */ + public boolean removeInternalClusterListener(ClusterListener clusterListener) { + synchronized (internalClusterListeners) { + return internalClusterListeners.remove(clusterListener); + } + } + + /** + *

+ * Get a List containing all of the internal {@link ClusterListener}s + * registered with the Scheduler. + *

+ */ + public List 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 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 buildSchedulerListenerList() { return allListeners; } - + + private List buildClusterListenerList() { + List allListeners = new LinkedList<>(); + allListeners.addAll(getListenerManager().getClusterListeners()); + allListeners.addAll(getInternalClusterListeners()); + + return allListeners; + } + private boolean matchJobListener(JobListener listener, JobKey key) { List> matchers = getListenerManager().getJobListenerMatchers(listener.getName()); if(matchers == null) diff --git a/quartz/src/main/java/org/quartz/core/SchedulerSignalerImpl.java b/quartz/src/main/java/org/quartz/core/SchedulerSignalerImpl.java index 406cde55a..79455ebb0 100644 --- a/quartz/src/main/java/org/quartz/core/SchedulerSignalerImpl.java +++ b/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); + } } diff --git a/quartz/src/main/java/org/quartz/impl/jdbcjobstore/JobStoreSupport.java b/quartz/src/main/java/org/quartz/impl/jdbcjobstore/JobStoreSupport.java index 2bb1c23a5..eb6476a73 100644 --- a/quartz/src/main/java/org/quartz/impl/jdbcjobstore/JobStoreSupport.java +++ b/quartz/src/main/java/org/quartz/impl/jdbcjobstore/JobStoreSupport.java @@ -3421,6 +3421,14 @@ protected void clusterRecover(Connection conn, List 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()); diff --git a/quartz/src/main/java/org/quartz/listeners/ClusterListenerSupport.java b/quartz/src/main/java/org/quartz/listeners/ClusterListenerSupport.java new file mode 100644 index 000000000..f0335e87b --- /dev/null +++ b/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 + * {@link ClusterListener}. + * + *

+ * The implementations of this class only override methods they are interested + * in receiving, while adhering to the contract of + * {@link ClusterListener}. + *

+ * + * @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 + } +} diff --git a/quartz/src/main/java/org/quartz/spi/SchedulerSignaler.java b/quartz/src/main/java/org/quartz/spi/SchedulerSignaler.java index a33932691..03300f777 100644 --- a/quartz/src/main/java/org/quartz/spi/SchedulerSignaler.java +++ b/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 {@link org.quartz.ClusterListener}s that a + * cluster node has failed. + * + * @param failedInstanceId the scheduler instance ID of the failed cluster node + */ + void notifyClusterListenersNodeFailed(String failedInstanceId); } diff --git a/quartz/src/test/java/org/quartz/AbstractJobStoreTest.java b/quartz/src/test/java/org/quartz/AbstractJobStoreTest.java index 40de0cf44..861261de5 100644 --- a/quartz/src/test/java/org/quartz/AbstractJobStoreTest.java +++ b/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. */ diff --git a/quartz/src/test/java/org/quartz/core/ClusterListenerTest.java b/quartz/src/test/java/org/quartz/core/ClusterListenerTest.java new file mode 100644 index 000000000..a998e6eeb --- /dev/null +++ b/quartz/src/test/java/org/quartz/core/ClusterListenerTest.java @@ -0,0 +1,274 @@ + +/* + * 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.core; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.junit.jupiter.api.Test; +import org.quartz.ClusterListener; +import org.quartz.listeners.ClusterListenerSupport; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +/** + * Test ClusterListener functionality + */ +class ClusterListenerTest { + + private static class TestClusterListener extends ClusterListenerSupport { + private final List failedInstances = new CopyOnWriteArrayList<>(); + + public TestClusterListener(String name) { + super(name); + } + + @Override + public void clusterNodeFailed(String failedInstanceId) { + failedInstances.add(failedInstanceId); + } + + public List getFailedInstances() { + return failedInstances; + } + + public void clear() { + failedInstances.clear(); + } + } + + /** + * Test that ClusterListenerSupport provides default name when not specified + */ + @Test + void testClusterListenerSupportDefaultName() { + ClusterListener listener = new ClusterListenerSupport() { + @Override + public void clusterNodeFailed(String failedInstanceId) { + // Do nothing + } + }; + + assertNotNull(listener.getName(), "Listener name should not be null"); + assertTrue(listener.getName().contains("ClusterListenerTest"), "Listener name should contain class name"); + } + + /** + * Test that ClusterListenerSupport uses provided name + */ + @Test + void testClusterListenerSupportCustomName() { + TestClusterListener listener = new TestClusterListener("myCustomListener"); + assertEquals("myCustomListener", listener.getName()); + } + + /** + * Test that ClusterListenerSupport throws exception for empty name + */ + @Test + void testClusterListenerSupportEmptyName() { + assertThrows(IllegalArgumentException.class, () -> { + new ClusterListenerSupport("") { + @Override + public void clusterNodeFailed(String failedInstanceId) { + // Do nothing + } + }; + }, "Expected IllegalArgumentException for empty name"); + } + + /** + * Test that ClusterListenerSupport throws exception for null name + */ + @Test + void testClusterListenerSupportNullName() { + assertThrows(IllegalArgumentException.class, () -> { + new ClusterListenerSupport(null) { + @Override + public void clusterNodeFailed(String failedInstanceId) { + // Do nothing + } + }; + }, "Expected IllegalArgumentException for null name"); + } + + /** + * Test clusterNodeFailed callback + */ + @Test + void testClusterNodeFailedCallback() { + TestClusterListener listener = new TestClusterListener("testListener"); + + assertTrue(listener.getFailedInstances().isEmpty(), "Failed instances list should be empty initially"); + + listener.clusterNodeFailed("node1"); + assertEquals(1, listener.getFailedInstances().size(), "Should have 1 failed instance"); + assertEquals("node1", listener.getFailedInstances().get(0)); + + listener.clusterNodeFailed("node2"); + assertEquals(2, listener.getFailedInstances().size(), "Should have 2 failed instances"); + assertEquals("node2", listener.getFailedInstances().get(1)); + } + + /** + * Test multiple listeners receive notifications independently + */ + @Test + void testMultipleListeners() { + TestClusterListener listener1 = new TestClusterListener("listener1"); + TestClusterListener listener2 = new TestClusterListener("listener2"); + + listener1.clusterNodeFailed("nodeA"); + listener2.clusterNodeFailed("nodeB"); + + assertEquals(1, listener1.getFailedInstances().size()); + assertEquals("nodeA", listener1.getFailedInstances().get(0)); + + assertEquals(1, listener2.getFailedInstances().size()); + assertEquals("nodeB", listener2.getFailedInstances().get(0)); + } + + /** + * Test clear functionality + */ + @Test + void testClearFailedInstances() { + TestClusterListener listener = new TestClusterListener("testListener"); + + listener.clusterNodeFailed("node1"); + listener.clusterNodeFailed("node2"); + assertEquals(2, listener.getFailedInstances().size()); + + listener.clear(); + assertTrue(listener.getFailedInstances().isEmpty(), "Failed instances list should be empty after clear"); + } + + /** + * Test exception handling in listener - should not propagate + */ + @Test + void testListenerExceptionHandling() { + final List receivedInstances = new ArrayList<>(); + + ClusterListener throwingListener = new ClusterListener() { + @Override + public String getName() { + return "throwingListener"; + } + + @Override + public void clusterNodeFailed(String failedInstanceId) { + throw new RuntimeException("Test exception for " + failedInstanceId); + } + }; + + ClusterListener normalListener = new ClusterListener() { + @Override + public String getName() { + return "normalListener"; + } + + @Override + public void clusterNodeFailed(String failedInstanceId) { + receivedInstances.add(failedInstanceId); + } + }; + + // Test that exceptions don't affect other listeners + assertThrows(RuntimeException.class, () -> { + throwingListener.clusterNodeFailed("node1"); + }, "Expected RuntimeException"); + + // Normal listener should still work + normalListener.clusterNodeFailed("node2"); + assertEquals(1, receivedInstances.size()); + assertEquals("node2", receivedInstances.get(0)); + } + + /** + * Test listener ordering by tracking notification order + */ + @Test + void testListenerNotificationOrder() { + final List notificationOrder = new ArrayList<>(); + + ClusterListener listener1 = new ClusterListener() { + @Override + public String getName() { return "listener1"; } + @Override + public void clusterNodeFailed(String failedInstanceId) { + notificationOrder.add("listener1:" + failedInstanceId); + } + }; + + ClusterListener listener2 = new ClusterListener() { + @Override + public String getName() { return "listener2"; } + @Override + public void clusterNodeFailed(String failedInstanceId) { + notificationOrder.add("listener2:" + failedInstanceId); + } + }; + + ClusterListener listener3 = new ClusterListener() { + @Override + public String getName() { return "listener3"; } + @Override + public void clusterNodeFailed(String failedInstanceId) { + notificationOrder.add("listener3:" + failedInstanceId); + } + }; + + // Simulate notification in order + listener1.clusterNodeFailed("node1"); + listener2.clusterNodeFailed("node1"); + listener3.clusterNodeFailed("node1"); + + assertEquals(3, notificationOrder.size()); + assertEquals("listener1:node1", notificationOrder.get(0)); + assertEquals("listener2:node1", notificationOrder.get(1)); + assertEquals("listener3:node1", notificationOrder.get(2)); + } + + /** + * Test listener name uniqueness + */ + @Test + void testListenerNameUniqueness() { + ClusterListener listener1 = new ClusterListenerSupport("sameName") { + @Override + public void clusterNodeFailed(String failedInstanceId) { + } + }; + + ClusterListener listener2 = new ClusterListenerSupport("sameName") { + @Override + public void clusterNodeFailed(String failedInstanceId) { + } + }; + + // Both can have the same name, it's up to the manager to handle uniqueness + assertEquals("sameName", listener1.getName()); + assertEquals("sameName", listener2.getName()); + } +} diff --git a/quartz/src/test/java/org/quartz/core/ListenerManagerTest.java b/quartz/src/test/java/org/quartz/core/ListenerManagerTest.java index 8a84fcbac..6d06210dc 100644 --- a/quartz/src/test/java/org/quartz/core/ListenerManagerTest.java +++ b/quartz/src/test/java/org/quartz/core/ListenerManagerTest.java @@ -17,6 +17,8 @@ package org.quartz.core; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; import static org.quartz.impl.matchers.GroupMatcher.jobGroupEquals; import static org.quartz.impl.matchers.GroupMatcher.triggerGroupEquals; @@ -27,11 +29,13 @@ import org.junit.jupiter.api.Test; +import org.quartz.ClusterListener; import org.quartz.JobListener; import org.quartz.SchedulerListener; import org.quartz.TriggerKey; import org.quartz.TriggerListener; import org.quartz.impl.matchers.NameMatcher; +import org.quartz.listeners.ClusterListenerSupport; import org.quartz.listeners.JobListenerSupport; import org.quartz.listeners.SchedulerListenerSupport; import org.quartz.listeners.TriggerListenerSupport; @@ -72,6 +76,27 @@ public static class TestSchedulerListener extends SchedulerListenerSupport { } + public static class TestClusterListener extends ClusterListenerSupport { + + public TestClusterListener() { + super(); + } + + public TestClusterListener(String name) { + super(name); + } + + private String lastFailedInstanceId; + + @Override + public void clusterNodeFailed(String failedInstanceId) { + this.lastFailedInstanceId = failedInstanceId; + } + + public String getLastFailedInstanceId() { + return lastFailedInstanceId; + } + } @Test @@ -191,4 +216,71 @@ void testManagementOfSchedulerListeners() throws Exception { } } + @Test + public void testManagementOfClusterListeners() throws Exception { + + TestClusterListener tl1 = new TestClusterListener("tl1"); + TestClusterListener tl2 = new TestClusterListener("tl2"); + + ListenerManagerImpl manager = new ListenerManagerImpl(); + + // test adding listener + manager.addClusterListener(tl1); + assertEquals(1, manager.getClusterListeners().size(), "Unexpected size of listener list"); + + // test adding another listener + manager.addClusterListener(tl2); + assertEquals(2, manager.getClusterListeners().size(), "Unexpected size of listener list"); + + // test getting a listener by name + assertSame(tl1, manager.getClusterListener("tl1"), "Expected to find listener by name"); + assertSame(tl2, manager.getClusterListener("tl2"), "Expected to find listener by name"); + + // test removing a listener + manager.removeClusterListener("tl1"); + assertEquals(1, manager.getClusterListeners().size(), "Unexpected size of listener list"); + assertNull(manager.getClusterListener("tl1"), "Expected listener to be removed"); + + // test removing non-existent listener + assertFalse(manager.removeClusterListener("nonexistent"), "Expected false when removing non-existent listener"); + + // Test ordering of registration is preserved. + final int numListenersToTestOrderOf = 15; + manager = new ListenerManagerImpl(); + ClusterListener[] lstners = new ClusterListener[numListenersToTestOrderOf]; + for(int i = 0; i < numListenersToTestOrderOf; i++) { + // use random name, to help test that order isn't based on naming or coincidental hashing + lstners[i] = new TestClusterListener(UUID.randomUUID().toString()); + manager.addClusterListener(lstners[i]); + } + List mls = manager.getClusterListeners(); + int i = 0; + for(ClusterListener lsnr: mls) { + assertSame(lstners[i], lsnr, "Unexpected order of listeners"); + i++; + } + } + + @Test + public void testClusterListenerNotification() throws Exception { + TestClusterListener listener = new TestClusterListener("testListener"); + + ListenerManagerImpl manager = new ListenerManagerImpl(); + manager.addClusterListener(listener); + + // Simulate notification through QuartzScheduler + QuartzScheduler scheduler = null; + try { + // Note: In actual usage, the QuartzScheduler's notifyClusterListenersNodeFailed() + // would be called by the SchedulerSignaler when a cluster node fails. + // Here we just verify the listener is properly registered and can be retrieved. + assertEquals(1, manager.getClusterListeners().size(), "Expected listener to be registered"); + assertSame(listener, manager.getClusterListener("testListener"), "Expected to find the registered listener"); + } finally { + if (scheduler != null) { + scheduler.shutdown(); + } + } + } + } diff --git a/quartz/src/test/java/org/quartz/core/SchedulerSignalerClusterTest.java b/quartz/src/test/java/org/quartz/core/SchedulerSignalerClusterTest.java new file mode 100644 index 000000000..03d3ad4d5 --- /dev/null +++ b/quartz/src/test/java/org/quartz/core/SchedulerSignalerClusterTest.java @@ -0,0 +1,228 @@ + +/* + * 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.core; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import org.junit.jupiter.api.Test; +import org.quartz.ClusterListener; +import org.quartz.listeners.ClusterListenerSupport; +import org.quartz.spi.SchedulerSignaler; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +/** + * Test SchedulerSignaler integration with ClusterListener + */ +class SchedulerSignalerClusterTest { + + private static class TestClusterListener extends ClusterListenerSupport { + // Use CopyOnWriteArrayList for thread-safe concurrent access + private final List failedInstances = new CopyOnWriteArrayList<>(); + + public TestClusterListener(String name) { + super(name); + } + + @Override + public void clusterNodeFailed(String failedInstanceId) { + failedInstances.add(failedInstanceId); + } + + public List getFailedInstances() { + return failedInstances; + } + + public void clear() { + failedInstances.clear(); + } + } + + /** + * Test that a SchedulerSignaler implementation properly delegates to ClusterListeners. + * This test uses a mock SchedulerSignaler to verify the contract. + */ + @Test + void testSchedulerSignalerNotifiesClusterListeners() { + final List notifiedInstances = new ArrayList<>(); + + // Create a mock SchedulerSignaler that records calls + SchedulerSignaler signaler = new SchedulerSignaler() { + @Override + public void notifyTriggerListenersMisfired(org.quartz.Trigger trigger) { + } + + @Override + public void notifySchedulerListenersFinalized(org.quartz.Trigger trigger) { + } + + @Override + public void notifySchedulerListenersJobDeleted(org.quartz.JobKey jobKey) { + } + + @Override + public void signalSchedulingChange(long candidateNewNextFireTime) { + } + + @Override + public void notifySchedulerListenersError(String msg, org.quartz.SchedulerException jpe) { + } + + @Override + public void notifyClusterListenersNodeFailed(String failedInstanceId) { + notifiedInstances.add(failedInstanceId); + } + }; + + // Test notification through signaler + signaler.notifyClusterListenersNodeFailed("failedNode1"); + signaler.notifyClusterListenersNodeFailed("failedNode2"); + + assertEquals(2, notifiedInstances.size()); + assertEquals("failedNode1", notifiedInstances.get(0)); + assertEquals("failedNode2", notifiedInstances.get(1)); + } + + /** + * Test that a cluster listener can be notified of multiple node failures + */ + @Test + void testClusterListenerMultipleNotifications() { + TestClusterListener listener = new TestClusterListener("testListener"); + + // Simulate multiple node failures + String[] failedNodes = {"node1", "node2", "node3", "node1"}; + for (String node : failedNodes) { + listener.clusterNodeFailed(node); + } + + assertEquals(4, listener.getFailedInstances().size()); + assertEquals("node1", listener.getFailedInstances().get(0)); + assertEquals("node2", listener.getFailedInstances().get(1)); + assertEquals("node3", listener.getFailedInstances().get(2)); + assertEquals("node1", listener.getFailedInstances().get(3)); // Same node can fail multiple times + } + + /** + * Test that multiple cluster listeners all receive notifications + */ + @Test + void testMultipleClusterListenersNotified() { + final List listener1Notifications = new ArrayList<>(); + final List listener2Notifications = new ArrayList<>(); + final List listener3Notifications = new ArrayList<>(); + + ClusterListener listener1 = new ClusterListener() { + @Override + public String getName() { return "listener1"; } + @Override + public void clusterNodeFailed(String failedInstanceId) { + listener1Notifications.add(failedInstanceId); + } + }; + + ClusterListener listener2 = new ClusterListener() { + @Override + public String getName() { return "listener2"; } + @Override + public void clusterNodeFailed(String failedInstanceId) { + listener2Notifications.add(failedInstanceId); + } + }; + + ClusterListener listener3 = new ClusterListener() { + @Override + public String getName() { return "listener3"; } + @Override + public void clusterNodeFailed(String failedInstanceId) { + listener3Notifications.add(failedInstanceId); + } + }; + + // Simulate signaler notifying all listeners + String failedNode = "failedClusterNode"; + listener1.clusterNodeFailed(failedNode); + listener2.clusterNodeFailed(failedNode); + listener3.clusterNodeFailed(failedNode); + + assertEquals(1, listener1Notifications.size()); + assertEquals(failedNode, listener1Notifications.get(0)); + + assertEquals(1, listener2Notifications.size()); + assertEquals(failedNode, listener2Notifications.get(0)); + + assertEquals(1, listener3Notifications.size()); + assertEquals(failedNode, listener3Notifications.get(0)); + } + + /** + * Test that the cluster listener interface contract is properly defined + */ + @Test + void testClusterListenerInterfaceContract() { + // Test that getName() returns the expected value + ClusterListener listener = new ClusterListener() { + @Override + public String getName() { + return "myTestListener"; + } + + @Override + public void clusterNodeFailed(String failedInstanceId) { + } + }; + + assertEquals("myTestListener", listener.getName()); + } + + /** + * Test concurrent access to ClusterListener (simulating real cluster scenario) + */ + @Test + void testConcurrentNotifications() throws InterruptedException { + final TestClusterListener listener = new TestClusterListener("concurrentTest"); + final int numThreads = 10; + final int notificationsPerThread = 100; + + Thread[] threads = new Thread[numThreads]; + for (int i = 0; i < numThreads; i++) { + final int threadNum = i; + threads[i] = new Thread(() -> { + for (int j = 0; j < notificationsPerThread; j++) { + listener.clusterNodeFailed("node-" + threadNum + "-" + j); + } + }); + } + + // Start all threads + for (Thread thread : threads) { + thread.start(); + } + + // Wait for all threads to complete + for (Thread thread : threads) { + thread.join(); + } + + // Verify all notifications were received + assertEquals(numThreads * notificationsPerThread, listener.getFailedInstances().size()); + } +}