[sling-org-apache-sling-distribution-journal] branch master updated: SLING-9482 - Fallback to seeding messages when no seed if found in the repository

[tm...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

tmaret pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-distribution-journal.git


The following commit(s) were added to refs/heads/master by this push:
     new e2714aa  SLING-9482 - Fallback to seeding messages when no seed if found in the repository
e2714aa is described below

commit e2714aafcc7aaf35aa3ae6f847fcd97b2221a0a8
Author: tmaret <tm...@adobe.com>
AuthorDate: Fri May 29 13:27:50 2020 +0200

    SLING-9482 - Fallback to seeding messages when no seed if found in the repository
---
 .../journal/impl/queue/impl/PubQueueCache.java     | 31 ++++++++++++++++------
 1 file changed, 23 insertions(+), 8 deletions(-)

diff --git a/src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueCache.java b/src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueCache.java
index 2b04a4a..b2f0135 100644
--- a/src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueCache.java
+++ b/src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueCache.java
@@ -130,7 +130,18 @@ public class PubQueueCache {
         this.distributionMetricsService = distributionMetricsService;
         this.topic = topic;
         this.seedStore = seedStore;
-        seedOffset = seedStore.load("offset", 0);
+        Long offset = seedStore.load().get("offset", Long.class);
+        if (offset != null) {
+            seedOffset = offset;
+        } else {
+            /*
+             * Fallback to seeding messages when
+             * no offset could be found in the
+             * repository.
+             */
+            seedOffset = messagingProvider.retrieveOffset(topic, Reset.latest);
+            storeSeed(seedOffset);
+        }
         seed(seedOffset);
     }
 
@@ -160,13 +171,17 @@ public class PubQueueCache {
     public void storeSeed() {
         long newSeed = maxOffset.longValue();
         if (newSeed > seedOffset) {
-            LOG.info("Store seed offset {}", newSeed);
-            try {
-                seedStore.store("offset", newSeed);
-                seedOffset = newSeed;
-            } catch (PersistenceException e) {
-                LOG.warn("Failed to persist seed offset", e);
-            }
+            storeSeed(newSeed);
+            seedOffset = newSeed;
+        }
+    }
+
+    private void storeSeed(long offset) {
+        LOG.info("Store seed offset {}", offset);
+        try {
+            seedStore.store("offset", offset);
+        } catch (PersistenceException e) {
+            LOG.warn("Failed to persist seed offset", e);
         }
     }