This is an automated email from the ASF dual-hosted git repository. joerghoh pushed a commit to branch SLING-13429 in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-event.git
commit cb68ff26410f68007b89ffc96d8ef36a3397f07f Author: Joerg Hoh <[email protected]> AuthorDate: Fri Nov 13 17:11:40 2026 +0100 SLING-13429 add diagnostics for unassigned jobs --- .../sling/event/impl/jobs/JobManagerImpl.java | 61 ++++++++- .../impl/jobs/config/JobManagerConfiguration.java | 15 +++ .../impl/jobs/config/TopologyCapabilities.java | 12 +- .../sling/event/impl/jobs/JobManagerImplTest.java | 149 +++++++++++++++++++-- .../jobs/config/JobManagerConfigurationTest.java | 21 +++ .../org/apache/sling/event/it/JobHandlingIT.java | 7 +- 6 files changed, 248 insertions(+), 17 deletions(-) diff --git a/src/main/java/org/apache/sling/event/impl/jobs/JobManagerImpl.java b/src/main/java/org/apache/sling/event/impl/jobs/JobManagerImpl.java index df0247c..51581ea 100644 --- a/src/main/java/org/apache/sling/event/impl/jobs/JobManagerImpl.java +++ b/src/main/java/org/apache/sling/event/impl/jobs/JobManagerImpl.java @@ -18,6 +18,7 @@ */ package org.apache.sling.event.impl.jobs; +import java.time.Clock; import java.util.ArrayList; import java.util.Calendar; import java.util.Collection; @@ -25,8 +26,10 @@ import java.util.Dictionary; import java.util.HashMap; import java.util.Hashtable; import java.util.Iterator; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.TimeUnit; import com.codahale.metrics.Gauge; import com.codahale.metrics.MetricRegistry; @@ -96,9 +99,31 @@ public class JobManagerImpl implements JobManager, EventHandler, Runnable { private static final String GAUGE_TOTAL_SCHEDULED_JOBS = "event.scheduledJobs.count"; + private static final long UNASSIGNED_LOG_INTERVAL_MILLIS = TimeUnit.MINUTES.toMillis(5); + + private static final int MAX_UNASSIGNED_LOG_TOPICS = 1000; + /** Default logger. */ private final Logger logger = LoggerFactory.getLogger(this.getClass()); + private Clock clock = Clock.systemUTC(); + + /** + * Remember the last diagnostic time per topic to limit logging to one message every five minutes. + * Guarded by the map's monitor. Eviction bounds memory for dynamic topics; an evicted topic may + * log again before the usual interval expires. + */ + private final Map<String, Long> unassignedLogTimes = new LinkedHashMap<String, Long>(16, 0.75f, true) { + @Override + protected boolean removeEldestEntry(final Map.Entry<String, Long> eldest) { + return size() > MAX_UNASSIGNED_LOG_TOPICS; + } + }; + + void setClock(final Clock clock) { + this.clock = clock; + } + @Reference(policyOption = ReferencePolicyOption.GREEDY) private EventAdmin eventAdmin; @@ -586,10 +611,7 @@ public class JobManagerImpl implements JobManager, EventHandler, Runnable { info.targetId = (caps == null ? null : caps.detectTarget(jobTopic, jobProperties, info)); if (info.targetId == null && logger.isInfoEnabled()) { - logger.info( - "Persisting job {} into queue {} with no assigned target", - Utility.toString(jobTopic, jobProperties), - info.queueName); + logUnassignedJob(jobTopic, info.queueName, caps); } else if (info.targetId != null && logger.isDebugEnabled()) { logger.debug( "Persisting job {} into queue {}, target={}", @@ -618,6 +640,37 @@ public class JobManagerImpl implements JobManager, EventHandler, Runnable { return null; } + /** + * Diagnose the routing decision before persistence, without exposing potentially sensitive job + * properties. + */ + private void logUnassignedJob(final String jobTopic, final String queueName, final TopologyCapabilities caps) { + final long now = this.clock.millis(); + // Log at max once per 5 minute interval and per topic + synchronized (this.unassignedLogTimes) { + final Long lastLogged = this.unassignedLogTimes.get(jobTopic); + if (lastLogged != null && now >= lastLogged && now - lastLogged < UNASSIGNED_LOG_INTERVAL_MILLIS) { + return; + } + this.unassignedLogTimes.put(jobTopic, now); + } + logger.info( + "Persisting job [topic={}] into queue {} with no assigned target; reason={}, instanceId={}, " + + "readiness={}, capabilitiesActive={}, viewCurrent={}, leader={}, capabilitiesChanged={}, " + + "lastTopologyEventTime={}, localTopics={}", + jobTopic, + queueName, + caps == null ? "no-topology" : "no-matching-consumer", + Environment.APPLICATION_ID, + this.configuration.isJobProcessingEnabled(), + caps == null ? null : caps.isActive(), + caps == null ? null : caps.isCurrent(), + caps == null ? null : caps.isLeader(), + caps != this.configuration.getTopologyCapabilities(), + this.configuration.getLastTopologyEventTime(), + this.jobConsumerManager.getTopics()); + } + /** * Write a job to the resource tree. * @param resolver The resolver resolver 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 638d74c..3d4c4bb 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 @@ -237,6 +237,9 @@ public class JobManagerConfiguration { */ private volatile TopologyCapabilities topologyCapabilities; + /** Last successful event-handler completion, distinct from discovery delivery or delayed maintenance. */ + private volatile long lastTopologyEventTime; + /** * 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. @@ -439,6 +442,16 @@ public class JobManagerConfiguration { return this.topologyCapabilities; } + /** + * Get the completion time of the last successfully handled topology event. + * A startup-delayed event is not counted until forwarded and handled; maintenance scheduled + * by that handler may still be pending. + * @return Epoch milliseconds, or zero if no event has completed yet. + */ + public long getLastTopologyEventTime() { + return this.lastTopologyEventTime; + } + public QueueConfigurationManager getQueueConfigurationManager() { return this.queueConfigManager; } @@ -729,6 +742,8 @@ public class JobManagerConfiguration { this.startProcessing(eventType, new TopologyCapabilities(event.getNewView(), this)); } + // Leave the previous timestamp intact if handling fails, rather than suggesting successful progress. + this.lastTopologyEventTime = System.currentTimeMillis(); } /** diff --git a/src/main/java/org/apache/sling/event/impl/jobs/config/TopologyCapabilities.java b/src/main/java/org/apache/sling/event/impl/jobs/config/TopologyCapabilities.java index f338716..86f0b3a 100644 --- a/src/main/java/org/apache/sling/event/impl/jobs/config/TopologyCapabilities.java +++ b/src/main/java/org/apache/sling/event/impl/jobs/config/TopologyCapabilities.java @@ -173,7 +173,17 @@ public class TopologyCapabilities { * @return {@code true} if still active. */ public boolean isActive() { - return this.active && this.jobManagerConfiguration.isActive() && this.view.isCurrent(); + return this.active && this.jobManagerConfiguration.isActive() && this.isCurrent(); + } + + /** + * Whether the discovery view used to build these capabilities is still current. + * This can remain true after local deactivation; unlike {@link #isActive()}, it checks only + * the discovery view so diagnostics can distinguish the two states. + * @return {@code true} if the view is current. + */ + public boolean isCurrent() { + return this.view.isCurrent(); } /** diff --git a/src/test/java/org/apache/sling/event/impl/jobs/JobManagerImplTest.java b/src/test/java/org/apache/sling/event/impl/jobs/JobManagerImplTest.java index 34162d6..fa4e230 100644 --- a/src/test/java/org/apache/sling/event/impl/jobs/JobManagerImplTest.java +++ b/src/test/java/org/apache/sling/event/impl/jobs/JobManagerImplTest.java @@ -22,6 +22,16 @@ import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.PrintStream; import java.lang.reflect.InvocationTargetException; +import java.time.Clock; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.sling.api.resource.Resource; @@ -32,6 +42,7 @@ import org.apache.sling.event.impl.jobs.config.JobManagerConfiguration; import org.apache.sling.event.impl.jobs.config.QueueConfigurationManager; import org.apache.sling.event.impl.jobs.config.QueueConfigurationManager.QueueInfo; import org.apache.sling.event.impl.jobs.config.TopologyCapabilities; +import org.apache.sling.event.jobs.Job; import org.apache.sling.event.jobs.JobManager.QueryType; import org.junit.After; import org.junit.Before; @@ -40,6 +51,7 @@ import org.slf4j.Logger; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; @@ -57,6 +69,8 @@ public class JobManagerImplTest { private PrintStream originalErr; private ByteArrayOutputStream outContent; private JobManagerConfiguration configuration; + private JobConsumerManager consumerManager; + private Clock clock; private static final int LEVEL_DEBUG = 10; private static final int LEVEL_INFO = 20; @@ -73,7 +87,12 @@ public class JobManagerImplTest { QueueConfigurationManager queueConfigMgr = mock(QueueConfigurationManager.class); this.info = new QueueInfo(); info.queueConfiguration = mock(InternalQueueConfiguration.class); - when(queueConfigMgr.getQueueInfo(anyString())).thenReturn(info); + when(queueConfigMgr.getQueueInfo(anyString())).thenAnswer(invocation -> { + QueueInfo result = new QueueInfo(); + result.queueConfiguration = info.queueConfiguration; + result.queueName = info.queueName; + return result; + }); when(configuration.getQueueConfigurationManager()).thenReturn(queueConfigMgr); this.capabilities = mock(TopologyCapabilities.class); @@ -85,6 +104,9 @@ public class JobManagerImplTest { when(resolver.getResource(anyString())).thenReturn(mock(Resource.class)); when(configuration.createResourceResolver()).thenReturn(resolver); + this.consumerManager = mock(JobConsumerManager.class); + this.clock = mock(Clock.class); + this.outContent = new ByteArrayOutputStream(); this.originalErr = System.err; System.setErr(new PrintStream(outContent)); @@ -115,14 +137,12 @@ public class JobManagerImplTest { jobManager.addJob("not/assigned", null, null); assertFalse(TestUtil.containsLine( this.outContent, - line -> line.contains( - "Persisting job Sling Job [topic=not/assigned] into queue null with no assigned target"))); + line -> line.contains("Persisting job [topic=not/assigned] into queue null with no assigned target"))); jobManager.addJob("is/assigned", null, null); assertFalse(TestUtil.containsLine( this.outContent, - line -> line.contains( - "Persisting job Sling Job [topic=is/assigned] into queue null with no assigned target"))); + line -> line.contains("Persisting job [topic=is/assigned] into queue null with no assigned target"))); } @Test @@ -133,14 +153,115 @@ public class JobManagerImplTest { jobManager.addJob("not/assigned", null, null); assertTrue(TestUtil.containsLine( this.outContent, - line -> line.contains( - "Persisting job Sling Job [topic=not/assigned] into queue null with no assigned target"))); + line -> line.contains("Persisting job [topic=not/assigned] into queue null with no assigned target"))); jobManager.addJob("is/assigned", null, null); - assertTrue(TestUtil.containsLine( + assertFalse(TestUtil.containsLine( this.outContent, - line -> !line.contains( - "Persisting job Sling Job [topic=is/assigned] into queue null with no assigned target"))); + line -> line.contains("Persisting job [topic=is/assigned] into queue null with no assigned target"))); + } + + @Test + public void logsMissingTopologyWithoutJobProperties() throws Exception { + JobManagerImpl jobManager = getJobManager(LEVEL_INFO); + when(configuration.getTopologyCapabilities()).thenReturn(null); + when(configuration.isJobProcessingEnabled()).thenReturn(true); + when(configuration.getLastTopologyEventTime()).thenReturn(1234L); + when(consumerManager.getTopics()).thenReturn("common/core/email"); + + assertNotNull( + jobManager.addJob("common/core/email", Collections.singletonMap("body", "private-mail-body"), null)); + + assertTrue(TestUtil.containsLine( + outContent, + line -> line.contains("reason=no-topology") + && line.contains("readiness=true") + && line.contains("capabilitiesActive=null") + && line.contains("viewCurrent=null") + && line.contains("lastTopologyEventTime=1234") + && line.contains("localTopics=common/core/email"))); + assertFalse(outContent.toString().contains("private-mail-body")); + } + + @Test + public void logsMissingConsumerWithCapabilityState() throws Exception { + JobManagerImpl jobManager = getJobManager(LEVEL_INFO); + when(capabilities.isActive()).thenReturn(false); + when(capabilities.isCurrent()).thenReturn(false); + when(capabilities.isLeader()).thenReturn(true); + when(consumerManager.getTopics()).thenReturn("common/core/email,other/**"); + + assertNotNull(jobManager.addJob("common/core/email", null, null)); + + assertTrue(TestUtil.containsLine( + outContent, + line -> line.contains("reason=no-matching-consumer") + && line.contains("readiness=false") + && line.contains("capabilitiesActive=false") + && line.contains("viewCurrent=false") + && line.contains("leader=true") + && line.contains("capabilitiesChanged=false") + && line.contains("localTopics=common/core/email,other/**"))); + } + + @Test + public void logsCapabilityReplacementDuringRouting() throws Exception { + JobManagerImpl jobManager = getJobManager(LEVEL_INFO); + when(configuration.getTopologyCapabilities()).thenReturn(capabilities, mock(TopologyCapabilities.class)); + + jobManager.addJob("not/assigned", null, null); + + assertTrue(TestUtil.containsLine(outContent, line -> line.contains("capabilitiesChanged=true"))); + } + + @Test + public void rateLimitsUnassignedDiagnosticsPerTopic() throws Exception { + JobManagerImpl jobManager = getJobManager(LEVEL_INFO); + assertNotNull(jobManager.addJob("not/assigned", null, null)); + assertEquals(1, countUnassignedMessages()); + + when(clock.millis()).thenReturn(TimeUnit.MINUTES.toMillis(5) - 1); + assertNotNull(jobManager.addJob("not/assigned", null, null)); + assertEquals(1, countUnassignedMessages()); + + assertNotNull(jobManager.addJob("other/topic", null, null)); + assertEquals(2, countUnassignedMessages()); + + when(clock.millis()).thenReturn(TimeUnit.MINUTES.toMillis(5)); + assertNotNull(jobManager.addJob("not/assigned", null, null)); + assertEquals(3, countUnassignedMessages()); + + when(clock.millis()).thenReturn(0L); + assertNotNull(jobManager.addJob("not/assigned", null, null)); + assertEquals(4, countUnassignedMessages()); + } + + @Test + public void rateLimitsConcurrentUnassignedDiagnostics() throws Exception { + JobManagerImpl jobManager = getJobManager(LEVEL_INFO); + ExecutorService executor = Executors.newFixedThreadPool(8); + CountDownLatch ready = new CountDownLatch(8); + CountDownLatch start = new CountDownLatch(1); + List<Future<Job>> submissions = new ArrayList<>(); + try { + for (int i = 0; i < 8; i++) { + submissions.add(executor.submit(() -> { + ready.countDown(); + assertTrue(start.await(5, TimeUnit.SECONDS)); + return jobManager.addJob("not/assigned", null, null); + })); + } + assertTrue(ready.await(5, TimeUnit.SECONDS)); + start.countDown(); + for (Future<Job> submission : submissions) { + assertNotNull(submission.get(5, TimeUnit.SECONDS)); + } + assertEquals(1, countUnassignedMessages()); + } finally { + start.countDown(); + executor.shutdownNow(); + assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + } } @Test @@ -171,6 +292,14 @@ public class JobManagerImplTest { FieldUtils.writeDeclaredField(logger, "currentLogLevel", level, true); FieldUtils.writeDeclaredField(jobManager, "configuration", configuration, true); + FieldUtils.writeDeclaredField(jobManager, "jobConsumerManager", consumerManager, true); + jobManager.setClock(clock); return jobManager; } + + private long countUnassignedMessages() { + return Arrays.stream(outContent.toString().split("\n")) + .filter(line -> line.contains("with no assigned target")) + .count(); + } } 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 5b150bd..8a2e8b7 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 @@ -98,6 +98,27 @@ public class JobManagerConfigurationTest { return config; } + @Test + public void testLastTopologyEventTimeTracksCompletedEvents() { + final JobManagerConfiguration config = Mockito.spy(new JobManagerConfiguration()); + assertEquals(0, config.getLastTopologyEventTime()); + + final long before = System.currentTimeMillis(); + config.doHandleTopologyEvent(new TopologyEvent(TopologyEvent.Type.TOPOLOGY_INIT, null, createView())); + final long completedAt = config.getLastTopologyEventTime(); + assertTrue(completedAt >= before); + assertTrue(completedAt <= System.currentTimeMillis()); + + Mockito.doThrow(new IllegalStateException("resolver unavailable")) + .when(config) + .createResourceResolver(); + assertThrows( + IllegalStateException.class, + () -> config.doHandleTopologyEvent( + new TopologyEvent(TopologyEvent.Type.TOPOLOGY_INIT, null, createView()))); + assertEquals(completedAt, config.getLastTopologyEventTime()); + } + @Test public void testConditionTogglePreservesTopology() throws Exception { final ChangeListener ccl = new ChangeListener(); diff --git a/src/test/java/org/apache/sling/event/it/JobHandlingIT.java b/src/test/java/org/apache/sling/event/it/JobHandlingIT.java index 5144aa3..6cf37c9 100644 --- a/src/test/java/org/apache/sling/event/it/JobHandlingIT.java +++ b/src/test/java/org/apache/sling/event/it/JobHandlingIT.java @@ -443,8 +443,11 @@ public class JobHandlingIT extends AbstractJobHandlingIT { assertEquals(JobState.QUEUED, job.getJobState()); Predicate<String> matcher = (line) -> line.contains( - "Persisting job Sling Job [topic=no/consumer/registered] into queue <main queue> with no assigned target"); - assertTrue(TestUtil.containsLine(outContent, matcher)); + "Persisting job [topic=no/consumer/registered] into queue <main queue> with no assigned target") + && line.contains("reason=no-matching-consumer"); + assertTrue( + "Missing unassigned-job diagnostic. Captured output:\n" + outContent, + TestUtil.containsLine(outContent, matcher)); } finally { System.setOut(originalOut);
