Merge branch 'master' into SLING-8908
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 9dd0d70..c77d588 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
@@ -298,16 +298,34 @@
}
private void handlePackageMessage(MessageInfo info, PackageMessage message) {
+ if (shouldEnqueue(message)) {
+ try {
+ DistributionQueueItem queueItem = QueueItemFactory.fromPackage(info, message, true);
+ enqueue(queueItem);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException();
+ }
+ } else {
+ try (ResourceResolver resolver = getServiceResolver("bookkeeper")) {
+ storeOffset(resolver, info.getOffset());
+ resolver.commit();
+ } catch (LoginException | PersistenceException e) {
+ LOG.warn("Error storing offset", e);
+ }
+ }
+ }
+
+ private boolean shouldEnqueue(PackageMessage message) {
if (!queueNames.contains(message.getPubAgentName())) {
LOG.info(String.format("Skipping package for Publisher agent %s (not subscribed)", message.getPubAgentName()));
- return;
+ return false;
}
if (!pkgType.equals(message.getPkgType())) {
LOG.warn(String.format("Skipping package with type %s", message.getPkgType()));
- return;
+ return false;
}
- DistributionQueueItem queueItem = QueueItemFactory.fromPackage(info, message, true);
- enqueue(queueItem);
+ return true;
}
/**
