You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sling.apache.org by cs...@apache.org on 2026/09/20 12:43:55 UTC
(sling-org-apache-sling-distribution-core) 01/01: SLING-13356 - Batch disposable package deletions in ResourceDistributionPackageCleanup
This is an automated email from the ASF dual-hosted git repository.
cschneider pushed a commit to branch SLING-13356
in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-distribution-core.git
commit 94a9a3be424cb9b39ed6663e6249bba9a8dac145
Author: Christian Schneider <cs...@adobe.com>
AuthorDate: Sun Sep 20 14:43:41 2026 +0200
SLING-13356 - Batch disposable package deletions in ResourceDistributionPackageCleanup
ResourceDistributionPackageCleanup previously deleted all disposable packages
in a single JCR commit. On a resource-persisted distribution agent with a
large package backlog, this becomes an unbounded transaction that can
overwhelm the underlying repository.
Commits are now issued in bounded batches, tracked via a new
cleanupBatchSize constructor parameter / OSGi config attribute (default
100) exposed on both DistributionPackageBuilderFactory and
VaultDistributionPackageBuilderFactory. A value <= 0 preserves the
previous single-commit-per-run behavior. The existing two-argument
constructor is kept for backward compatibility and defaults to batching
disabled.
---
.../impl/ResourceDistributionPackageCleanup.java | 25 ++-
.../impl/DistributionPackageBuilderFactory.java | 14 +-
.../VaultDistributionPackageBuilderFactory.java | 14 +-
.../ResourceDistributionPackageCleanupTest.java | 247 +++++++++++++++++++++
4 files changed, 295 insertions(+), 5 deletions(-)
diff --git a/src/main/java/org/apache/sling/distribution/packaging/impl/ResourceDistributionPackageCleanup.java b/src/main/java/org/apache/sling/distribution/packaging/impl/ResourceDistributionPackageCleanup.java
index b51142c6..185f6a26 100644
--- a/src/main/java/org/apache/sling/distribution/packaging/impl/ResourceDistributionPackageCleanup.java
+++ b/src/main/java/org/apache/sling/distribution/packaging/impl/ResourceDistributionPackageCleanup.java
@@ -32,6 +32,9 @@ import org.slf4j.LoggerFactory;
/**
* This runnable removes unreferenced {@link ResourceDistributionPackage} packages.
* It is meant to be run periodically. See SLING-6503.
+ * Deletions are committed in batches (see SLING-13356) rather than in a single commit for
+ * the whole run, to avoid an unbounded transaction when a large number of packages have
+ * accumulated.
*/
public class ResourceDistributionPackageCleanup implements Runnable {
@@ -44,18 +47,33 @@ public class ResourceDistributionPackageCleanup implements Runnable {
private final ResourceResolverFactory resolverFactory;
+ /**
+ * Maximum number of disposable packages deleted per JCR commit during a cleanup run.
+ * A value {@code <= 0} disables batching, restoring the previous behavior of a single
+ * commit for the whole run.
+ */
+ private final int cleanupBatchSize;
+
public ResourceDistributionPackageCleanup(
@NotNull ResourceResolverFactory resolverFactory,
@NotNull ResourceDistributionPackageBuilder packageBuilder) {
+ this(resolverFactory, packageBuilder, 0);
+ }
+
+ public ResourceDistributionPackageCleanup(
+ @NotNull ResourceResolverFactory resolverFactory,
+ @NotNull ResourceDistributionPackageBuilder packageBuilder,
+ int cleanupBatchSize) {
this.resolverFactory = resolverFactory;
this.packageBuilder = packageBuilder;
+ this.cleanupBatchSize = cleanupBatchSize;
}
public void run() {
log.debug("Cleaning up {} packages", packageBuilder.getType());
ResourceResolver serviceResolver = null;
try {
- int deleted = 0, total = 0;
+ int deleted = 0, total = 0, pendingInBatch = 0;
serviceResolver = resolverFactory.getServiceResourceResolver(null);
for (Iterator<ResourceDistributionPackage> pkgs = packageBuilder.getPackages(serviceResolver);
pkgs.hasNext();
@@ -65,6 +83,11 @@ public class ResourceDistributionPackageCleanup implements Runnable {
log.debug("Delete package {}", pkg.getId());
deleted++;
pkg.delete(false);
+ pendingInBatch++;
+ if (cleanupBatchSize > 0 && pendingInBatch >= cleanupBatchSize) {
+ serviceResolver.commit();
+ pendingInBatch = 0;
+ }
} else {
log.debug("package {} is not disposable", pkg.getId());
}
diff --git a/src/main/java/org/apache/sling/distribution/serialization/impl/DistributionPackageBuilderFactory.java b/src/main/java/org/apache/sling/distribution/serialization/impl/DistributionPackageBuilderFactory.java
index 96d5fc61..03ca7359 100644
--- a/src/main/java/org/apache/sling/distribution/serialization/impl/DistributionPackageBuilderFactory.java
+++ b/src/main/java/org/apache/sling/distribution/serialization/impl/DistributionPackageBuilderFactory.java
@@ -140,6 +140,14 @@ public class DistributionPackageBuilderFactory implements DistributionPackageBui
+ "The delay between two runs of the cleanup phase can be configured with this setting. 60 seconds by default")
long cleanupDelay() default DEFAULT_PACKAGE_CLEANUP_DELAY;
+ @AttributeDefinition(
+ name = "The number of disposable packages deleted per commit during the cleanup phase.",
+ description = "The resource persisted packages are deleted in batches of this size during each cleanup "
+ + "run, rather than in a single commit for the whole run, to avoid an unbounded "
+ + "transaction when a large number of packages have accumulated. A value <= 0 "
+ + "restores the previous behavior of a single commit per cleanup run. 100 by default")
+ int cleanupBatchSize() default DEFAULT_PACKAGE_CLEANUP_BATCH_SIZE;
+
@AttributeDefinition(
name = "Package Node Filters",
description = "The package node path filters. Filter format: path|+include|-exclude",
@@ -170,6 +178,7 @@ public class DistributionPackageBuilderFactory implements DistributionPackageBui
private static final String DEFAULT_DIGEST_ALGORITHM = "NONE";
private static final int DEFAULT_MONITORING_QUEUE_SIZE = 0;
private static final long DEFAULT_PACKAGE_CLEANUP_DELAY = 60L;
+ private static final int DEFAULT_PACKAGE_CLEANUP_BATCH_SIZE = 100;
@Activate
public void activate(BundleContext context, Config conf) {
@@ -180,6 +189,7 @@ public class DistributionPackageBuilderFactory implements DistributionPackageBui
String tempFsFolder = SettingsUtils.removeEmptyEntry(conf.tempFsFolder());
String digestAlgorithm = conf.digestAlgorithm();
long cleanupDelay = conf.cleanupDelay();
+ int cleanupBatchSize = conf.cleanupBatchSize();
if (DEFAULT_DIGEST_ALGORITHM.equals(digestAlgorithm)) {
digestAlgorithm = null;
}
@@ -212,8 +222,8 @@ public class DistributionPackageBuilderFactory implements DistributionPackageBui
digestAlgorithm,
nodeFilters,
propertyFilters);
- Runnable cleanup =
- new ResourceDistributionPackageCleanup(resolverFactory, resourceDistributionPackageBuilder);
+ Runnable cleanup = new ResourceDistributionPackageCleanup(
+ resolverFactory, resourceDistributionPackageBuilder, cleanupBatchSize);
Dictionary<String, Object> props = new Hashtable<String, Object>();
props.put(Scheduler.PROPERTY_SCHEDULER_CONCURRENT, false);
props.put(Scheduler.PROPERTY_SCHEDULER_PERIOD, cleanupDelay);
diff --git a/src/main/java/org/apache/sling/distribution/serialization/impl/vlt/VaultDistributionPackageBuilderFactory.java b/src/main/java/org/apache/sling/distribution/serialization/impl/vlt/VaultDistributionPackageBuilderFactory.java
index e3c6695d..aa5d5a52 100644
--- a/src/main/java/org/apache/sling/distribution/serialization/impl/vlt/VaultDistributionPackageBuilderFactory.java
+++ b/src/main/java/org/apache/sling/distribution/serialization/impl/vlt/VaultDistributionPackageBuilderFactory.java
@@ -135,6 +135,14 @@ public class VaultDistributionPackageBuilderFactory implements DistributionPacka
+ "The delay between two runs of the cleanup phase can be configured with this setting. 60 seconds by default")
long cleanupDelay() default DEFAULT_PACKAGE_CLEANUP_DELAY;
+ @AttributeDefinition(
+ name = "The number of disposable packages deleted per commit during the cleanup phase.",
+ description = "The resource persisted packages are deleted in batches of this size during each cleanup "
+ + "run, rather than in a single commit for the whole run, to avoid an unbounded "
+ + "transaction when a large number of packages have accumulated. A value <= 0 "
+ + "restores the previous behavior of a single commit per cleanup run. 100 by default")
+ int cleanupBatchSize() default DEFAULT_PACKAGE_CLEANUP_BATCH_SIZE;
+
@AttributeDefinition(
name = "File threshold (in bytes)",
description =
@@ -202,6 +210,7 @@ public class VaultDistributionPackageBuilderFactory implements DistributionPacka
}
private static final long DEFAULT_PACKAGE_CLEANUP_DELAY = 60L;
+ private static final int DEFAULT_PACKAGE_CLEANUP_BATCH_SIZE = 100;
// 1M
private static final int DEFAULT_FILE_THRESHOLD_VALUE = 1;
private static final String DEFAULT_MEMORY_UNIT = "MEGA_BYTES";
@@ -236,6 +245,7 @@ public class VaultDistributionPackageBuilderFactory implements DistributionPacka
String[] packagePropertyFilters = SettingsUtils.removeEmptyEntries(conf.property_filters());
long cleanupDelay = conf.cleanupDelay();
+ int cleanupBatchSize = conf.cleanupBatchSize();
String tempFsFolder = SettingsUtils.removeEmptyEntry(conf.tempFsFolder());
boolean useBinaryReferences = conf.useBinaryReferences();
@@ -315,8 +325,8 @@ public class VaultDistributionPackageBuilderFactory implements DistributionPacka
digestAlgorithm,
packageNodeFilters,
packagePropertyFilters);
- Runnable cleanup =
- new ResourceDistributionPackageCleanup(resolverFactory, resourceDistributionPackageBuilder);
+ Runnable cleanup = new ResourceDistributionPackageCleanup(
+ resolverFactory, resourceDistributionPackageBuilder, cleanupBatchSize);
Dictionary<String, Object> props = new Hashtable<String, Object>();
props.put(Scheduler.PROPERTY_SCHEDULER_CONCURRENT, false);
props.put(Scheduler.PROPERTY_SCHEDULER_PERIOD, cleanupDelay);
diff --git a/src/test/java/org/apache/sling/distribution/packaging/impl/ResourceDistributionPackageCleanupTest.java b/src/test/java/org/apache/sling/distribution/packaging/impl/ResourceDistributionPackageCleanupTest.java
new file mode 100644
index 00000000..8e97a560
--- /dev/null
+++ b/src/test/java/org/apache/sling/distribution/packaging/impl/ResourceDistributionPackageCleanupTest.java
@@ -0,0 +1,247 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sling.distribution.packaging.impl;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Iterator;
+
+import org.apache.sling.api.resource.LoginException;
+import org.apache.sling.api.resource.ResourceResolver;
+import org.apache.sling.api.resource.ResourceResolverFactory;
+import org.apache.sling.distribution.DistributionRequest;
+import org.apache.sling.distribution.common.DistributionException;
+import org.apache.sling.distribution.packaging.DistributionPackage;
+import org.apache.sling.distribution.serialization.DistributionContentSerializer;
+import org.apache.sling.distribution.serialization.DistributionExportOptions;
+import org.apache.sling.distribution.util.impl.FileBackedMemoryOutputStream.MemoryUnit;
+import org.apache.sling.testing.mock.osgi.MockOsgi;
+import org.apache.sling.testing.mock.sling.MockSling;
+import org.apache.sling.testing.mock.sling.ResourceResolverType;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.osgi.framework.BundleContext;
+import org.osgi.framework.ServiceReference;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class ResourceDistributionPackageCleanupTest {
+
+ private BundleContext bundleContext;
+ private ResourceResolver resolver;
+ private ResourceDistributionPackageBuilder builder;
+
+ @Before
+ public void setUp() {
+ bundleContext = MockOsgi.newBundleContext();
+ MockSling.setAdapterManagerBundleContext(bundleContext);
+ resolver = MockSling.newResourceResolver(ResourceResolverType.JCR_MOCK, bundleContext);
+ builder = new ResourceDistributionPackageBuilder(
+ "test",
+ new TestSerializer(),
+ null,
+ 0,
+ MemoryUnit.valueOf("MEGA_BYTES"),
+ false,
+ null,
+ new String[0],
+ new String[0]);
+ }
+
+ @After
+ public void tearDown() {
+ if (resolver.isLive()) {
+ resolver.close();
+ }
+ MockSling.clearAdapterManagerBundleContext();
+ }
+
+ @Test
+ public void testAllDisposablePackagesAreDeletedInBatches() throws Exception {
+ int total = 7;
+ for (int i = 0; i < total; i++) {
+ createDisposablePackage();
+ }
+ assertEquals(total, countPackages(resolver));
+
+ ResourceResolver serviceResolver = Mockito.spy(resolver);
+ ResourceResolverFactory resolverFactory = mockResolverFactory(serviceResolver);
+ ResourceDistributionPackageCleanup cleanup =
+ new ResourceDistributionPackageCleanup(resolverFactory, builder, 3);
+ cleanup.run();
+
+ assertEquals("all disposable packages should be deleted", 0, countPackages(newVerificationResolver()));
+ // ceil(7 / 3) = 3 commits: after the 3rd and 6th deletions, plus a final commit for the
+ // 7th (the "if hasChanges()" fallback after the loop).
+ verify(serviceResolver, times(3)).commit();
+ }
+
+ @Test
+ public void testNonDisposablePackagesAreLeftAlone() throws Exception {
+ createDisposablePackage();
+ createDisposablePackage();
+ createNonDisposablePackage();
+ assertEquals(3, countPackages(resolver));
+
+ ResourceResolverFactory resolverFactory = mockResolverFactory(resolver);
+ ResourceDistributionPackageCleanup cleanup =
+ new ResourceDistributionPackageCleanup(resolverFactory, builder, 100);
+ cleanup.run();
+
+ assertEquals("only the non-disposable package should remain", 1, countPackages(newVerificationResolver()));
+ }
+
+ @Test
+ public void testExactMultipleOfBatchSizeCommitsOnlyOncePerBatch() throws Exception {
+ int total = 6;
+ for (int i = 0; i < total; i++) {
+ createDisposablePackage();
+ }
+
+ ResourceResolver serviceResolver = Mockito.spy(resolver);
+ ResourceResolverFactory resolverFactory = mockResolverFactory(serviceResolver);
+ ResourceDistributionPackageCleanup cleanup =
+ new ResourceDistributionPackageCleanup(resolverFactory, builder, 3);
+ cleanup.run();
+
+ assertEquals(0, countPackages(newVerificationResolver()));
+ // 6 / 3 = exactly 2 commits, no extra empty commit at the end since nothing is pending.
+ verify(serviceResolver, times(2)).commit();
+ }
+
+ @Test
+ public void testNonPositiveBatchSizeFallsBackToSingleCommit() throws Exception {
+ int total = 5;
+ for (int i = 0; i < total; i++) {
+ createDisposablePackage();
+ }
+
+ ResourceResolver serviceResolver = Mockito.spy(resolver);
+ ResourceResolverFactory resolverFactory = mockResolverFactory(serviceResolver);
+ ResourceDistributionPackageCleanup cleanup =
+ new ResourceDistributionPackageCleanup(resolverFactory, builder, 0);
+ cleanup.run();
+
+ assertEquals(0, countPackages(newVerificationResolver()));
+ verify(serviceResolver, times(1)).commit();
+ }
+
+ @Test
+ public void testDeprecatedTwoArgConstructorFallsBackToSingleCommit() throws Exception {
+ int total = 4;
+ for (int i = 0; i < total; i++) {
+ createDisposablePackage();
+ }
+
+ ResourceResolver serviceResolver = Mockito.spy(resolver);
+ ResourceResolverFactory resolverFactory = mockResolverFactory(serviceResolver);
+ ResourceDistributionPackageCleanup cleanup = new ResourceDistributionPackageCleanup(resolverFactory, builder);
+ cleanup.run();
+
+ assertEquals(0, countPackages(newVerificationResolver()));
+ verify(serviceResolver, times(1)).commit();
+ }
+
+ private ResourceResolverFactory mockResolverFactory(ResourceResolver serviceResolver) throws LoginException {
+ ResourceResolverFactory resolverFactory = mock(ResourceResolverFactory.class);
+ when(resolverFactory.getServiceResourceResolver(null)).thenReturn(serviceResolver);
+ return resolverFactory;
+ }
+
+ /**
+ * {@code cleanup.run()} closes the resolver it was given (per its own {@code finally} block),
+ * so post-run assertions need a fresh resolver against the same underlying (JCR_MOCK) repository
+ * rather than reusing the one passed to the cleanup run. {@code MockSling.newResourceResolver()}
+ * can't be called a second time on the same bundle context (it tries to register a second
+ * {@code ResourceResolverFactory} service) -- fetch the one already registered in {@link #setUp()}
+ * instead.
+ */
+ private ResourceResolver newVerificationResolver() throws LoginException {
+ ServiceReference<ResourceResolverFactory> ref =
+ bundleContext.getServiceReference(ResourceResolverFactory.class);
+ ResourceResolverFactory factory = bundleContext.getService(ref);
+ return factory.getAdministrativeResourceResolver(null);
+ }
+
+ private void createDisposablePackage() throws DistributionException, IOException {
+ ResourceDistributionPackage pkg = createPackage();
+ pkg.acquire("holder");
+ pkg.release("holder");
+ }
+
+ private void createNonDisposablePackage() throws DistributionException, IOException {
+ ResourceDistributionPackage pkg = createPackage();
+ pkg.acquire("holder");
+ // never released -> not disposable
+ }
+
+ private ResourceDistributionPackage createPackage() throws DistributionException, IOException {
+ DistributionRequest mockRequest = mock(DistributionRequest.class);
+ String path = "/content/" + java.util.UUID.randomUUID();
+ when(mockRequest.getPaths()).thenReturn(new String[] {path});
+ when(mockRequest.isDeep(path)).thenReturn(false);
+ DistributionPackage pkg = builder.createPackageForAdd(resolver, mockRequest);
+ return (ResourceDistributionPackage) pkg;
+ }
+
+ private int countPackages(ResourceResolver withResolver) throws DistributionException {
+ int count = 0;
+ for (Iterator<ResourceDistributionPackage> it = builder.getPackages(withResolver); it.hasNext(); it.next()) {
+ count++;
+ }
+ return count;
+ }
+
+ private static class TestSerializer implements DistributionContentSerializer {
+
+ @Override
+ public void exportToStream(
+ ResourceResolver resourceResolver, DistributionExportOptions exportOptions, OutputStream outputStream)
+ throws DistributionException {
+ try {
+ outputStream.write("test".getBytes());
+ } catch (IOException ex) {
+ throw new DistributionException(ex);
+ }
+ }
+
+ @Override
+ public void importFromStream(ResourceResolver resourceResolver, InputStream inputStream)
+ throws DistributionException {
+ throw new DistributionException("unsupported");
+ }
+
+ @Override
+ public String getName() {
+ return "test";
+ }
+
+ @Override
+ public boolean isRequestFiltering() {
+ return true;
+ }
+ }
+}