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-distribution-core.git
The following commit(s) were added to refs/heads/master by this push:
new c99143ab SLING-13329 : Cap Distribution Queue status count at 1,000
items
c99143ab is described below
commit c99143ab4fe45bbcbcafddb27251609e8e10dda4
Author: aamirbhat <[email protected]>
AuthorDate: Fri Nov 13 18:12:14 2026 +0530
SLING-13329 : Cap Distribution Queue status count at 1,000 items
Co-authored-by: aamirh <[email protected]>
---
.../queue/DistributionQueueStatus.java | 16 +++++
.../queue/impl/SimpleAgentDistributionQueue.java | 3 +-
.../queue/impl/resource/ActiveResourceQueue.java | 24 +++++--
.../queue/impl/resource/ResourceQueue.java | 23 ++++++-
.../queue/impl/resource/ResourceQueueProvider.java | 17 ++++-
.../resource/ResourceQueueProviderFactory.java | 13 +++-
.../queue/impl/resource/ResourceQueueUtils.java | 52 +++++++++++++++
...xtendedDistributionServiceResourceProvider.java | 1 +
.../impl/resource/ResourceQueueUtilsTest.java | 75 ++++++++++++++++++++++
9 files changed, 214 insertions(+), 10 deletions(-)
diff --git
a/src/main/java/org/apache/sling/distribution/queue/DistributionQueueStatus.java
b/src/main/java/org/apache/sling/distribution/queue/DistributionQueueStatus.java
index 7f59df7d..bb8b9390 100644
---
a/src/main/java/org/apache/sling/distribution/queue/DistributionQueueStatus.java
+++
b/src/main/java/org/apache/sling/distribution/queue/DistributionQueueStatus.java
@@ -27,11 +27,19 @@ import org.jetbrains.annotations.NotNull;
public final class DistributionQueueStatus {
private final int itemsCount;
+ private final boolean itemsCountCapped;
+
private final DistributionQueueState queueState;
public DistributionQueueStatus(int itemsCount, @NotNull
DistributionQueueState queueState) {
+ this(itemsCount, queueState, false);
+ }
+
+ public DistributionQueueStatus(
+ int itemsCount, @NotNull DistributionQueueState queueState,
boolean itemsCountCapped) {
this.itemsCount = itemsCount;
this.queueState = queueState;
+ this.itemsCountCapped = itemsCountCapped;
}
/**
@@ -42,6 +50,14 @@ public final class DistributionQueueStatus {
return itemsCount;
}
+ /**
+ * Indicates whether the reported item count is capped rather than exact.
+ * @return {@code true} if the reported item count is capped
+ */
+ public boolean isItemsCountCapped() {
+ return itemsCountCapped;
+ }
+
/**
* get the state associated with the status of this queue
* @return a {@link DistributionQueueItemState}
diff --git
a/src/main/java/org/apache/sling/distribution/queue/impl/SimpleAgentDistributionQueue.java
b/src/main/java/org/apache/sling/distribution/queue/impl/SimpleAgentDistributionQueue.java
index c297c8d3..6f6418ea 100644
---
a/src/main/java/org/apache/sling/distribution/queue/impl/SimpleAgentDistributionQueue.java
+++
b/src/main/java/org/apache/sling/distribution/queue/impl/SimpleAgentDistributionQueue.java
@@ -54,7 +54,8 @@ public class SimpleAgentDistributionQueue extends
DistributionQueueWrapper {
private DistributionQueueStatus calculateStatus() {
DistributionQueueStatus status = super.getStatus();
if (isPaused) {
- return new DistributionQueueStatus(status.getItemsCount(),
DistributionQueueState.PAUSED);
+ return new DistributionQueueStatus(
+ status.getItemsCount(), DistributionQueueState.PAUSED,
status.isItemsCountCapped());
} else {
return status;
}
diff --git
a/src/main/java/org/apache/sling/distribution/queue/impl/resource/ActiveResourceQueue.java
b/src/main/java/org/apache/sling/distribution/queue/impl/resource/ActiveResourceQueue.java
index bb8fe5b2..33374c7d 100644
---
a/src/main/java/org/apache/sling/distribution/queue/impl/resource/ActiveResourceQueue.java
+++
b/src/main/java/org/apache/sling/distribution/queue/impl/resource/ActiveResourceQueue.java
@@ -45,7 +45,16 @@ public class ActiveResourceQueue extends ResourceQueue {
public ActiveResourceQueue(
ResourceResolverFactory resolverFactory, String serviceName,
String queueName, String rootPath) {
- super(resolverFactory, serviceName, queueName, rootPath);
+ this(resolverFactory, serviceName, queueName, rootPath, false);
+ }
+
+ public ActiveResourceQueue(
+ ResourceResolverFactory resolverFactory,
+ String serviceName,
+ String queueName,
+ String rootPath,
+ boolean useExactQueueLength) {
+ super(resolverFactory, serviceName, queueName, rootPath,
useExactQueueLength);
}
@NotNull
@@ -56,14 +65,21 @@ public class ActiveResourceQueue extends ResourceQueue {
resourceResolver = DistributionUtils.loginService(resolverFactory,
serviceName);
Resource queueRoot =
ResourceQueueUtils.getRootResource(resourceResolver, queueRootPath);
- int count = ResourceQueueUtils.getResourceCount(queueRoot);
+ ResourceQueueUtils.ResourceCountResult countResult =
useExactQueueLength
+ ? new
ResourceQueueUtils.ResourceCountResult(ResourceQueueUtils.getResourceCount(queueRoot),
false)
+ :
ResourceQueueUtils.getResourceCountCappedResult(queueRoot);
DistributionQueueEntry head =
ResourceQueueUtils.getHead(queueRoot);
DistributionQueueItem firstItem = (null != head) ? head.getItem()
: null;
DistributionQueueItemStatus firstItemStatus = (null != head) ?
head.getStatus() : null;
- log.debug("Queue has {} items, with following status for the head:
{}", count, firstItemStatus);
+ log.debug(
+ "Queue has {} items, with following status for the head:
{}",
+ countResult.getCount(),
+ firstItemStatus);
return new DistributionQueueStatus(
- count, DistributionQueueUtils.calculateState(firstItem,
firstItemStatus));
+ countResult.getCount(),
+ DistributionQueueUtils.calculateState(firstItem,
firstItemStatus),
+ countResult.isCapped());
} catch (LoginException | PersistenceException e) {
throw new RuntimeException(e);
} finally {
diff --git
a/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueue.java
b/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueue.java
index 11bd698b..506b47e5 100644
---
a/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueue.java
+++
b/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueue.java
@@ -57,13 +57,29 @@ public class ResourceQueue implements DistributionQueue {
protected final String queueRootPath;
protected String serviceName;
protected String queueName;
+ protected final boolean useExactQueueLength;
public ResourceQueue(
ResourceResolverFactory resolverFactory, String serviceName,
String queueName, String rootPath) {
+ this(resolverFactory, serviceName, queueName, rootPath, false);
+ }
+
+ /**
+ * @param useExactQueueLength if true, getStatus() reports the exact
(uncapped) queue length instead
+ * of the capped approximation. Backed by the
resourceQueue provider's
+ * "status.useExactQueueLength" OSGi
configuration.
+ */
+ public ResourceQueue(
+ ResourceResolverFactory resolverFactory,
+ String serviceName,
+ String queueName,
+ String rootPath,
+ boolean useExactQueueLength) {
this.resolverFactory = resolverFactory;
this.serviceName = serviceName;
this.queueName = queueName;
this.queueRootPath = rootPath + "/" + queueName;
+ this.useExactQueueLength = useExactQueueLength;
log.debug("starting a Resource Queue {}", queueName);
}
@@ -217,9 +233,12 @@ public class ResourceQueue implements DistributionQueue {
resourceResolver = DistributionUtils.loginService(resolverFactory,
serviceName);
Resource queueRoot =
ResourceQueueUtils.getRootResource(resourceResolver, queueRootPath);
- int count = ResourceQueueUtils.getResourceCount(queueRoot);
+ ResourceQueueUtils.ResourceCountResult countResult =
useExactQueueLength
+ ? new
ResourceQueueUtils.ResourceCountResult(ResourceQueueUtils.getResourceCount(queueRoot),
false)
+ :
ResourceQueueUtils.getResourceCountCappedResult(queueRoot);
- return new DistributionQueueStatus(count,
DistributionQueueState.PASSIVE);
+ return new DistributionQueueStatus(
+ countResult.getCount(), DistributionQueueState.PASSIVE,
countResult.isCapped());
} catch (LoginException e) {
throw new RuntimeException(e);
} catch (PersistenceException e) {
diff --git
a/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueProvider.java
b/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueProvider.java
index 74af08fa..7f82debf 100644
---
a/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueProvider.java
+++
b/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueProvider.java
@@ -53,6 +53,7 @@ public class ResourceQueueProvider implements
DistributionQueueProvider {
private Scheduler scheduler;
private String agentName;
private boolean isActive;
+ private boolean useExactQueueLength;
private final Map<String, ResourceQueue> queueMap = new
ConcurrentHashMap<>();
@@ -65,6 +66,17 @@ public class ResourceQueueProvider implements
DistributionQueueProvider {
String agentName,
Scheduler scheduler,
boolean isActive) {
+ this(context, resolverFactory, serviceName, agentName, scheduler,
isActive, false);
+ }
+
+ public ResourceQueueProvider(
+ BundleContext context,
+ ResourceResolverFactory resolverFactory,
+ String serviceName,
+ String agentName,
+ Scheduler scheduler,
+ boolean isActive,
+ boolean useExactQueueLength) {
if (serviceName == null
|| (scheduler == null && isActive)
|| context == null
@@ -78,6 +90,7 @@ public class ResourceQueueProvider implements
DistributionQueueProvider {
this.agentRootPath = QUEUES_ROOT + agentName;
this.scheduler = scheduler;
this.isActive = isActive;
+ this.useExactQueueLength = useExactQueueLength;
register(context);
}
@@ -87,9 +100,9 @@ public class ResourceQueueProvider implements
DistributionQueueProvider {
public DistributionQueue getQueue(@NotNull String queueName) throws
DistributionException {
return queueMap.computeIfAbsent(queueName, name -> {
if (isActive) {
- return new ActiveResourceQueue(resolverFactory, serviceName,
name, agentRootPath);
+ return new ActiveResourceQueue(resolverFactory, serviceName,
name, agentRootPath, useExactQueueLength);
} else {
- return new ResourceQueue(resolverFactory, serviceName, name,
agentRootPath);
+ return new ResourceQueue(resolverFactory, serviceName, name,
agentRootPath, useExactQueueLength);
}
});
}
diff --git
a/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueProviderFactory.java
b/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueProviderFactory.java
index 170a602c..1cf2b72e 100644
---
a/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueProviderFactory.java
+++
b/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueProviderFactory.java
@@ -45,6 +45,14 @@ public class ResourceQueueProviderFactory implements
DistributionQueueProviderFa
public @interface Config {
@AttributeDefinition(name = "Should the Resource-backed queue created
with a Queue Processor (i.e., ACTIVE)")
boolean queue_isActive() default false;
+
+ @AttributeDefinition(
+ name = "Use exact (uncapped) queue length for status",
+ description = "By default, queue status/console reporting caps
the counted queue length for "
+ + "performance on very large queues, showing an
approximate 'N+' count instead of the "
+ + "exact size. Enable this to always compute the exact
queue length instead, restoring "
+ + "the pre-cap behavior. This can be slow for queues
with a very large backlog.")
+ boolean status_useExactQueueLength() default false;
}
@Reference
@@ -56,16 +64,19 @@ public class ResourceQueueProviderFactory implements
DistributionQueueProviderFa
BundleContext context;
private boolean isActive;
+ private boolean useExactQueueLength;
@Activate
protected void activate(BundleContext context, Config conf) {
this.isActive = conf.queue_isActive();
+ this.useExactQueueLength = conf.status_useExactQueueLength();
this.context = context;
}
@Override
public DistributionQueueProvider getProvider(String agentName, String
serviceName) {
- return new ResourceQueueProvider(context, resourceResolverFactory,
serviceName, agentName, scheduler, isActive);
+ return new ResourceQueueProvider(
+ context, resourceResolverFactory, serviceName, agentName,
scheduler, isActive, useExactQueueLength);
}
@Override
diff --git
a/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueUtils.java
b/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueUtils.java
index e7d141d3..3b4d36d7 100644
---
a/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueUtils.java
+++
b/src/main/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueUtils.java
@@ -64,6 +64,13 @@ public class ResourceQueueUtils {
private static final String ENTERED_DATE = "entered.date";
private static final String PROCESSING_ATTEMPTS = "processing.attempts";
+ /**
+ * Cap on how many entries getResourceCountCapped() will traverse when
computing queue size for
+ * status/console. Queues larger than this report STATUS_ITEMS_COUNT_CAP
instead of the exact count,
+ * so status lookups on very large queues stay bounded.
+ */
+ static final int STATUS_ITEMS_COUNT_CAP = 1_000;
+
private static final AtomicLong itemCounter = new AtomicLong(0);
private static final Logger log =
LoggerFactory.getLogger(ResourceQueueUtils.class);
@@ -307,6 +314,51 @@ public class ResourceQueueUtils {
return getEntries(root, 0, -1).size();
}
+ /**
+ * Like getResourceCount(), but bounds the amount of traversal done: once
more than
+ * STATUS_ITEMS_COUNT_CAP entries have been seen, stops early and returns
STATUS_ITEMS_COUNT_CAP
+ * instead of the exact count. Intended for status/console displays on
potentially very large
+ * queues, where an exact count is not required and a full traversal would
be too expensive.
+ * Callers that need the exact size should use getResourceCount() instead.
+ */
+ public static int getResourceCountCapped(Resource root) {
+ return getResourceCountCappedResult(root).getCount();
+ }
+
+ static ResourceCountResult getResourceCountCappedResult(Resource root) {
+ Iterator<Resource> it = new ResourceIterator(root, RESOURCE_FOLDER,
false, true);
+
+ int count = 0;
+ while (it.hasNext() && count <= STATUS_ITEMS_COUNT_CAP) {
+ it.next();
+ count++;
+
+ if (count > STATUS_ITEMS_COUNT_CAP) {
+ return new ResourceCountResult(STATUS_ITEMS_COUNT_CAP, true);
+ }
+ }
+
+ return new ResourceCountResult(count, false);
+ }
+
+ static final class ResourceCountResult {
+ private final int count;
+ private final boolean capped;
+
+ ResourceCountResult(int count, boolean capped) {
+ this.count = count;
+ this.capped = capped;
+ }
+
+ int getCount() {
+ return count;
+ }
+
+ boolean isCapped() {
+ return capped;
+ }
+ }
+
private static String getUniqueEntryPath(Resource parent) {
final StringBuilder sb = new StringBuilder();
sb.append(parent.getPath());
diff --git
a/src/main/java/org/apache/sling/distribution/resources/impl/ExtendedDistributionServiceResourceProvider.java
b/src/main/java/org/apache/sling/distribution/resources/impl/ExtendedDistributionServiceResourceProvider.java
index b3ac4345..a0136e45 100644
---
a/src/main/java/org/apache/sling/distribution/resources/impl/ExtendedDistributionServiceResourceProvider.java
+++
b/src/main/java/org/apache/sling/distribution/resources/impl/ExtendedDistributionServiceResourceProvider.java
@@ -141,6 +141,7 @@ public class ExtendedDistributionServiceResourceProvider
extends DistributionSer
result.put("state", queueStatus.getState().name());
result.put("empty", queueStatus.isEmpty());
result.put("itemsCount", queueStatus.getItemsCount());
+ result.put("itemsCountCapped",
queueStatus.isItemsCountCapped());
result.put("capabilities", queueCapabilities(queue));
if
(queueName.startsWith(ErrorQueueDispatchingStrategy.ERROR_PREFIX)) {
diff --git
a/src/test/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueUtilsTest.java
b/src/test/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueUtilsTest.java
index 69cca437..003605c3 100644
---
a/src/test/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueUtilsTest.java
+++
b/src/test/java/org/apache/sling/distribution/queue/impl/resource/ResourceQueueUtilsTest.java
@@ -23,6 +23,8 @@ import java.util.GregorianCalendar;
import org.apache.sling.api.resource.Resource;
import org.apache.sling.api.resource.ResourceResolver;
import org.apache.sling.distribution.queue.DistributionQueueItem;
+import org.apache.sling.distribution.queue.DistributionQueueStatus;
+import org.apache.sling.distribution.queue.impl.SimpleAgentDistributionQueue;
import org.apache.sling.distribution.queue.spi.DistributionQueue;
import org.apache.sling.testing.resourceresolver.MockResourceResolverFactory;
import org.junit.Before;
@@ -79,6 +81,79 @@ public class ResourceQueueUtilsTest {
assertEquals(3, ResourceQueueUtils.getResourceCount(root));
}
+ @Test
+ public void testResourceCountCapped() throws Exception {
+ String agentPath = QUEUES_ROOT + QUEUE_NAME;
+ DistributionQueue queue = new ResourceQueue(rrf, "test", QUEUE_NAME,
agentPath);
+
+ for (int i = 0; i < ResourceQueueUtils.STATUS_ITEMS_COUNT_CAP + 5;
i++) {
+ queue.add(new DistributionQueueItem(randomUUID().toString(),
emptyMap()));
+ }
+
+ ResourceResolver rr = rrf.getResourceResolver(null);
+ Resource root = ResourceQueueUtils.getRootResource(rr, agentPath);
+ assertEquals(ResourceQueueUtils.STATUS_ITEMS_COUNT_CAP,
ResourceQueueUtils.getResourceCountCapped(root));
+ assertEquals(ResourceQueueUtils.STATUS_ITEMS_COUNT_CAP + 5,
ResourceQueueUtils.getResourceCount(root));
+ }
+
+ @Test
+ public void testGetStatusUsesCappedCountByDefault() throws Exception {
+ String agentPath = QUEUES_ROOT + QUEUE_NAME;
+ DistributionQueue queue = new ResourceQueue(rrf, "test", QUEUE_NAME,
agentPath);
+
+ for (int i = 0; i < ResourceQueueUtils.STATUS_ITEMS_COUNT_CAP + 5;
i++) {
+ queue.add(new DistributionQueueItem(randomUUID().toString(),
emptyMap()));
+ }
+
+ DistributionQueueStatus status = queue.getStatus();
+ assertEquals(ResourceQueueUtils.STATUS_ITEMS_COUNT_CAP,
status.getItemsCount());
+ assertTrue(status.isItemsCountCapped());
+
+ DistributionQueueStatus activeDefault = new ActiveResourceQueue(rrf,
"test", QUEUE_NAME, agentPath).getStatus();
+ DistributionQueueStatus activeExplicit =
+ new ActiveResourceQueue(rrf, "test", QUEUE_NAME, agentPath,
false).getStatus();
+ assertEquals(status.getItemsCount(), activeDefault.getItemsCount());
+ assertTrue(activeDefault.isItemsCountCapped());
+ assertEquals(activeDefault.getItemsCount(),
activeExplicit.getItemsCount());
+ assertTrue(activeExplicit.isItemsCountCapped());
+
+ DistributionQueueStatus pausedStatus = new
SimpleAgentDistributionQueue(queue, true, "test-agent").getStatus();
+ assertTrue(pausedStatus.isItemsCountCapped());
+ }
+
+ @Test
+ public void testGetStatusUsesExactCountWhenConfigured() throws Exception {
+ String agentPath = QUEUES_ROOT + QUEUE_NAME;
+ DistributionQueue queue = new ResourceQueue(rrf, "test", QUEUE_NAME,
agentPath, true);
+
+ for (int i = 0; i < ResourceQueueUtils.STATUS_ITEMS_COUNT_CAP + 5;
i++) {
+ queue.add(new DistributionQueueItem(randomUUID().toString(),
emptyMap()));
+ }
+
+ DistributionQueueStatus status = queue.getStatus();
+ assertEquals(ResourceQueueUtils.STATUS_ITEMS_COUNT_CAP + 5,
status.getItemsCount());
+ assertFalse(status.isItemsCountCapped());
+
+ DistributionQueueStatus activeExact =
+ new ActiveResourceQueue(rrf, "test", QUEUE_NAME, agentPath,
true).getStatus();
+ assertEquals(status.getItemsCount(), activeExact.getItemsCount());
+ assertFalse(activeExact.isItemsCountCapped());
+ }
+
+ @Test
+ public void testGetStatusAtCapIsNotCapped() throws Exception {
+ String agentPath = QUEUES_ROOT + QUEUE_NAME;
+ DistributionQueue queue = new ResourceQueue(rrf, "test", QUEUE_NAME,
agentPath);
+
+ for (int i = 0; i < ResourceQueueUtils.STATUS_ITEMS_COUNT_CAP; i++) {
+ queue.add(new DistributionQueueItem(randomUUID().toString(),
emptyMap()));
+ }
+
+ assertEquals(
+ ResourceQueueUtils.STATUS_ITEMS_COUNT_CAP,
queue.getStatus().getItemsCount());
+ assertFalse(queue.getStatus().isItemsCountCapped());
+ }
+
@Test
public void testIsSafeToDelete() throws Exception {
