This is an automated email from the ASF dual-hosted git repository.
cschneider 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 85b8b86 SLING-13403 - Wait boundedly for in-flight imports on
subscriber deactivation (#192)
85b8b86 is described below
commit 85b8b863aeef511a858cdf65a5330d2612bfb772
Author: Christian Schneider <[email protected]>
AuthorDate: Sun Nov 1 22:21:45 2026 +0100
SLING-13403 - Wait boundedly for in-flight imports on subscriber
deactivation (#192)
---
.../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());
}
