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 <[email protected]> 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; + } + } +}
