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 cdf5537 SLING-13169 - Fix JobManager readiness condition to preserve
topology on transient probe failures (#54)
cdf5537 is described below
commit cdf5537bc6fc3af808355274397c34a38458de93
Author: Daniel Iancu <[email protected]>
AuthorDate: Thu Jun 25 15:54:05 2026 +0300
SLING-13169 - Fix JobManager readiness condition to preserve topology on
transient probe failures (#54)
* SLING-13169 - Fix Sling JobManager readiness condition to preserve
topology on transient probe failures
Remove stopProcessing() from unbindJobProcessingEnabledCondition() so that
topology state survives readiness condition removal. Update
notifyListeners()
and addListener() to send combined state (topology AND readiness) ensuring
all listeners — including JobSchedulerImpl — correctly stop when the
readiness condition is absent.
Co-authored-by: Daniel Iancu <[email protected]>
---
.../impl/jobs/config/JobManagerConfiguration.java | 46 ++++--
.../jobs/config/JobManagerConfigurationTest.java | 169 ++++++++++++++++++++-
2 files changed, 203 insertions(+), 12 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 8f2ccd5..638d74c 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
@@ -222,10 +222,25 @@ public class JobManagerConfiguration {
/** Is this still active? */
private final AtomicBoolean active = new AtomicBoolean(false);
- /** The topology capabilities. */
+ /**
+ * The current cluster topology capabilities, or {@code null} if no
topology event has been
+ * received yet (or the topology has been explicitly invalidated by a
topology change event).
+ *
+ * <p>SLING-12743: A non-null value means "we know the cluster layout" —
it does NOT by itself
+ * mean that job processing is active. The readiness condition ({@link
#jobProcessingEnabledCondition})
+ * must also be present. Use {@link #isProcessingActive()} for the
combined check.
+ *
+ * <p>This field is only set to {@code null} by {@link #stopProcessing()},
which is called
+ * during real topology changes (TOPOLOGY_CHANGING). It is deliberately
NOT nulled when the
+ * readiness condition is unbound, so that topology state is preserved for
automatic recovery
+ * when the condition returns.
+ */
private volatile TopologyCapabilities topologyCapabilities;
- /** The condition that determines if job processing is enabled. */
+ /**
+ * External readiness signal (e.g. from a Kubernetes readiness probe via
an OSGi Condition
+ * service). When {@code null}, job processing is paused but the topology
state is preserved.
+ */
private volatile Condition jobProcessingEnabledCondition;
/**
@@ -252,14 +267,17 @@ public class JobManagerConfiguration {
/**
* Handle unbinding of the job processing condition.
+ * <p>
+ * SLING-12743: Do NOT call stopProcessing() here — that destroys topology
state and prevents
+ * recovery when the condition returns (e.g. after a transient readiness
probe blip).
+ * Instead, notifyListeners() propagates the combined state (topology AND
readiness) to all
+ * listeners, allowing automatic recovery when the condition is rebound.
* @param condition The condition being unbound
*/
protected void unbindJobProcessingEnabledCondition(final Condition
condition) {
if (this.jobProcessingEnabledCondition == condition) {
this.jobProcessingEnabledCondition = null;
logger.info("Job processing readiness condition has been removed -
jobs will not be processed");
- // Signal jobs to stop before notifying listeners
- stopProcessing();
notifyListeners();
}
}
@@ -411,8 +429,11 @@ public class JobManagerConfiguration {
}
/**
- * Get the current topology capabilities.
- * @return The capabilities or {@code null}
+ * Get the current topology capabilities — the cluster layout (instances,
leaders, topic
+ * assignments). A non-null return value means the cluster layout is known
but does NOT imply
+ * that job processing is active; check {@link #isProcessingActive()} for
that.
+ *
+ * @return The capabilities, or {@code null} if no topology has been
established yet
*/
public TopologyCapabilities getTopologyCapabilities() {
return this.topologyCapabilities;
@@ -645,14 +666,21 @@ public class JobManagerConfiguration {
logger.debug("Job processing started");
}
+ /**
+ * Processing is active: topology must be present AND job processing
+ * must be enabled (readiness condition bound).
+ */
+ private boolean isProcessingActive() {
+ return this.topologyCapabilities != null && isJobProcessingEnabled();
+ }
+
/**
* Notify all listeners
*/
private void notifyListeners() {
synchronized (this.listeners) {
- final TopologyCapabilities caps = this.topologyCapabilities;
for (final ConfigurationChangeListener l : this.listeners) {
- l.configurationChanged(caps != null);
+ l.configurationChanged(isProcessingActive());
}
}
}
@@ -710,7 +738,7 @@ public class JobManagerConfiguration {
public void addListener(final ConfigurationChangeListener service) {
synchronized (this.listeners) {
this.listeners.add(service);
- service.configurationChanged(this.topologyCapabilities != null);
+ service.configurationChanged(isProcessingActive());
}
}
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 05e40ed..5b150bd 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
@@ -36,9 +36,7 @@ import org.junit.Test;
import org.mockito.Mockito;
import org.osgi.service.condition.Condition;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.*;
import static org.mockito.Mockito.mock;
public class JobManagerConfigurationTest {
@@ -100,6 +98,171 @@ public class JobManagerConfigurationTest {
return config;
}
+ @Test
+ public void testConditionTogglePreservesTopology() throws Exception {
+ final ChangeListener ccl = new ChangeListener();
+ final JobManagerConfiguration config = new JobManagerConfiguration();
+ ((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);
+
+ // Bind condition and register listener
+ Condition condition = mock(Condition.class);
+ config.bindJobProcessingEnabledCondition(condition);
+
+ ccl.init(1);
+ config.addListener(ccl);
+ ccl.await();
+ // No topology yet → active=false
+ assertEquals(1, ccl.events.size());
+ assertFalse(ccl.events.get(0));
+
+ // Establish topology
+ ccl.init(1);
+ final TopologyView initView = createView();
+ config.handleTopologyEvent(new
TopologyEvent(TopologyEvent.Type.TOPOLOGY_INIT, null, initView));
+ ccl.await();
+ assertEquals(1, ccl.events.size());
+ assertTrue("Listener should be active with topology + condition",
ccl.events.get(0));
+
+ // Unbind condition — topology must be preserved, listener gets
active=false
+ ccl.init(1);
+ config.unbindJobProcessingEnabledCondition(condition);
+ ccl.await();
+ assertEquals(1, ccl.events.size());
+ assertFalse("Listener should be inactive when condition is removed",
ccl.events.get(0));
+ assertFalse("isJobProcessingEnabled should be false",
config.isJobProcessingEnabled());
+ // Topology must still be present (this is the core fix)
+ assertNotNull("Topology must be preserved after condition unbind",
config.getTopologyCapabilities());
+
+ // Rebind condition — listener should become active again without a
topology event
+ ccl.init(1);
+ config.bindJobProcessingEnabledCondition(condition);
+ ccl.await();
+ assertEquals(1, ccl.events.size());
+ assertTrue("Listener should be active again after condition rebind",
ccl.events.get(0));
+ assertTrue("isJobProcessingEnabled should be true",
config.isJobProcessingEnabled());
+ }
+
+ /**
+ * Test that addListener() sends the combined state (topology AND
readiness),
+ * not just topology-only. This covers the JobSchedulerImpl startup path.
+ */
+ @Test
+ public void testAddListenerSendsCombinedState() throws Exception {
+ final JobManagerConfiguration config = new JobManagerConfiguration();
+ ((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);
+
+ // Establish topology WITHOUT condition — use a helper listener to
await async init
+ final ChangeListener setupListener = new ChangeListener();
+ setupListener.init(1);
+ config.addListener(setupListener);
+ setupListener.await(); // addListener sends initial state (false — no
topology yet)
+
+ final TopologyView initView = createView();
+ setupListener.init(1);
+ config.handleTopologyEvent(new
TopologyEvent(TopologyEvent.Type.TOPOLOGY_INIT, null, initView));
+ setupListener.await(); // topology is now established
+
+ // Topology exists but condition is absent — addListener must send
active=false
+ final ChangeListener listener1 = new ChangeListener();
+ listener1.init(1);
+ config.addListener(listener1);
+ listener1.await();
+ assertEquals(1, listener1.events.size());
+ assertFalse(
+ "addListener must send false when topology exists but
condition is absent", listener1.events.get(0));
+
+ // Now bind condition — addListener must send active=true
+ Condition condition = mock(Condition.class);
+ config.bindJobProcessingEnabledCondition(condition);
+
+ final ChangeListener listener2 = new ChangeListener();
+ listener2.init(1);
+ config.addListener(listener2);
+ listener2.await();
+ assertEquals(1, listener2.events.size());
+ assertTrue("addListener must send true when both topology and
condition are present", listener2.events.get(0));
+
+ // Unbind condition — addListener must send active=false again
+ config.unbindJobProcessingEnabledCondition(condition);
+
+ final ChangeListener listener3 = new ChangeListener();
+ listener3.init(1);
+ config.addListener(listener3);
+ listener3.await();
+ assertEquals(1, listener3.events.size());
+ assertFalse("addListener must send false after condition is unbound",
listener3.events.get(0));
+ }
+
+ /**
+ * Test rapid condition toggling with topology present — verifies that
+ * the final state is consistent after multiple toggles.
+ */
+ @Test
+ public void testRapidConditionToggle() throws Exception {
+ final ChangeListener ccl = new ChangeListener();
+ final JobManagerConfiguration config = new JobManagerConfiguration();
+ ((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);
+
+ // Bind condition, register listener (init latch before addListener),
establish topology
+ Condition condition = mock(Condition.class);
+ config.bindJobProcessingEnabledCondition(condition);
+ ccl.init(1);
+ config.addListener(ccl);
+ ccl.await(); // initial state (no topology yet → false)
+
+ ccl.init(1);
+ final TopologyView initView = createView();
+ config.handleTopologyEvent(new
TopologyEvent(TopologyEvent.Type.TOPOLOGY_INIT, null, initView));
+ ccl.await(); // topology established → true
+
+ // Rapid toggle 5 times — each fires notifyListeners, ccl absorbs them
+ ccl.init(10); // 5 unbinds + 5 binds
+ for (int i = 0; i < 5; i++) {
+ config.unbindJobProcessingEnabledCondition(condition);
+ config.bindJobProcessingEnabledCondition(condition);
+ }
+ ccl.await();
+
+ // Final state: condition is bound, topology is preserved
+ assertTrue("isJobProcessingEnabled should be true after toggles",
config.isJobProcessingEnabled());
+ assertNotNull("Topology must survive rapid toggles",
config.getTopologyCapabilities());
+
+ // Verify listener gets correct state on explicit toggles
+ ccl.init(1);
+ config.unbindJobProcessingEnabledCondition(condition);
+ ccl.await();
+ assertFalse("Last event should be false after final unbind",
ccl.events.get(0));
+
+ ccl.init(1);
+ config.bindJobProcessingEnabledCondition(condition);
+ ccl.await();
+ assertTrue("Last event should be true after final rebind",
ccl.events.get(0));
+ }
+
@Test
public void testTopologyChange() throws Exception {
final ChangeListener ccl = new ChangeListener();
