This is an automated email from the ASF dual-hosted git repository. cschneider pushed a commit to branch SLING-13395-duplicate-import-on-restart in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-distribution-journal.git
commit 8de4d400030e9bfe77da23b23f060194af3907f9 Author: Christian Schneider <[email protected]> AuthorDate: Thu Oct 22 12:29:01 2026 +0200 SLING-13395 - Avoid duplicate imports when the subscriber is restarted OnlyOnLeader no longer unregisters on TOPOLOGY_CHANGING. It updates its registration only once the new view is known (TOPOLOGY_INIT/CHANGED), so DistributionSubscriber and other leader-only components are no longer deactivated and re-activated on every topology change when the leader stays the same. DistributionSubscriber.deactivate() now wakes up a pending retry back-off and waits up to 30 s for a package that is still being processed. Before, a re-activated subscriber loaded the last committed offset and imported the same package again in parallel with the import still running. Co-authored-by: Copilot App <[email protected]> --- .../impl/subscriber/DistributionSubscriber.java | 40 ++++++++++++-- .../distribution/journal/shared/OnlyOnLeader.java | 5 +- .../journal/impl/subscriber/SubscriberTest.java | 61 ++++++++++++++++++++++ .../journal/shared/OnlyOnLeaderTest.java | 13 ++++- 4 files changed, 113 insertions(+), 6 deletions(-) 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..d5018dd 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 @@ -32,6 +32,7 @@ import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.ReentrantLock; import java.util.function.BiConsumer; import java.util.function.Consumer; import java.util.function.LongSupplier; @@ -104,6 +105,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 DEACTIVATE_IMPORT_WAIT_MILLIS = SECONDS.toMillis(30); private static final Supplier<LongSupplier> catchAllDelays = () -> exponential(RETRY_DELAY_MILLIS, MAX_RETRY_DELAY_MILLIS); @@ -136,6 +138,9 @@ public class DistributionSubscriber { private LongSupplier catchAllDelay = catchAllDelays.get(); private final Delay delay = new Delay(); + + // Held by the poller thread while a package message is processed + private final ReentrantLock processingLock = new ReentrantLock(); private AtomicReference<DistributionAgentState> state = new AtomicReference<DistributionAgentState>(DistributionAgentState.IDLE); @@ -265,6 +270,9 @@ public class DistributionSubscriber { IOUtils.closeQuietly(announcer, packagePoller, idleReadyCheck, idleCheck, commandPoller); running = false; + // Wake up a retry back-off so the processing loop can exit without waiting it out + delay.signal(); + awaitInFlightImport(); LOG.info( "Stopped Subscriber agent {}, subscribed to Publisher agent names {} with package builder {}", subAgentName, @@ -272,15 +280,41 @@ public class DistributionSubscriber { pkgType); } + /** + * Wait (bounded) for a package that is currently being processed. Otherwise a subscriber + * activated right after this one would load the same offset and import the package again + * while it is still being imported here. + */ + private void awaitInFlightImport() { + try { + if (processingLock.tryLock(DEACTIVATE_IMPORT_WAIT_MILLIS, MILLISECONDS)) { + processingLock.unlock(); + } else { + LOG.warn( + "Subscriber agent {} is still processing a package after waiting {} ms on deactivation." + + " A new subscriber instance may import the same package again.", + subAgentName, + DEACTIVATE_IMPORT_WAIT_MILLIS); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + public DistributionAgentState getState() { boolean isBlocked = bookKeeper.getPackageRetries().getSum() > 0; return (isBlocked) ? DistributionAgentState.BLOCKED : state.get(); } private void handlePackageMessage(MessageInfo info, PackageMessage message) { - boolean done = false; - while (!done && running) { - done = tryProcess(info, message); + processingLock.lock(); + try { + boolean done = false; + while (!done && running) { + done = tryProcess(info, message); + } + } finally { + processingLock.unlock(); } } diff --git a/src/main/java/org/apache/sling/distribution/journal/shared/OnlyOnLeader.java b/src/main/java/org/apache/sling/distribution/journal/shared/OnlyOnLeader.java index d2b2d83..41960bd 100644 --- a/src/main/java/org/apache/sling/distribution/journal/shared/OnlyOnLeader.java +++ b/src/main/java/org/apache/sling/distribution/journal/shared/OnlyOnLeader.java @@ -61,14 +61,15 @@ public class OnlyOnLeader implements TopologyEventListener { @Override public void handleTopologyEvent(TopologyEvent event) { TopologyEvent.Type eventType = event.getType(); + // TOPOLOGY_CHANGING is deliberately ignored. Unregistering on it would deactivate and + // re-activate every dependent component on each topology change, even when the leader + // does not change. The registration is only updated once the new view is known. if (eventType == TOPOLOGY_INIT || eventType == TOPOLOGY_CHANGED) { if (event.getNewView().getLocalInstance().isLeader()) { register(); } else { unregister(); } - } else if (eventType == TOPOLOGY_CHANGING) { - unregister(); } } diff --git a/src/test/java/org/apache/sling/distribution/journal/impl/subscriber/SubscriberTest.java b/src/test/java/org/apache/sling/distribution/journal/impl/subscriber/SubscriberTest.java index f532b91..ec1fc81 100644 --- a/src/test/java/org/apache/sling/distribution/journal/impl/subscriber/SubscriberTest.java +++ b/src/test/java/org/apache/sling/distribution/journal/impl/subscriber/SubscriberTest.java @@ -91,6 +91,8 @@ import static org.hamcrest.CoreMatchers.nullValue; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.lessThan; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.eq; @@ -209,6 +211,7 @@ public class SubscriberTest { public void before() throws URISyntaxException { DistributionSubscriber.QUEUE_FETCH_DELAY_MILLIS = 100; DistributionSubscriber.RETRY_DELAY_MILLIS = 100; + DistributionSubscriber.DEACTIVATE_IMPORT_WAIT_MILLIS = 30000; Awaitility.setDefaultPollDelay(Duration.ZERO); Awaitility.setDefaultPollInterval(Duration.ONE_HUNDRED_MILLISECONDS); @@ -406,6 +409,64 @@ public class SubscriberTest { thread.join(); } + @Test + public void testDeactivateWaitsForInFlightImport() throws Exception { + Semaphore sem = new Semaphore(0); + assumeWaitingForPrecondition(sem); + initSubscriber(); + + Thread poller = new Thread(() -> packageHandler.handle(createInfo(0L), BASIC_ADD_PACKAGE)); + poller.start(); + verify(precondition, timeout(10000)).canProcess(eq(SUB1_AGENT_NAME), anyLong()); + + Thread deactivator = new Thread(subscriber::deactivate); + deactivator.start(); + deactivator.join(500); + assertTrue("deactivate must wait for the in-flight import", deactivator.isAlive()); + + sem.release(); + deactivator.join(10000); + poller.join(10000); + assertFalse(deactivator.isAlive()); + } + + @Test + public void testDeactivateWaitForInFlightImportIsBounded() throws Exception { + DistributionSubscriber.DEACTIVATE_IMPORT_WAIT_MILLIS = 200; + Semaphore sem = new Semaphore(0); + assumeWaitingForPrecondition(sem); + initSubscriber(); + + Thread poller = new Thread(() -> packageHandler.handle(createInfo(0L), BASIC_ADD_PACKAGE)); + poller.start(); + verify(precondition, timeout(10000)).canProcess(eq(SUB1_AGENT_NAME), anyLong()); + + long startedAt = System.currentTimeMillis(); + subscriber.deactivate(); + assertThat(System.currentTimeMillis() - startedAt, lessThan(5000L)); + assertTrue("import is still running", poller.isAlive()); + + sem.release(); + poller.join(10000); + } + + @Test + public void testDeactivateDoesNotWaitOutRetryBackoff() throws Exception { + DistributionSubscriber.RETRY_DELAY_MILLIS = 60000; + when(precondition.canProcess(eq(SUB1_AGENT_NAME), anyLong())).thenThrow(new IllegalStateException("fail")); + initSubscriber(); + + Thread poller = new Thread(() -> packageHandler.handle(createInfo(0L), BASIC_ADD_PACKAGE)); + poller.start(); + verify(precondition, timeout(10000)).canProcess(eq(SUB1_AGENT_NAME), anyLong()); + + long startedAt = System.currentTimeMillis(); + subscriber.deactivate(); + poller.join(10000); + assertThat(System.currentTimeMillis() - startedAt, lessThan(5000L)); + assertFalse(poller.isAlive()); + } + private void verifyNoStatusMessageSent() { verify(statusSender, times(0)).accept(any()); } diff --git a/src/test/java/org/apache/sling/distribution/journal/shared/OnlyOnLeaderTest.java b/src/test/java/org/apache/sling/distribution/journal/shared/OnlyOnLeaderTest.java index 380eaa1..6cd14b8 100644 --- a/src/test/java/org/apache/sling/distribution/journal/shared/OnlyOnLeaderTest.java +++ b/src/test/java/org/apache/sling/distribution/journal/shared/OnlyOnLeaderTest.java @@ -75,11 +75,22 @@ public class OnlyOnLeaderTest { public void testServiceDeRegistration() { onlyOnLeader.handleTopologyEvent(topologyEvent(TOPOLOGY_CHANGED, true)); assertTrue(onlyOnLeader.isLeader()); - onlyOnLeader.handleTopologyEvent(topologyEvent(TOPOLOGY_CHANGING, true)); + onlyOnLeader.handleTopologyEvent(topologyEvent(TOPOLOGY_CHANGED, false)); verify(reg).unregister(); assertFalse(onlyOnLeader.isLeader()); } + @Test + public void testTopologyChangingKeepsRegistration() { + onlyOnLeader.handleTopologyEvent(topologyEvent(TOPOLOGY_CHANGED, true)); + onlyOnLeader.handleTopologyEvent(topologyEvent(TOPOLOGY_CHANGING, true)); + assertTrue(onlyOnLeader.isLeader()); + onlyOnLeader.handleTopologyEvent(topologyEvent(TOPOLOGY_CHANGED, true)); + assertTrue(onlyOnLeader.isLeader()); + verify(reg, never()).unregister(); + verify(context, times(1)).registerService(eq(OnlyOnLeader.class), eq(onlyOnLeader), any()); + } + @Test public void testTopologyChangedOnLeader() { assertFalse(onlyOnLeader.isLeader());
