This is an automated email from the ASF dual-hosted git repository.
joerghoh pushed a commit to branch master
in repository
https://gitbox.apache.org/repos/asf/sling-org-apache-sling-event.git
The following commit(s) were added to refs/heads/master by this push:
new 10f5854 SLING-13170 Replace Timer with injectable
ScheduledExecutorService (#55)
10f5854 is described below
commit 10f58542032116887f6fe0c4cf61918a122a0358
Author: Daniel Iancu <[email protected]>
AuthorDate: Wed Jun 24 15:18:31 2026 +0300
SLING-13170 Replace Timer with injectable ScheduledExecutorService (#55)
Replace the fire-and-forget Timer in startProcessing() with a
ScheduledExecutorService that is properly shut down on deactivate
and can be replaced in tests via setScheduler().
This eliminates the flaky Thread.sleep(4000) in testTopologyChange
by using a ManualScheduler that captures tasks for on-demand execution,
making the test deterministic and ~8x faster.
Co-authored-by: Daniel Iancu <[email protected]>
---
.../impl/jobs/config/JobManagerConfiguration.java | 44 +++--
.../jobs/config/JobManagerConfigurationTest.java | 198 ++++++++++++++++++---
.../event/impl/jobs/config/ManualScheduler.java | 191 ++++++++++++++++++++
3 files changed, 398 insertions(+), 35 deletions(-)
diff --git
a/src/main/java/org/apache/sling/event/impl/jobs/config/JobManagerConfiguration.java
b/src/main/java/org/apache/sling/event/impl/jobs/config/JobManagerConfiguration.java
index 1cba173..8f2ccd5 100644
---
a/src/main/java/org/apache/sling/event/impl/jobs/config/JobManagerConfiguration.java
+++
b/src/main/java/org/apache/sling/event/impl/jobs/config/JobManagerConfiguration.java
@@ -23,10 +23,12 @@ import java.util.Calendar;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Timer;
-import java.util.TimerTask;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.sling.api.resource.LoginException;
import org.apache.sling.api.resource.PersistenceException;
@@ -172,6 +174,14 @@ public class JobManagerConfiguration {
private volatile long startupDelay;
+ /** Scheduler for delayed topology change processing. Replaceable in
tests. */
+ private final AtomicReference<ScheduledExecutorService> scheduler = new
AtomicReference<>();
+
+ /** Package-private for testing only. */
+ void setScheduler(ScheduledExecutorService s) {
+ this.scheduler.set(s);
+ }
+
/**
* The max count of job progress log messages
*/
@@ -307,6 +317,7 @@ public class JobManagerConfiguration {
resolver.close();
}
this.active.set(true);
+ this.scheduler.compareAndSet(null,
Executors.newSingleThreadScheduledExecutor());
// SLING-5560 : use an InitDelayingTopologyEventListener
if (this.startupDelay > 0) {
@@ -357,6 +368,11 @@ public class JobManagerConfiguration {
this.startupDelayListener.dispose();
this.startupDelayListener = null;
}
+ // this only applies to tests
+ final ScheduledExecutorService s = this.scheduler.getAndSet(null);
+ if (s != null) {
+ s.shutdownNow();
+ }
this.stopProcessing();
}
@@ -604,23 +620,27 @@ public class JobManagerConfiguration {
} else {
// and run checker again in some seconds (if leader)
// notify listeners afterwards
- final Timer timer = new Timer();
- timer.schedule(
- new TimerTask() {
-
- @Override
- public void run() {
+ final ScheduledExecutorService s = this.scheduler.get();
+ if (s == null) {
+ logger.debug("Scheduler already shut down — skipping delayed
listener notification");
+ return;
+ }
+ try {
+ s.schedule(
+ () -> {
if (newCaps == topologyCapabilities &&
newCaps.isActive()) {
- // start listeners
notifyListeners();
if (newCaps.isLeader() && newCaps.isActive()) {
final CheckTopologyTask mt = new
CheckTopologyTask(JobManagerConfiguration.this);
mt.fullRun();
}
}
- }
- },
- this.backgroundLoadDelay * 1000);
+ },
+ this.backgroundLoadDelay,
+ TimeUnit.SECONDS);
+ } catch (final java.util.concurrent.RejectedExecutionException e) {
+ logger.debug("Scheduler shut down before task could be
submitted — skipping delayed notification", e);
+ }
}
logger.debug("Job processing started");
}
diff --git
a/src/test/java/org/apache/sling/event/impl/jobs/config/JobManagerConfigurationTest.java
b/src/test/java/org/apache/sling/event/impl/jobs/config/JobManagerConfigurationTest.java
index 23a08db..05e40ed 100644
---
a/src/test/java/org/apache/sling/event/impl/jobs/config/JobManagerConfigurationTest.java
+++
b/src/test/java/org/apache/sling/event/impl/jobs/config/JobManagerConfigurationTest.java
@@ -84,15 +84,10 @@ public class JobManagerConfigurationTest {
}
}
- @Test
- public void testTopologyChange() throws Exception {
- // mock scheduler
- final ChangeListener ccl = new ChangeListener();
-
- // add change listener and verify
- ccl.init(1);
+ private JobManagerConfiguration createConfig(ManualScheduler
manualScheduler) {
final JobManagerConfiguration config = new JobManagerConfiguration();
((AtomicBoolean) TestUtil.getFieldValue(config, "active")).set(true);
+ config.setScheduler(manualScheduler);
InitDelayingTopologyEventListener startupDelayListener =
new InitDelayingTopologyEventListener(1, new
TopologyEventListener() {
@@ -102,46 +97,203 @@ public class JobManagerConfigurationTest {
}
});
TestUtil.setFieldValue(config, "startupDelayListener",
startupDelayListener);
+ return config;
+ }
+
+ @Test
+ public void testTopologyChange() throws Exception {
+ final ChangeListener ccl = new ChangeListener();
+ final ManualScheduler manualScheduler = new ManualScheduler();
+ final JobManagerConfiguration config = createConfig(manualScheduler);
// Create and bind the condition
Condition condition = mock(Condition.class);
config.bindJobProcessingEnabledCondition(condition);
+ ccl.init(1);
config.addListener(ccl);
ccl.await();
assertEquals(1, ccl.events.size());
assertFalse(ccl.events.get(0));
- // create init view
+ // TOPOLOGY_INIT notifies listeners synchronously (no scheduler
involved)
ccl.init(1);
final TopologyView initView = createView();
- final TopologyEvent init = new
TopologyEvent(TopologyEvent.Type.TOPOLOGY_INIT, null, initView);
- config.handleTopologyEvent(init);
+ config.handleTopologyEvent(new
TopologyEvent(TopologyEvent.Type.TOPOLOGY_INIT, null, initView));
ccl.await();
assertEquals(1, ccl.events.size());
assertTrue(ccl.events.get(0));
- // change view, followed by change props
- ccl.init(2);
+ // TOPOLOGY_CHANGED: stopProcessing fires synchronous false,
startProcessing schedules delayed task
+ ccl.init(1);
final TopologyView view2 = createView();
Mockito.when(initView.isCurrent()).thenReturn(false);
- final TopologyEvent change1 = new
TopologyEvent(TopologyEvent.Type.TOPOLOGY_CHANGED, initView, view2);
- final TopologyView view3 = createView();
- final TopologyEvent change2 = new
TopologyEvent(TopologyEvent.Type.PROPERTIES_CHANGED, view2, view3);
+ config.handleTopologyEvent(new
TopologyEvent(TopologyEvent.Type.TOPOLOGY_CHANGED, initView, view2));
+ ccl.await();
+
+ assertEquals(1, ccl.events.size());
+ assertFalse("stopProcessing should fire active=false",
ccl.events.get(0));
+ assertEquals("TOPOLOGY_CHANGED should schedule one delayed task", 1,
manualScheduler.pendingCount());
- config.handleTopologyEvent(change1);
+ // PROPERTIES_CHANGED: for views with same capabilities,
stopProcessing is skipped,
+ // only a new delayed task is scheduled
+ final TopologyView view3 = createView();
Mockito.when(view2.isCurrent()).thenReturn(false);
- config.handleTopologyEvent(change2);
+ config.handleTopologyEvent(new
TopologyEvent(TopologyEvent.Type.PROPERTIES_CHANGED, view2, view3));
+
+ assertEquals("Two delayed tasks pending", 2,
manualScheduler.pendingCount());
+ // Execute all pending tasks — only the latest should fire (earlier
caps were deactivated)
+ ccl.init(1);
+ manualScheduler.runAll();
ccl.await();
- assertEquals(2, ccl.events.size());
- assertFalse(ccl.events.get(0));
- assertTrue(ccl.events.get(1));
- // we wait another 4 secs to see if there is no another event
- Thread.sleep(4000);
- assertEquals(2, ccl.events.size());
+ assertEquals(1, ccl.events.size());
+ assertTrue("Delayed task should notify active=true",
ccl.events.get(0));
+ assertEquals("No pending tasks remain", 0,
manualScheduler.pendingCount());
+ }
+
+ /**
+ * Verify that deactivation prevents delayed listener notifications from
firing
+ * and that a schedule/shutdown race does not produce exceptions.
+ */
+ @Test
+ public void testDeactivatePreventsDelayedNotification() throws Exception {
+ final ChangeListener ccl = new ChangeListener();
+ final ManualScheduler manualScheduler = new ManualScheduler();
+ final JobManagerConfiguration config = createConfig(manualScheduler);
+
+ Condition condition = mock(Condition.class);
+ config.bindJobProcessingEnabledCondition(condition);
+
+ ccl.init(1);
+ config.addListener(ccl);
+ ccl.await(); // initial state: no topology → false
+
+ // Establish topology via TOPOLOGY_INIT (synchronous notification)
+ ccl.init(1);
+ final TopologyView initView = createView();
+ config.handleTopologyEvent(new
TopologyEvent(TopologyEvent.Type.TOPOLOGY_INIT, null, initView));
+ ccl.await();
+ assertTrue("Should be active after TOPOLOGY_INIT", ccl.events.get(0));
+
+ // TOPOLOGY_CHANGED queues a delayed task via the scheduler
+ ccl.init(1);
+ final TopologyView view2 = createView();
+ Mockito.when(initView.isCurrent()).thenReturn(false);
+ config.handleTopologyEvent(new
TopologyEvent(TopologyEvent.Type.TOPOLOGY_CHANGED, initView, view2));
+ ccl.await(); // synchronous false from stopProcessing
+ assertEquals("One delayed task should be pending", 1,
manualScheduler.pendingCount());
+
+ // Deactivate before the delayed task runs — this calls shutdownNow()
on the scheduler
+ // (deactivate also calls stopProcessing → notifyListeners which fires
a synchronous event)
+ config.deactivate();
+ ccl.events.clear();
+
+ // shutdownNow() must have cleared the pending task
+ assertEquals("Pending tasks must be cleared after deactivate", 0,
manualScheduler.pendingCount());
+ assertTrue("Scheduler must be shut down",
manualScheduler.isShutdown());
+
+ // Running the scheduler after shutdown must not deliver any events
+ manualScheduler.runAll();
+ assertTrue("No events should be delivered after deactivate",
ccl.events.isEmpty());
+
+ // Simulate the race: a topology event arrives while deactivation is
in progress.
+ // The production code reads scheduler.get() which now returns null
(AtomicReference
+ // was cleared by deactivate). This must not throw NPE.
+ // To also exercise the RejectedExecutionException path, we re-inject
the shut-down
+ // scheduler and trigger a non-INIT event.
+ config.setScheduler(manualScheduler);
+ ((AtomicBoolean) TestUtil.getFieldValue(config, "active")).set(true);
+ InitDelayingTopologyEventListener startupDelayListener =
+ new InitDelayingTopologyEventListener(1, new
TopologyEventListener() {
+
+ @Override
+ public void handleTopologyEvent(TopologyEvent event) {
+ config.doHandleTopologyEvent(event);
+ }
+ });
+ TestUtil.setFieldValue(config, "startupDelayListener",
startupDelayListener);
+
+ // This TOPOLOGY_INIT sets up topology state so the next
TOPOLOGY_CHANGED reaches startProcessing
+ final TopologyView view3 = createView();
+ config.handleTopologyEvent(new
TopologyEvent(TopologyEvent.Type.TOPOLOGY_INIT, null, view3));
+
+ // Now trigger TOPOLOGY_CHANGED — scheduler.schedule() will throw
RejectedExecutionException
+ // which must be caught by the production code, not escape to the
caller
+ ccl.events.clear();
+ final TopologyView view4 = createView();
+ Mockito.when(view3.isCurrent()).thenReturn(false);
+ config.handleTopologyEvent(new
TopologyEvent(TopologyEvent.Type.TOPOLOGY_CHANGED, view3, view4));
+
+ // No task should have been queued (scheduler rejected it)
+ assertEquals("No tasks should be queued on shut-down scheduler", 0,
manualScheduler.pendingCount());
+ }
+
+ /**
+ * Verify that startProcessing() handles a null scheduler reference
gracefully.
+ * This covers the race where deactivate() has already cleared the
AtomicReference
+ * (via getAndSet(null)) before a topology event reaches startProcessing().
+ */
+ @Test
+ public void testStartProcessingWithNullScheduler() throws Exception {
+ final ChangeListener ccl = new ChangeListener();
+ final ManualScheduler manualScheduler = new ManualScheduler();
+ final JobManagerConfiguration config = createConfig(manualScheduler);
+
+ Condition condition = mock(Condition.class);
+ config.bindJobProcessingEnabledCondition(condition);
+
+ ccl.init(1);
+ config.addListener(ccl);
+ ccl.await();
+
+ // Establish topology via TOPOLOGY_INIT
+ ccl.init(1);
+ final TopologyView initView = createView();
+ config.handleTopologyEvent(new
TopologyEvent(TopologyEvent.Type.TOPOLOGY_INIT, null, initView));
+ ccl.await();
+ assertTrue(ccl.events.get(0));
+
+ // Clear the scheduler reference directly — simulates deactivate()
having already
+ // called scheduler.getAndSet(null) before the topology event arrives
+ config.setScheduler(null);
+
+ // TOPOLOGY_CHANGED triggers stopProcessing (synchronous false) then
startProcessing
+ // which reads scheduler.get() → null and must return early without
throwing NPE
+ ccl.init(1);
+ final TopologyView view2 = createView();
+ Mockito.when(initView.isCurrent()).thenReturn(false);
+ config.handleTopologyEvent(new
TopologyEvent(TopologyEvent.Type.TOPOLOGY_CHANGED, initView, view2));
+ ccl.await();
+
+ // Only the synchronous false from stopProcessing should have been
delivered
+ assertEquals(1, ccl.events.size());
+ assertFalse("Only stopProcessing event should fire",
ccl.events.get(0));
+ // No task queued anywhere — scheduler was null
+ assertEquals("No tasks should be pending", 0,
manualScheduler.pendingCount());
+ }
+
+ /**
+ * Verify that deactivate() handles a null scheduler reference gracefully.
+ * This covers the case where the scheduler was never initialized (e.g.
activate()
+ * was never called) or was already cleared by a prior deactivate().
+ */
+ @Test
+ public void testDeactivateWithNullScheduler() {
+ final JobManagerConfiguration config = new JobManagerConfiguration();
+ ((AtomicBoolean) TestUtil.getFieldValue(config, "active")).set(true);
+ // Do NOT set a scheduler — the AtomicReference holds null
+
+ // deactivate() must not throw NPE when scheduler.getAndSet(null)
returns null
+ config.deactivate();
+
+ assertFalse("Component should be inactive after deactivate",
config.isActive());
+
+ // Calling deactivate() a second time must also be safe
(double-deactivate)
+ config.deactivate();
+ assertFalse("Component should remain inactive", config.isActive());
}
}
diff --git
a/src/test/java/org/apache/sling/event/impl/jobs/config/ManualScheduler.java
b/src/test/java/org/apache/sling/event/impl/jobs/config/ManualScheduler.java
new file mode 100644
index 0000000..e27bf7c
--- /dev/null
+++ b/src/test/java/org/apache/sling/event/impl/jobs/config/ManualScheduler.java
@@ -0,0 +1,191 @@
+/*
+ * 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.sling.event.impl.jobs.config;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.*;
+
+/**
+ * A ScheduledExecutorService that captures submitted tasks without executing
them.
+ * Models realistic shutdown semantics: {@link #shutdownNow()} marks the
scheduler as shut down
+ * and clears pending tasks; {@link #schedule} after shutdown throws {@link
RejectedExecutionException};
+ * {@link #runAll()} skips execution after shutdown.
+ */
+class ManualScheduler implements ScheduledExecutorService {
+
+ private final List<Runnable> tasks = new ArrayList<>();
+ private boolean shutdown;
+
+ /**
+ * Execute all captured tasks and clear the list. No-op after shutdown.
+ */
+ public void runAll() {
+ if (shutdown) {
+ return;
+ }
+ final List<Runnable> copy = new ArrayList<>(tasks);
+ tasks.clear();
+ copy.forEach(Runnable::run);
+ }
+
+ /**
+ * Return the number of pending (un-executed) tasks.
+ */
+ public int pendingCount() {
+ return tasks.size();
+ }
+
+ @Override
+ public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit
unit) {
+ if (shutdown) {
+ throw new RejectedExecutionException("Scheduler has been shut
down");
+ }
+ tasks.add(command);
+ return new CompletedFuture<>();
+ }
+
+ // -- unused ScheduledExecutorService methods --
+
+ @Override
+ public <V> ScheduledFuture<V> schedule(java.util.concurrent.Callable<V>
callable, long delay, TimeUnit unit) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long
initialDelay, long period, TimeUnit unit) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long
initialDelay, long delay, TimeUnit unit) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void shutdown() {
+ shutdown = true;
+ }
+
+ @Override
+ public List<Runnable> shutdownNow() {
+ shutdown = true;
+ final List<Runnable> pending = new ArrayList<>(tasks);
+ tasks.clear();
+ return pending;
+ }
+
+ @Override
+ public boolean isShutdown() {
+ return shutdown;
+ }
+
+ @Override
+ public boolean isTerminated() {
+ return shutdown;
+ }
+
+ @Override
+ public boolean awaitTermination(long timeout, TimeUnit unit) {
+ return true;
+ }
+
+ @Override
+ public <T> java.util.concurrent.Future<T>
submit(java.util.concurrent.Callable<T> task) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public <T> java.util.concurrent.Future<T> submit(Runnable task, T result) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public java.util.concurrent.Future<?> submit(Runnable task) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public <T> List<java.util.concurrent.Future<T>> invokeAll(
+ java.util.Collection<? extends java.util.concurrent.Callable<T>>
tasks) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public <T> List<java.util.concurrent.Future<T>> invokeAll(
+ java.util.Collection<? extends java.util.concurrent.Callable<T>>
tasks, long timeout, TimeUnit unit) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public <T> T invokeAny(java.util.Collection<? extends
java.util.concurrent.Callable<T>> tasks) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public <T> T invokeAny(
+ java.util.Collection<? extends java.util.concurrent.Callable<T>>
tasks, long timeout, TimeUnit unit) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void execute(Runnable command) {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * Minimal no-op ScheduledFuture returned by schedule().
+ */
+ private static class CompletedFuture<V> implements ScheduledFuture<V> {
+ @Override
+ public long getDelay(TimeUnit unit) {
+ return 0;
+ }
+
+ @Override
+ public int compareTo(Delayed o) {
+ return 0;
+ }
+
+ @Override
+ public boolean cancel(boolean mayInterruptIfRunning) {
+ return false;
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return false;
+ }
+
+ @Override
+ public boolean isDone() {
+ return true;
+ }
+
+ @Override
+ public V get() {
+ return null;
+ }
+
+ @Override
+ public V get(long timeout, TimeUnit unit) {
+ return null;
+ }
+ }
+}
