This is an automated email from the ASF dual-hosted git repository. cschneider pushed a commit to branch SLING-13403-bounded-subscriber-deactivation in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-distribution-journal.git
commit ce10c605db5c0e7f3139c46a5e628dddb54fcf30 Author: Christian Schneider <[email protected]> AuthorDate: Wed Oct 28 14:48:04 2026 +0100 fix: SLING-13403 - Wait boundedly for in-flight imports on deactivation Co-authored-by: Copilot App <[email protected]> --- .../impl/subscriber/DistributionSubscriber.java | 40 ++++++++++++-- .../journal/impl/subscriber/SubscriberTest.java | 61 ++++++++++++++++++++++ 2 files changed, 98 insertions(+), 3 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/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()); }
