This is an automated email from the ASF dual-hosted git repository. cschneider pushed a commit to branch SLING-13396-batch-install in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-distribution-journal.git
commit ba0ed319fa04f9127c069dd74c0d9265d074f68b Author: Christian Schneider <[email protected]> AuthorDate: Thu Oct 22 19:38:13 2026 +0200 SLING-13396 - Subscriber batch import for slow package builders Add an opt-in batch import to the journal subscriber. A package builder implementing the new exported SPI org.apache.sling.distribution.journal.spi.BatchDistributionPackageBuilder receives consecutive ADD/DELETE packages in one installPackages call and may install independent packages concurrently. - New subscriber setting maxBatchSize (default 1 = disabled). Active only for non editable agents with packageHandling=Off and a batch capable package builder; otherwise a warning is logged. - A bounded queue decouples the poller from an importer thread, which takes whatever is already queued (up to maxBatchSize), so there is no extra latency for single packages. - The offset of the last package is stored in the same commit as the batch. On failure no offset is stored and the packages are imported one by one with the existing retry/error queue handling (at least once). - New metrics imported_batch_size and imported_batch_duration. Co-authored-by: Copilot App <[email protected]> --- docs/documentation.md | 17 + docs/metrics_overview.md | 12 + .../journal/bookkeeper/BatchEntry.java | 52 ++++ .../journal/bookkeeper/BookKeeper.java | 83 +++++ .../journal/bookkeeper/PackageHandler.java | 27 ++ .../journal/bookkeeper/SubscriberMetrics.java | 20 ++ .../impl/subscriber/DistributionSubscriber.java | 224 +++++++++++++- .../impl/subscriber/SubscriberConfiguration.java | 8 + .../spi/BatchDistributionPackageBuilder.java | 67 ++++ .../distribution/journal/spi/package-info.java | 25 ++ .../journal/bookkeeper/BookKeeperBatchTest.java | 175 +++++++++++ .../impl/subscriber/SubscriberBatchTest.java | 343 +++++++++++++++++++++ 12 files changed, 1050 insertions(+), 3 deletions(-) diff --git a/docs/documentation.md b/docs/documentation.md index a64666b..6d8588f 100644 --- a/docs/documentation.md +++ b/docs/documentation.md @@ -162,6 +162,23 @@ Unlike other error queues in SCD, this error queue does _not_ allow to retry a f The implementation of the _Subscriber_ agent queue is a blocking queue that fetches a small (currently 8) batch of packages. Using a batch batch reduces the processing delay between two items (avoids the request delay) but keep the memory footprint small by not fetching the entire queue. +### Batch import + +By default the _Subscriber_ agent installs one package at a time. When installing a package is slow, for example because the package builder calls a remote system for every package, a backlog of many small packages is processed slowly even though most of them are independent. + +A package builder can implement `org.apache.sling.distribution.journal.spi.BatchDistributionPackageBuilder` to receive several consecutive packages in one call. The subscriber uses it when `maxBatchSize` is set to a value greater than 1, the agent is not editable and `packageHandling` is `Off`. Otherwise it logs a warning and imports packages one by one. + +With batching enabled: + +* The poller thread hands packages to a separate importer thread through a queue of `maxBatchSize` entries, so the poller reads at most `maxBatchSize` packages ahead. +* The importer takes all packages that are already queued, up to `maxBatchSize`. It does not wait for a batch to fill, so a single package is imported without delay and a backlog is imported in batches. +* Consecutive ADD and DELETE packages that pass the precondition are installed together. Any other package (for example INVALIDATE, a package skipped by the precondition or a package for another agent) ends the current batch, which is imported first, and is then processed as before. +* The offset of the last package of a batch is stored in the same commit as the batch, so offsets are always stored in journal order. +* If the batch fails, no offset is stored and each package of the batch is imported again one by one, with the usual retry and error queue handling. Packages of a failed batch can therefore be installed twice, so installing a package must be idempotent. +* The package builder receives the packages in journal order and must produce the same result as installing them one by one. It may only install packages concurrently when that cannot change the result, for example when their paths do not overlap. + +Batching is not supported for editable agents, because they store a single package status per commit and must report a status for every package. + ## Handling of blocked Queue The queues sometimes get blocked, because a package can't be imported at destination. The root cause for a blocked queue can vary but generally is one of I. lack of authorisation, II. repository constraint violation. diff --git a/docs/metrics_overview.md b/docs/metrics_overview.md index ebed07f..62719eb 100644 --- a/docs/metrics_overview.md +++ b/docs/metrics_overview.md @@ -98,6 +98,18 @@ All subscriber metrics are prefixed with `sling_distribution_journal_subscriber_ - **Description**: Duration of successful package import operations - **Tags**: `sub_name`, `pub_name`, `editable` +#### `sling_distribution_journal_subscriber_imported_batch_size` (Histogram) +- **Type**: Histogram +- **Unit**: Packages +- **Description**: Number of packages imported together in one successful batch (only with `maxBatchSize` > 1) +- **Tags**: `sub_name`, `pub_name`, `editable` + +#### `sling_distribution_journal_subscriber_imported_batch_duration` (Timer) +- **Type**: Timer +- **Unit**: Milliseconds +- **Description**: Duration of batch import operations, successful or not (only with `maxBatchSize` > 1) +- **Tags**: `sub_name`, `pub_name`, `editable` + #### `sling_distribution_journal_subscriber_current_import_duration` (Gauge) - **Type**: Gauge - **Unit**: Milliseconds diff --git a/src/main/java/org/apache/sling/distribution/journal/bookkeeper/BatchEntry.java b/src/main/java/org/apache/sling/distribution/journal/bookkeeper/BatchEntry.java new file mode 100644 index 0000000..591675d --- /dev/null +++ b/src/main/java/org/apache/sling/distribution/journal/bookkeeper/BatchEntry.java @@ -0,0 +1,52 @@ +/* + * 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.journal.bookkeeper; + +import java.util.Date; + +import org.apache.sling.distribution.journal.messages.PackageMessage; + +import static java.util.Objects.requireNonNull; + +/** + * A package message together with its journal offset, as part of a batch import. + */ +public final class BatchEntry { + private final PackageMessage message; + private final long offset; + private final Date createdTime; + + public BatchEntry(PackageMessage message, long offset, Date createdTime) { + this.message = requireNonNull(message); + this.offset = offset; + this.createdTime = requireNonNull(createdTime); + } + + public PackageMessage getMessage() { + return message; + } + + public long getOffset() { + return offset; + } + + public Date getCreatedTime() { + return createdTime; + } +} diff --git a/src/main/java/org/apache/sling/distribution/journal/bookkeeper/BookKeeper.java b/src/main/java/org/apache/sling/distribution/journal/bookkeeper/BookKeeper.java index a0e1192..5c07c8d 100644 --- a/src/main/java/org/apache/sling/distribution/journal/bookkeeper/BookKeeper.java +++ b/src/main/java/org/apache/sling/distribution/journal/bookkeeper/BookKeeper.java @@ -22,8 +22,10 @@ import java.io.IOException; import java.io.PrintWriter; import java.io.StringWriter; import java.time.Duration; +import java.util.ArrayList; import java.util.Date; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Objects; import java.util.concurrent.TimeUnit; @@ -208,6 +210,87 @@ public class BookKeeper { } } + /** + * Import several consecutive packages with a single call to the package builder and a single + * commit that stores the offset of the last package. + * + * <p>Unlike {@link #importPackage(PackageMessage, long, Date, Date)} this method does not + * apply the retry or error queue handling. If it throws, no offset has been stored for the + * batch, unless an import post processor failed after the commit, and the caller is expected + * to import the packages one by one.</p> + * + * <p>Batch imports are only supported for non editable subscribers, because the status store + * holds a single status and editable subscribers must report a status for every package.</p> + */ + public void importPackages(List<BatchEntry> entries, Date importStartTime) throws DistributionException { + if (entries.isEmpty()) { + return; + } + if (config.isEditable()) { + throw new IllegalStateException("Batch import is not supported for editable subscribers"); + } + BatchEntry first = entries.get(0); + BatchEntry last = entries.get(entries.size() - 1); + log.debug( + "Importing batch of {} distribution packages at offsets {}-{}", + entries.size(), + first.getOffset(), + last.getOffset()); + List<PackageMessage> pkgMsgs = new ArrayList<>(entries.size()); + for (BatchEntry entry : entries) { + pkgMsgs.add(entry.getMessage()); + } + try (Timer.Context context = + subscriberMetrics.getImportedBatchDuration().time(); + ResourceResolver importerResolver = getServiceResolver(SUBSERVICE_IMPORTER)) { + for (PackageMessage pkgMsg : pkgMsgs) { + preProcess(pkgMsg); + } + subscriberMetrics.setCurrentImport( + new CurrentImportInfo(first.getMessage(), first.getOffset(), importStartTime.getTime())); + packageHandler.applyBatch(importerResolver, pkgMsgs); + storeOffset(importerResolver, last.getOffset()); + importerResolver.commit(); + subscriberMetrics.getImportedBatchSize().update(entries.size()); + + for (BatchEntry entry : entries) { + PackageMessage pkgMsg = entry.getMessage(); + subscriberMetrics.getImportedPackageSize().update(pkgMsg.getPkgLength()); + subscriberMetrics + .getPackageDistributedDuration() + .update((currentTimeMillis() - entry.getCreatedTime().getTime()), TimeUnit.MILLISECONDS); + postProcess(pkgMsg); + clearPackageRetriesOnSuccess(pkgMsg); + eventAdmin.postEvent(new AppliedEvent(pkgMsg, config.getSubAgentName()).toEvent()); + subscriberMetrics + .getPackageStatusCounter(pkgMsg.getPubAgentName(), Status.IMPORTED) + .increment(); + } + log.info( + "Imported batch of {} distribution packages at offsets {}-{} took importDurationMs={}", + entries.size(), + first.getOffset(), + last.getOffset(), + System.currentTimeMillis() - importStartTime.getTime()); + } catch (LoginException + | IOException + | RuntimeException + | ImportPreProcessException + | ImportPostProcessException e) { + throw new DistributionException( + format( + "Failed to import batch of %d distribution packages at offsets %d-%d because of '%s'", + entries.size(), first.getOffset(), last.getOffset(), e.getMessage()), + e); + } finally { + subscriberMetrics.clearCurrentImport(); + } + } + + public boolean supportsBatchImport() { + return !config.isEditable() && packageHandler.supportsBatch(); + } + public void invalidateCache(PackageMessage pkgMsg, long offset, Date createdTime, Date importStartTime) throws DistributionException { log.debug("Invalidating the cache for the package {} at offset={}", pkgMsg, offset); diff --git a/src/main/java/org/apache/sling/distribution/journal/bookkeeper/PackageHandler.java b/src/main/java/org/apache/sling/distribution/journal/bookkeeper/PackageHandler.java index 920a544..14df5ae 100644 --- a/src/main/java/org/apache/sling/distribution/journal/bookkeeper/PackageHandler.java +++ b/src/main/java/org/apache/sling/distribution/journal/bookkeeper/PackageHandler.java @@ -23,6 +23,8 @@ import javax.annotation.Nonnull; import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; +import java.util.ArrayList; +import java.util.List; import org.apache.sling.api.resource.ResourceResolver; import org.apache.sling.distribution.common.DistributionException; @@ -30,6 +32,7 @@ import org.apache.sling.distribution.journal.BinaryStore; import org.apache.sling.distribution.journal.messages.PackageMessage; import org.apache.sling.distribution.journal.messages.PackageMessage.ReqType; import org.apache.sling.distribution.journal.shared.JournalDistributionPackage; +import org.apache.sling.distribution.journal.spi.BatchDistributionPackageBuilder; import org.apache.sling.distribution.packaging.DistributionPackage; import org.apache.sling.distribution.packaging.DistributionPackageBuilder; import org.apache.sling.distribution.packaging.DistributionPackageInfo; @@ -65,6 +68,30 @@ class PackageHandler { } } + public boolean supportsBatch() { + return packageBuilder instanceof BatchDistributionPackageBuilder; + } + + /** + * Install all packages with one call to the {@link BatchDistributionPackageBuilder}. + */ + public void applyBatch(ResourceResolver resolver, List<PackageMessage> pkgMsgs) throws DistributionException { + if (!supportsBatch()) { + throw new IllegalStateException("Package builder " + packageBuilder.getType() + " does not implement " + + BatchDistributionPackageBuilder.class.getName()); + } + List<DistributionPackage> distributionPackages = new ArrayList<>(pkgMsgs.size()); + for (PackageMessage pkgMsg : pkgMsgs) { + distributionPackages.add(toDistributionPackage(pkgMsg)); + } + ((BatchDistributionPackageBuilder) packageBuilder).installPackages(resolver, distributionPackages); + for (PackageMessage pkgMsg : pkgMsgs) { + if (pkgMsg.getReqType() == ADD) { + extractor.handle(resolver, pkgMsg.getPaths()); + } + } + } + private DistributionPackage toDistributionPackage(PackageMessage pkgMsg) throws DistributionException { LOG.debug("Importing paths {}", pkgMsg.getPaths()); final byte[] data; diff --git a/src/main/java/org/apache/sling/distribution/journal/bookkeeper/SubscriberMetrics.java b/src/main/java/org/apache/sling/distribution/journal/bookkeeper/SubscriberMetrics.java index 887390a..2bb5c9b 100644 --- a/src/main/java/org/apache/sling/distribution/journal/bookkeeper/SubscriberMetrics.java +++ b/src/main/java/org/apache/sling/distribution/journal/bookkeeper/SubscriberMetrics.java @@ -90,6 +90,8 @@ public class SubscriberMetrics { private static final String IMPORT_PRE_PROCESS_SUCCESS_COUNT = SUB_COMPONENT + "import_pre_process_success_count"; private static final String IMPORTED_PACKAGE_DURATION = SUB_COMPONENT + "imported_package_duration"; + private static final String IMPORTED_BATCH_SIZE = SUB_COMPONENT + "imported_batch_size"; + private static final String IMPORTED_BATCH_DURATION = SUB_COMPONENT + "imported_batch_duration"; private static final String IMPORT_PACKAGE_INSTALL_DURATION = SUB_COMPONENT + "package_install_duration"; private static final String IMPORT_PACKAGE_INSTALL_COUNT = SUB_COMPONENT + "package_install_count"; private static final String REMOVED_PACKAGE_DURATION = SUB_COMPONENT + "removed_package_duration"; @@ -145,6 +147,24 @@ public class SubscriberMetrics { return metricsService.timer(getMetricName(IMPORTED_PACKAGE_DURATION, tags)); } + /** + * Histogram of the number of packages imported together in one batch. + * + * @return a Sling Metrics histogram + */ + public Histogram getImportedBatchSize() { + return metricsService.histogram(getMetricName(IMPORTED_BATCH_SIZE, tags)); + } + + /** + * Timer capturing the duration in ms of batch import operations, successful or not. + * + * @return a Sling Metrics timer + */ + public Timer getImportedBatchDuration() { + return metricsService.timer(getMetricName(IMPORTED_BATCH_DURATION, tags)); + } + /** * Timer capturing the duration in ms of packages successfully removed from an editable subscriber. * diff --git a/src/main/java/org/apache/sling/distribution/journal/impl/subscriber/DistributionSubscriber.java b/src/main/java/org/apache/sling/distribution/journal/impl/subscriber/DistributionSubscriber.java index 2575fbe..f8aad40 100644 --- a/src/main/java/org/apache/sling/distribution/journal/impl/subscriber/DistributionSubscriber.java +++ b/src/main/java/org/apache/sling/distribution/journal/impl/subscriber/DistributionSubscriber.java @@ -24,11 +24,15 @@ import java.io.Closeable; import java.io.IOException; import java.net.URI; import java.time.Duration; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.Date; +import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; @@ -51,9 +55,11 @@ import org.apache.sling.distribution.journal.HandlerAdapter; import org.apache.sling.distribution.journal.MessageInfo; import org.apache.sling.distribution.journal.MessagingProvider; import org.apache.sling.distribution.journal.Reset; +import org.apache.sling.distribution.journal.bookkeeper.BatchEntry; import org.apache.sling.distribution.journal.bookkeeper.BookKeeper; import org.apache.sling.distribution.journal.bookkeeper.BookKeeperConfig; import org.apache.sling.distribution.journal.bookkeeper.BookKeeperFactory; +import org.apache.sling.distribution.journal.bookkeeper.PackageHandling; import org.apache.sling.distribution.journal.bookkeeper.SubscriberMetrics; import org.apache.sling.distribution.journal.impl.precondition.Precondition; import org.apache.sling.distribution.journal.impl.precondition.Precondition.Decision; @@ -65,6 +71,7 @@ import org.apache.sling.distribution.journal.messages.PackageStatusMessage; import org.apache.sling.distribution.journal.shared.Delay; import org.apache.sling.distribution.journal.shared.OnlyOnLeader; import org.apache.sling.distribution.journal.shared.Topics; +import org.apache.sling.distribution.journal.spi.BatchDistributionPackageBuilder; import org.apache.sling.distribution.packaging.DistributionPackageBuilder; import org.apache.sling.settings.SlingSettingsService; import org.osgi.framework.BundleContext; @@ -83,6 +90,8 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.MINUTES; import static java.util.concurrent.TimeUnit.SECONDS; import static java.util.stream.Collectors.toSet; +import static org.apache.sling.distribution.journal.messages.PackageMessage.ReqType.ADD; +import static org.apache.sling.distribution.journal.messages.PackageMessage.ReqType.DELETE; import static org.apache.sling.distribution.journal.messages.PackageMessage.ReqType.INVALIDATE; import static org.apache.sling.distribution.journal.shared.Delay.exponential; import static org.apache.sling.distribution.journal.shared.Strings.requireNotBlank; @@ -104,6 +113,7 @@ public class DistributionSubscriber { static long RETRY_DELAY_MILLIS = SECONDS.toMillis(5); static long MAX_RETRY_DELAY_MILLIS = MINUTES.toMillis(15); static long QUEUE_FETCH_DELAY_MILLIS = SECONDS.toMillis(1); + static long BATCH_IMPORTER_STOP_WAIT_MILLIS = SECONDS.toMillis(30); private static final Supplier<LongSupplier> catchAllDelays = () -> exponential(RETRY_DELAY_MILLIS, MAX_RETRY_DELAY_MILLIS); @@ -136,6 +146,14 @@ public class DistributionSubscriber { private LongSupplier catchAllDelay = catchAllDelays.get(); private final Delay delay = new Delay(); + + private int maxBatchSize = 1; + + // Only used when batching is enabled: packages read by the poller, waiting for the batch importer + private BlockingQueue<QueuedPackage> pendingPackages; + + private Thread batchImporter; + private AtomicReference<DistributionAgentState> state = new AtomicReference<DistributionAgentState>(DistributionAgentState.IDLE); @@ -207,11 +225,18 @@ public class DistributionSubscriber { long startOffset = bookKeeper.loadOffset() + 1; String assign = startOffset > 0 ? messagingProvider.assignTo(startOffset) : null; + maxBatchSize = effectiveBatchSize(config); + if (maxBatchSize > 1) { + pendingPackages = new ArrayBlockingQueue<>(maxBatchSize); + } + packagePoller = messagingProvider.createPoller( Topics.PACKAGE_TOPIC, Reset.latest, assign, - HandlerAdapter.create(PackageMessage.class, this::handlePackageMessage), + HandlerAdapter.create( + PackageMessage.class, + maxBatchSize > 1 ? this::enqueuePackageMessage : this::handlePackageMessage), HandlerAdapter.create(OffsetMessage.class, this::handleOffsetMessage)); int announceDelay = Converters.standardConverter() @@ -228,12 +253,45 @@ public class DistributionSubscriber { config.editable(), announceDelay); + if (maxBatchSize > 1) { + batchImporter = + new Thread(this::runBatchImporter, "Distribution subscriber batch importer " + subAgentName); + batchImporter.setDaemon(true); + batchImporter.start(); + } + LOG.info( - "Started Subscriber agent={} at offset={}, subscribed to agent names {}, readyCheck={}", + "Started Subscriber agent={} at offset={}, subscribed to agent names {}, readyCheck={}, maxBatchSize={}", subAgentName, startOffset, queueNames, - config.subscriberIdleCheck()); + config.subscriberIdleCheck(), + maxBatchSize); + } + + private int effectiveBatchSize(SubscriberConfiguration config) { + int requested = config.maxBatchSize(); + if (requested <= 1) { + return 1; + } + String reason = null; + if (config.editable()) { + reason = "the agent is editable"; + } else if (config.packageHandling() != PackageHandling.Off) { + reason = "packageHandling is " + config.packageHandling(); + } else if (!bookKeeper.supportsBatchImport()) { + reason = "the package builder " + pkgType + " does not implement " + + BatchDistributionPackageBuilder.class.getName(); + } + if (reason != null) { + LOG.warn( + "Subscriber agent {} is configured with maxBatchSize={} but imports packages one by one because {}", + subAgentName, + requested, + reason); + return 1; + } + return requested; } private String getFirst(String[] agentNames) { @@ -265,6 +323,7 @@ public class DistributionSubscriber { IOUtils.closeQuietly(announcer, packagePoller, idleReadyCheck, idleCheck, commandPoller); running = false; + awaitBatchImporter(); LOG.info( "Stopped Subscriber agent {}, subscribed to Publisher agent names {} with package builder {}", subAgentName, @@ -284,6 +343,155 @@ public class DistributionSubscriber { } } + /** + * Wait (bounded) for the batch importer to finish the batch it is processing. Packages still + * waiting in the in-memory queue are dropped. Their offsets have not been stored, so they are + * read again from the journal when the subscriber starts the next time. + */ + private void awaitBatchImporter() { + if (batchImporter == null) { + return; + } + try { + batchImporter.join(BATCH_IMPORTER_STOP_WAIT_MILLIS); + if (batchImporter.isAlive()) { + LOG.warn( + "Batch importer of subscriber agent {} is still running after waiting {} ms on deactivation", + subAgentName, + BATCH_IMPORTER_STOP_WAIT_MILLIS); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + /** + * Called by the poller thread when batching is enabled. Blocks while the queue is full, so the + * poller reads at most {@code maxBatchSize} packages ahead of the importer. + */ + private void enqueuePackageMessage(MessageInfo info, PackageMessage message) { + QueuedPackage queued = new QueuedPackage(info, message); + try { + while (running) { + if (pendingPackages.offer(queued, QUEUE_FETCH_DELAY_MILLIS, MILLISECONDS)) { + return; + } + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + /** + * Importer loop used when batching is enabled. Takes all packages that are already queued, up + * to {@code maxBatchSize}, so a backlog is imported in batches while single packages are + * imported without waiting. + */ + private void runBatchImporter() { + List<QueuedPackage> batch = new ArrayList<>(maxBatchSize); + while (running) { + try { + QueuedPackage first = pendingPackages.poll(QUEUE_FETCH_DELAY_MILLIS, MILLISECONDS); + if (first == null) { + continue; + } + batch.add(first); + pendingPackages.drainTo(batch, maxBatchSize - 1); + processBatch(batch); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } catch (RuntimeException e) { + // Must not happen as processing retries until success. Stop rather than losing packages. + LOG.error("Unexpected error in batch importer of subscriber agent {}", subAgentName, e); + return; + } finally { + batch.clear(); + } + } + } + + /** + * Imports consecutive batchable packages together. Any other package ends the current run, + * which is imported first, and is then processed on its own, so offsets are always stored in + * journal order. + */ + private void processBatch(List<QueuedPackage> batch) { + List<QueuedPackage> run = new ArrayList<>(batch.size()); + for (QueuedPackage queued : batch) { + if (!running) { + return; + } + if (isBatchable(queued)) { + run.add(queued); + } else { + importRun(run); + run.clear(); + handlePackageMessage(queued.info, queued.message); + } + } + importRun(run); + } + + private boolean isBatchable(QueuedPackage queued) { + PackageMessage message = queued.message; + PackageMessage.ReqType type = message.getReqType(); + if (type != ADD && type != DELETE) { + return false; + } + if (!queueNames.contains(message.getPubAgentName()) || !pkgType.equals(message.getPkgType())) { + return false; + } + try { + return !shouldRemove(queued.info.getOffset()); + } catch (PreConditionTimeoutException e) { + return false; + } + } + + private void importRun(List<QueuedPackage> run) { + if (run.isEmpty() || !running) { + return; + } + if (run.size() == 1 || !tryImportBatch(run)) { + for (QueuedPackage queued : run) { + handlePackageMessage(queued.info, queued.message); + } + } + } + + private boolean tryImportBatch(List<QueuedPackage> run) { + List<BatchEntry> entries = new ArrayList<>(run.size()); + for (QueuedPackage queued : run) { + MessageInfo info = queued.info; + entries.add(new BatchEntry(queued.message, info.getOffset(), new Date(info.getCreateTime()))); + subscriberMetrics + .getPackageJournalDistributionDuration() + .update((currentTimeMillis() - info.getCreateTime()), TimeUnit.MILLISECONDS); + } + try { + this.state.set(DistributionAgentState.RUNNING); + idleCheck.busy( + bookKeeper.getRetries(run.get(0).message.getPubAgentName()), + run.get(0).info.getCreateTime()); + bookKeeper.importPackages(entries, new Date()); + catchAllDelay = catchAllDelays.get(); + return true; + } catch (Exception e) { + LOG.warn( + "Failed to import batch of {} packages at offsets {}-{}, importing them one by one", + run.size(), + run.get(0).info.getOffset(), + run.get(run.size() - 1).info.getOffset(), + e); + return false; + } finally { + idleCheck.idle(); + this.state.set(DistributionAgentState.IDLE); + announcer.run(); + } + } + public boolean tryProcess(MessageInfo info, PackageMessage message) { if (shouldSkip(info, message)) { try { @@ -407,4 +615,14 @@ public class DistributionSubscriber { MILLISECONDS.toSeconds(PRECONDITION_TIMEOUT_MILLIS), offset); throw new PreConditionTimeoutException(msg); } + + private static final class QueuedPackage { + private final MessageInfo info; + private final PackageMessage message; + + QueuedPackage(MessageInfo info, PackageMessage message) { + this.info = info; + this.message = message; + } + } } diff --git a/src/main/java/org/apache/sling/distribution/journal/impl/subscriber/SubscriberConfiguration.java b/src/main/java/org/apache/sling/distribution/journal/impl/subscriber/SubscriberConfiguration.java index 81bf0bb..9f841c7 100644 --- a/src/main/java/org/apache/sling/distribution/journal/impl/subscriber/SubscriberConfiguration.java +++ b/src/main/java/org/apache/sling/distribution/journal/impl/subscriber/SubscriberConfiguration.java @@ -89,4 +89,12 @@ public @interface SubscriberConfiguration { @AttributeDefinition(description = "Number of ms to wait before retrying to process a package.") int acceptableAgeDiffMs() default 120 * 1000; + + @AttributeDefinition( + name = "maxBatchSize", + description = + "Maximum number of consecutive ADD and DELETE packages that are installed together with one call to the package builder." + + " Only applies if the package builder implements BatchDistributionPackageBuilder, the agent is not editable and" + + " packageHandling is Off. Default is 1 (no batching).") + int maxBatchSize() default 1; } diff --git a/src/main/java/org/apache/sling/distribution/journal/spi/BatchDistributionPackageBuilder.java b/src/main/java/org/apache/sling/distribution/journal/spi/BatchDistributionPackageBuilder.java new file mode 100644 index 0000000..45cd179 --- /dev/null +++ b/src/main/java/org/apache/sling/distribution/journal/spi/BatchDistributionPackageBuilder.java @@ -0,0 +1,67 @@ +/* + * 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.journal.spi; + +import javax.annotation.Nonnull; + +import java.util.List; + +import org.apache.sling.api.resource.ResourceResolver; +import org.apache.sling.distribution.common.DistributionException; +import org.apache.sling.distribution.packaging.DistributionPackage; +import org.apache.sling.distribution.packaging.DistributionPackageBuilder; +import org.osgi.annotation.versioning.ConsumerType; + +/** + * A {@link DistributionPackageBuilder} that can install several packages in one call. + * + * <p>A journal subscriber configured with {@code maxBatchSize > 1} hands consecutive + * ADD and DELETE packages to {@link #installPackages(ResourceResolver, List)} instead of + * installing them one by one. This lets a builder whose install step is slow (for example + * because it calls a remote system per package) process independent packages concurrently.</p> + * + * <p>Contract for implementations:</p> + * <ul> + * <li>The packages are passed in journal order. After a successful call, the result must be + * the same as installing the packages one by one in that order. Packages may only be installed + * concurrently when that cannot change the result, for example when their paths do not overlap + * (a package deleting {@code /a} and a package updating {@code /a/b} do overlap).</li> + * <li>Changes made through the given resource resolver are committed by the subscriber after the + * call returns, together with the journal offset of the last package. Implementations must not + * commit the resolver themselves.</li> + * <li>If the call throws, the subscriber does not store the offset and installs each package of + * the batch again through {@link #installPackage(ResourceResolver, DistributionPackage)}, with the + * usual retry and error-queue handling. Packages of a failed batch can therefore be installed more + * than once, so installing a package must be idempotent.</li> + * </ul> + */ +@ConsumerType +public interface BatchDistributionPackageBuilder extends DistributionPackageBuilder { + + /** + * Install the given packages. + * + * @param resourceResolver the resource resolver used to install the packages + * @param distributionPackages the packages to install, in journal order + * @throws DistributionException if any package of the batch could not be installed + */ + void installPackages( + @Nonnull ResourceResolver resourceResolver, @Nonnull List<DistributionPackage> distributionPackages) + throws DistributionException; +} diff --git a/src/main/java/org/apache/sling/distribution/journal/spi/package-info.java b/src/main/java/org/apache/sling/distribution/journal/spi/package-info.java new file mode 100644 index 0000000..8b12872 --- /dev/null +++ b/src/main/java/org/apache/sling/distribution/journal/spi/package-info.java @@ -0,0 +1,25 @@ +/* + * 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. + */ +/** + * Extension points that other bundles can implement to customise the journal subscriber. + */ +@Version("1.0.0") +package org.apache.sling.distribution.journal.spi; + +import org.osgi.annotation.versioning.Version; diff --git a/src/test/java/org/apache/sling/distribution/journal/bookkeeper/BookKeeperBatchTest.java b/src/test/java/org/apache/sling/distribution/journal/bookkeeper/BookKeeperBatchTest.java new file mode 100644 index 0000000..6d071ed --- /dev/null +++ b/src/test/java/org/apache/sling/distribution/journal/bookkeeper/BookKeeperBatchTest.java @@ -0,0 +1,175 @@ +/* + * 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.journal.bookkeeper; + +import java.util.Arrays; +import java.util.Date; +import java.util.List; +import java.util.function.Consumer; + +import org.apache.sling.api.resource.ResourceResolver; +import org.apache.sling.api.resource.ResourceResolverFactory; +import org.apache.sling.commons.metrics.MetricsService; +import org.apache.sling.commons.metrics.internal.MetricsServiceImpl; +import org.apache.sling.distribution.ImportPostProcessor; +import org.apache.sling.distribution.ImportPreProcessor; +import org.apache.sling.distribution.InvalidationProcessor; +import org.apache.sling.distribution.common.DistributionException; +import org.apache.sling.distribution.journal.messages.LogMessage; +import org.apache.sling.distribution.journal.messages.PackageMessage; +import org.apache.sling.distribution.journal.messages.PackageStatusMessage; +import org.apache.sling.testing.mock.osgi.junit.OsgiContext; +import org.apache.sling.testing.resourceresolver.MockResourceResolverFactory; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; +import org.osgi.service.event.Event; +import org.osgi.service.event.EventAdmin; + +import static java.util.Collections.singletonList; +import static org.hamcrest.CoreMatchers.equalTo; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.junit.Assert.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +@RunWith(MockitoJUnitRunner.class) +public class BookKeeperBatchTest { + + private static final String PUB_AGENT_NAME = "pubAgent"; + + private final ResourceResolverFactory resolverFactory = new MockResourceResolverFactory(); + + private final OsgiContext context = new OsgiContext(); + + @Mock + private EventAdmin eventAdmin; + + @Mock + private Consumer<PackageStatusMessage> sender; + + @Mock + private Consumer<LogMessage> logSender; + + @Mock + private PackageHandler packageHandler; + + @Mock + private ImportPreProcessor importPreProcessor; + + @Mock + private ImportPostProcessor importPostProcessor; + + @Mock + private InvalidationProcessor invalidationProcessor; + + @Test + public void testImportPackagesStoresLastOffsetOnce() throws Exception { + BookKeeper bookKeeper = createBookKeeper(false); + List<BatchEntry> entries = Arrays.asList(entry("a", 10), entry("b", 11), entry("c", 12)); + + bookKeeper.importPackages(entries, new Date()); + + verify(packageHandler) + .applyBatch( + any(ResourceResolver.class), + eq(Arrays.asList( + entries.get(0).getMessage(), + entries.get(1).getMessage(), + entries.get(2).getMessage()))); + verify(importPreProcessor, times(3)).process(any()); + verify(importPostProcessor, times(3)).process(any()); + verify(eventAdmin, times(3)).postEvent(any(Event.class)); + verifyNoInteractions(sender); + assertThat(bookKeeper.loadOffset(), equalTo(12L)); + } + + @Test + public void testFailedBatchStoresNoOffsetAndDoesNotCountRetries() throws Exception { + BookKeeper bookKeeper = createBookKeeper(false); + doThrow(new DistributionException("install failed")) + .when(packageHandler) + .applyBatch(any(ResourceResolver.class), any()); + List<BatchEntry> entries = Arrays.asList(entry("a", 10), entry("b", 11)); + + assertThrows(DistributionException.class, () -> bookKeeper.importPackages(entries, new Date())); + + assertThat(bookKeeper.loadOffset(), equalTo(-1L)); + assertThat(bookKeeper.getRetries(PUB_AGENT_NAME), equalTo(0)); + verify(importPostProcessor, never()).process(any()); + verify(eventAdmin, never()).postEvent(any(Event.class)); + verifyNoInteractions(logSender); + } + + @Test + public void testEditableSubscriberDoesNotSupportBatch() { + BookKeeper bookKeeper = createBookKeeper(true); + + assertThat(bookKeeper.supportsBatchImport(), equalTo(false)); + assertThrows( + IllegalStateException.class, + () -> bookKeeper.importPackages(singletonList(entry("a", 10)), new Date())); + } + + @Test + public void testSupportsBatchImportFollowsPackageHandler() { + when(packageHandler.supportsBatch()).thenReturn(true); + + assertThat(createBookKeeper(false).supportsBatchImport(), equalTo(true)); + } + + private BookKeeper createBookKeeper(boolean editable) { + MetricsService metricsService = context.registerInjectActivateService(MetricsServiceImpl.class); + BookKeeperConfig bkConfig = new BookKeeperConfig( + "subAgentName", "subSlingId", editable, 10, PackageHandling.Off, "package", "command", true); + SubscriberMetrics subscriberMetrics = + new SubscriberMetrics(metricsService, bkConfig.getSubAgentName(), "publish", bkConfig.isEditable()); + return new BookKeeper( + resolverFactory, + subscriberMetrics, + packageHandler, + eventAdmin, + sender, + logSender, + bkConfig, + importPreProcessor, + importPostProcessor, + invalidationProcessor); + } + + private static BatchEntry entry(String pkgId, long offset) { + PackageMessage message = PackageMessage.builder() + .pkgId(pkgId) + .pubSlingId("pubSlingId") + .pubAgentName(PUB_AGENT_NAME) + .reqType(PackageMessage.ReqType.ADD) + .pkgType("journal") + .paths(singletonList("/content/" + pkgId)) + .pkgBinary(new byte[10]) + .build(); + return new BatchEntry(message, offset, new Date()); + } +} diff --git a/src/test/java/org/apache/sling/distribution/journal/impl/subscriber/SubscriberBatchTest.java b/src/test/java/org/apache/sling/distribution/journal/impl/subscriber/SubscriberBatchTest.java new file mode 100644 index 0000000..0ddaf11 --- /dev/null +++ b/src/test/java/org/apache/sling/distribution/journal/impl/subscriber/SubscriberBatchTest.java @@ -0,0 +1,343 @@ +/* + * 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.journal.impl.subscriber; + +import java.io.Closeable; +import java.net.URI; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; + +import org.apache.sling.api.resource.ResourceResolver; +import org.apache.sling.api.resource.ResourceResolverFactory; +import org.apache.sling.commons.metrics.MetricsService; +import org.apache.sling.distribution.ImportPostProcessor; +import org.apache.sling.distribution.ImportPreProcessor; +import org.apache.sling.distribution.InvalidationProcessor; +import org.apache.sling.distribution.common.DistributionException; +import org.apache.sling.distribution.journal.HandlerAdapter; +import org.apache.sling.distribution.journal.MessageHandler; +import org.apache.sling.distribution.journal.MessageSender; +import org.apache.sling.distribution.journal.MessagingProvider; +import org.apache.sling.distribution.journal.Reset; +import org.apache.sling.distribution.journal.bookkeeper.BookKeeper; +import org.apache.sling.distribution.journal.bookkeeper.BookKeeperFactory; +import org.apache.sling.distribution.journal.bookkeeper.LocalStore; +import org.apache.sling.distribution.journal.impl.precondition.Precondition; +import org.apache.sling.distribution.journal.impl.precondition.Precondition.Decision; +import org.apache.sling.distribution.journal.messages.DiscoveryMessage; +import org.apache.sling.distribution.journal.messages.PackageMessage; +import org.apache.sling.distribution.journal.messages.PackageMessage.ReqType; +import org.apache.sling.distribution.journal.messages.PackageStatusMessage; +import org.apache.sling.distribution.journal.messages.PingMessage; +import org.apache.sling.distribution.journal.shared.NoOpImportPostProcessor; +import org.apache.sling.distribution.journal.shared.NoOpImportPreProcessor; +import org.apache.sling.distribution.journal.shared.NoOpInvalidationProcessor; +import org.apache.sling.distribution.journal.shared.OnlyOnLeader; +import org.apache.sling.distribution.journal.shared.TestMessageInfo; +import org.apache.sling.distribution.journal.shared.Topics; +import org.apache.sling.distribution.journal.spi.BatchDistributionPackageBuilder; +import org.apache.sling.distribution.packaging.DistributionPackage; +import org.apache.sling.settings.SlingSettingsService; +import org.apache.sling.testing.resourceresolver.MockResourceResolverFactory; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Captor; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.Mockito; +import org.mockito.Spy; +import org.mockito.junit.MockitoJUnitRunner; +import org.osgi.framework.BundleContext; +import org.osgi.service.event.EventAdmin; +import org.osgi.util.converter.Converters; + +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.awaitility.Awaitility.await; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.contains; +import static org.hamcrest.Matchers.equalTo; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.timeout; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@RunWith(MockitoJUnitRunner.class) +public class SubscriberBatchTest { + + private static final String SUB1_SLING_ID = "sub1sling"; + private static final String SUB1_AGENT_NAME = "sub1agent"; + private static final String PUB1_AGENT_NAME = "pub1agent"; + private static final String STORE_PACKAGE_NODE_NAME = "myserver.apache.org_somepath_package"; + + @Mock + private BundleContext context; + + @Mock + private BatchDistributionPackageBuilder packageBuilder; + + @Mock + private Precondition precondition; + + @Mock + private SlingSettingsService slingSettings; + + @Spy + private ResourceResolverFactory resolverFactory = new MockResourceResolverFactory(); + + @Mock + private MessagingProvider clientProvider; + + @Mock + private EventAdmin eventAdmin; + + @Mock + private MessageSender<DiscoveryMessage> discoverySender; + + @Mock + private MessageSender<PackageStatusMessage> statusSender; + + @Spy + private MetricsService metricsService = MetricsService.NOOP; + + @Spy + private ImportPreProcessor importPreProcessor = new NoOpImportPreProcessor(); + + @Spy + private ImportPostProcessor importPostProcessor = new NoOpImportPostProcessor(); + + @Spy + private InvalidationProcessor invalidationProcessor = new NoOpInvalidationProcessor(); + + @Spy + private SubscriberReadyStore subscriberReadyStore = new SubscriberReadyStore(); + + @InjectMocks + private BookKeeperFactory bookKeeperFactory; + + @Captor + private ArgumentCaptor<HandlerAdapter<PackageMessage>> packageCaptor; + + @Captor + private ArgumentCaptor<HandlerAdapter<PingMessage>> pingCaptor; + + @Mock + private Closeable poller; + + private DistributionSubscriber subscriber; + + private MessageHandler<PackageMessage> packageHandler; + + private final List<List<String>> installedBatches = Collections.synchronizedList(new ArrayList<>()); + + @Before + public void before() throws Exception { + DistributionSubscriber.QUEUE_FETCH_DELAY_MILLIS = 100; + DistributionSubscriber.RETRY_DELAY_MILLIS = 100; + when(packageBuilder.getType()).thenReturn("journal"); + when(slingSettings.getSlingId()).thenReturn(SUB1_SLING_ID); + when(clientProvider.getServerUri()).thenReturn(new URI("http://myserver.apache.org:1234/somepath";)); + when(clientProvider.<PackageStatusMessage>createSender(Topics.STATUS_TOPIC)) + .thenReturn(statusSender); + when(clientProvider.<DiscoveryMessage>createSender(Topics.DISCOVERY_TOPIC)) + .thenReturn(discoverySender); + when(precondition.canProcess(eq(SUB1_AGENT_NAME), anyLong())).thenReturn(Decision.ACCEPT); + } + + @After + public void after() { + subscriber.deactivate(); + } + + @Test + public void testBacklogIsImportedAsBatch() throws Exception { + recordInstalledBatches(); + CountDownLatch firstImportStarted = new CountDownLatch(1); + CountDownLatch releaseFirstImport = new CountDownLatch(1); + when(packageBuilder.installPackage(any(ResourceResolver.class), any(DistributionPackage.class))) + .thenAnswer(invocation -> { + firstImportStarted.countDown(); + releaseFirstImport.await(); + return true; + }); + initSubscriber(3); + + // The importer picks up the first package alone and blocks in the single install + packageHandler.handle(new TestMessageInfo("", 1, 0, 0), message("p0", ReqType.ADD)); + assertTrue(firstImportStarted.await(10, SECONDS)); + + // Meanwhile the next packages queue up and are imported together + packageHandler.handle(new TestMessageInfo("", 1, 1, 0), message("p1", ReqType.ADD)); + packageHandler.handle(new TestMessageInfo("", 1, 2, 0), message("p2", ReqType.DELETE)); + packageHandler.handle(new TestMessageInfo("", 1, 3, 0), message("p3", ReqType.ADD)); + releaseFirstImport.countDown(); + + await().atMost(10, SECONDS).until(this::getStoredOffset, equalTo(3L)); + assertThat(installedBatches, contains(Arrays.asList("p1", "p2", "p3"))); + verify(packageBuilder, times(1)).installPackage(any(ResourceResolver.class), any(DistributionPackage.class)); + } + + @Test + public void testFailedBatchFallsBackToSinglePackages() throws Exception { + doThrow(new DistributionException("batch failed")) + .when(packageBuilder) + .installPackages(any(ResourceResolver.class), anyList()); + CountDownLatch firstImportStarted = new CountDownLatch(1); + CountDownLatch releaseFirstImport = new CountDownLatch(1); + when(packageBuilder.installPackage(any(ResourceResolver.class), any(DistributionPackage.class))) + .thenAnswer(invocation -> { + firstImportStarted.countDown(); + releaseFirstImport.await(); + return true; + }); + initSubscriber(3); + + packageHandler.handle(new TestMessageInfo("", 1, 0, 0), message("p0", ReqType.ADD)); + assertTrue(firstImportStarted.await(10, SECONDS)); + packageHandler.handle(new TestMessageInfo("", 1, 1, 0), message("p1", ReqType.ADD)); + packageHandler.handle(new TestMessageInfo("", 1, 2, 0), message("p2", ReqType.ADD)); + releaseFirstImport.countDown(); + + await().atMost(10, SECONDS).until(this::getStoredOffset, equalTo(2L)); + verify(packageBuilder, timeout(10000).times(1)).installPackages(any(ResourceResolver.class), anyList()); + verify(packageBuilder, times(3)).installPackage(any(ResourceResolver.class), any(DistributionPackage.class)); + } + + @Test + public void testInvalidateEndsBatch() throws Exception { + recordInstalledBatches(); + CountDownLatch firstImportStarted = new CountDownLatch(1); + CountDownLatch releaseFirstImport = new CountDownLatch(1); + when(packageBuilder.installPackage(any(ResourceResolver.class), any(DistributionPackage.class))) + .thenAnswer(invocation -> { + firstImportStarted.countDown(); + releaseFirstImport.await(); + return true; + }); + initSubscriber(10); + + packageHandler.handle(new TestMessageInfo("", 1, 0, 0), message("p0", ReqType.ADD)); + assertTrue(firstImportStarted.await(10, SECONDS)); + packageHandler.handle(new TestMessageInfo("", 1, 1, 0), message("p1", ReqType.ADD)); + packageHandler.handle(new TestMessageInfo("", 1, 2, 0), message("p2", ReqType.ADD)); + packageHandler.handle(new TestMessageInfo("", 1, 3, 0), message("p3", ReqType.INVALIDATE)); + packageHandler.handle(new TestMessageInfo("", 1, 4, 0), message("p4", ReqType.ADD)); + packageHandler.handle(new TestMessageInfo("", 1, 5, 0), message("p5", ReqType.ADD)); + releaseFirstImport.countDown(); + + await().atMost(10, SECONDS).until(this::getStoredOffset, equalTo(5L)); + assertThat(installedBatches, contains(Arrays.asList("p1", "p2"), Arrays.asList("p4", "p5"))); + } + + @Test + public void testNoBatchingForEditableAgent() throws Exception { + Map<String, String> overrides = new HashMap<>(); + overrides.put("editable", "true"); + when(clientProvider.createPoller(Mockito.eq(Topics.COMMAND_TOPIC), Mockito.eq(Reset.earliest), any())) + .thenReturn(poller); + initSubscriber(3, overrides); + + packageHandler.handle(new TestMessageInfo("", 1, 0, 0), message("p0", ReqType.ADD)); + packageHandler.handle(new TestMessageInfo("", 1, 1, 0), message("p1", ReqType.ADD)); + + assertThat(getStoredOffset(), equalTo(1L)); + verify(packageBuilder, times(2)).installPackage(any(ResourceResolver.class), any(DistributionPackage.class)); + verify(packageBuilder, never()).installPackages(any(ResourceResolver.class), anyList()); + } + + private void recordInstalledBatches() throws DistributionException { + doAnswer(invocation -> { + List<DistributionPackage> packages = invocation.getArgument(1); + List<String> ids = new ArrayList<>(); + for (DistributionPackage pkg : packages) { + ids.add(pkg.getId()); + } + installedBatches.add(ids); + return null; + }) + .when(packageBuilder) + .installPackages(any(ResourceResolver.class), anyList()); + } + + private static PackageMessage message(String pkgId, ReqType reqType) { + return PackageMessage.builder() + .pkgId(pkgId) + .pubSlingId("pub1sling") + .pubAgentName(PUB1_AGENT_NAME) + .reqType(reqType) + .pkgType("journal") + .paths(Collections.singletonList("/content/" + pkgId)) + .pkgBinary(new byte[10]) + .build(); + } + + private Long getStoredOffset() { + LocalStore store = new LocalStore(resolverFactory, STORE_PACKAGE_NODE_NAME, SUB1_AGENT_NAME); + return store.load(BookKeeper.KEY_OFFSET, Long.class); + } + + private void initSubscriber(int maxBatchSize) { + initSubscriber(maxBatchSize, Collections.emptyMap()); + } + + private void initSubscriber(int maxBatchSize, Map<String, String> overrides) { + Map<String, Object> props = new HashMap<>(); + props.put("name", SUB1_AGENT_NAME); + props.put("agentNames", PUB1_AGENT_NAME); + props.put("maxBatchSize", maxBatchSize); + props.putAll(overrides); + SubscriberConfiguration config = + Converters.standardConverter().convert(props).to(SubscriberConfiguration.class); + subscriber = new DistributionSubscriber( + packageBuilder, + slingSettings, + clientProvider, + precondition, + metricsService, + bookKeeperFactory, + subscriberReadyStore, + new OnlyOnLeader(context), + config, + context, + props); + verify(clientProvider) + .createPoller( + Mockito.eq(Topics.PACKAGE_TOPIC), + Mockito.eq(Reset.latest), + Mockito.isNull(String.class), + packageCaptor.capture(), + pingCaptor.capture()); + packageHandler = packageCaptor.getValue().getHandler(); + } +}
