This is an automated email from the ASF dual-hosted git repository. cschneider pushed a commit to branch SLING-13151 in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-distribution-journal.git
commit 5613387436bea74d56bf4956c77e982d8695a00e Author: Christian Schneider <[email protected]> AuthorDate: Mon Mar 30 11:41:15 2026 +0200 SLING-13151 - Clear queue cache after 12h --- 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++) {
