You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sling.apache.org by cs...@apache.org on 2026/10/22 10:35:24 UTC
(sling-org-apache-sling-distribution-journal) 01/01: SLING-13395 - Avoid duplicate imports when the subscriber is restarted
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 <cs...@adobe.com>
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 <22...@users.noreply.github.com>
---
.../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());