This is an automated email from the ASF dual-hosted git repository.
cschneider pushed a commit to branch master
in repository
https://gitbox.apache.org/repos/asf/sling-org-apache-sling-distribution-journal.git
The following commit(s) were added to refs/heads/master by this push:
new ae40176 SLING-13151 - Clear queue cache after 12h (#187)
ae40176 is described below
commit ae40176cd3d4728b3095fc953af2cf42796f1209
Author: Christian Schneider <[email protected]>
AuthorDate: Mon Mar 30 16:18:34 2026 +0200
SLING-13151 - Clear queue cache after 12h (#187)
---
docs/documentation.md | 1 +
.../journal/queue/impl/PubQueueProviderImpl.java | 16 +++-------------
.../journal/queue/impl/PubQueueProviderTest.java | 2 ++
3 files changed, 6 insertions(+), 13 deletions(-)
diff --git a/docs/documentation.md b/docs/documentation.md
index 91ebe66..a64666b 100644
--- a/docs/documentation.md
+++ b/docs/documentation.md
@@ -147,6 +147,7 @@ The tail consumer keeps reading the new packages at the
tail of the queue. The h
This mechanism has multiple advantages, the queue are readily exposed when the
agent start (potentially blocking on the head poller if needed), the minimum
amount of messages are fetched from the package topic, queues can be computed
quickly and efficiently from the cache.
In order to avoid an infinite cache growth, we implemented a cleanup task that
reduce the cache memory usage.
+The publisher package cache is fully recycled on a fixed interval (12 hours)
regardless of size, so queue views are rebuilt from the journal and do not
retain entries that the package topic no longer holds.
### Error queue
diff --git
a/src/main/java/org/apache/sling/distribution/journal/queue/impl/PubQueueProviderImpl.java
b/src/main/java/org/apache/sling/distribution/journal/queue/impl/PubQueueProviderImpl.java
index 224177d..7505a1f 100644
---
a/src/main/java/org/apache/sling/distribution/journal/queue/impl/PubQueueProviderImpl.java
+++
b/src/main/java/org/apache/sling/distribution/journal/queue/impl/PubQueueProviderImpl.java
@@ -64,11 +64,6 @@ import org.slf4j.LoggerFactory;
@ParametersAreNonnullByDefault
public class PubQueueProviderImpl implements PubQueueProvider, Runnable {
- /**
- * The minimum size to collect the cache. Each cache entry requires
- * around 500B of heap space. 10'000 entries ~= 5MB on heap.
- */
- private static final int CLEANUP_THRESHOLD = 10_000;
static final long QUEUE_SIZE_REFRESH_INTERVAL_SECONDS = 30;
@@ -119,7 +114,6 @@ public class PubQueueProviderImpl implements
PubQueueProvider, Runnable {
}
private void startCleanupTask(BundleContext context) {
- // Register periodic task to update the topology view
Dictionary<String, Object> props = new Hashtable<>();
props.put(PROPERTY_SCHEDULER_CONCURRENT, false);
props.put(PROPERTY_SCHEDULER_PERIOD, 12*60*60L); // every 12 h
@@ -152,13 +146,9 @@ public class PubQueueProviderImpl implements
PubQueueProvider, Runnable {
PubQueueCache queueCache = this.cache;
if (queueCache != null) {
int size = queueCache.size();
- if (size > CLEANUP_THRESHOLD) {
- LOG.info("Cleanup package cache (size={}/{})", size,
CLEANUP_THRESHOLD);
- queueCache.close();
- this.cache = newCache();
- } else {
- LOG.info("No cleanup required for package cache (size={}/{})",
size, CLEANUP_THRESHOLD);
- }
+ LOG.info("Recycling package cache (size={})", size);
+ queueCache.close();
+ this.cache = newCache();
}
LOG.info("Stopping package cache cleanup task");
}
diff --git
a/src/test/java/org/apache/sling/distribution/journal/queue/impl/PubQueueProviderTest.java
b/src/test/java/org/apache/sling/distribution/journal/queue/impl/PubQueueProviderTest.java
index 05108b0..e2a6368 100644
---
a/src/test/java/org/apache/sling/distribution/journal/queue/impl/PubQueueProviderTest.java
+++
b/src/test/java/org/apache/sling/distribution/journal/queue/impl/PubQueueProviderTest.java
@@ -252,6 +252,8 @@ public class PubQueueProviderTest {
assertThat(queueSize(), equalTo(1));
queueProvider.run();
+ handler = handlerCaptor.getValue();
+ handler.handle(info(0L), packageMessage("packageid1",
PUB1_AGENT_NAME));
assertThat(queueSize(), equalTo(1));
for (long c=0; c<10001;c++) {
