This is an automated email from the ASF dual-hosted git repository.
chrisdutz pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
The following commit(s) were added to refs/heads/develop by this push:
new afa9d43b4d fix: Addressed several potential issues in the event-pump.
afa9d43b4d is described below
commit afa9d43b4d1d9baa3b6722e17d9b191a4cfef084
Author: Christofer Dutz <[email protected]>
AuthorDate: Thu Aug 13 10:19:40 2026 +0200
fix: Addressed several potential issues in the event-pump.
---
.../plc4x/java/tools/eventpump/TagBatch.java | 88 ++++++++++---
.../tools/eventpump/config/BatchConfiguration.java | 26 ++++
.../tools/eventpump/config/EventPumpFactory.java | 4 +
.../tools/eventpump/triggers/TimerTrigger.java | 55 +++++++--
.../plc4x/java/tools/eventpump/TagBatchTest.java | 137 +++++++++++++++++++++
.../tools/eventpump/triggers/TimerTriggerTest.java | 86 +++++++++++++
6 files changed, 370 insertions(+), 26 deletions(-)
diff --git
a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/TagBatch.java
b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/TagBatch.java
index 7f30e54dab..fec1e70dc3 100644
---
a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/TagBatch.java
+++
b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/TagBatch.java
@@ -33,6 +33,7 @@ import org.slf4j.LoggerFactory;
import java.util.*;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
/**
@@ -84,8 +85,20 @@ public class TagBatch implements AutoCloseable {
private static final long DEFAULT_INITIAL_BACKOFF_MS = 1_000;
private final long maxBackoffMs;
private final long initialBackoffMs;
- private int consecutiveFailures;
- private long nextAllowedFetchTimeMs;
+ // Written from whichever thread completes a fetch, read from the trigger
+ // thread in fetchTags() — must be safely published.
+ private volatile int consecutiveFailures;
+ private volatile long nextAllowedFetchTimeMs;
+ private volatile long currentBackoffMs;
+
+ // Upper bound on a single fetch cycle. This is NOT a replacement for the
driver's
+ // own request timeout (configure that in the connection string, e.g.
+ // "?request-timeout=10000") — it is deliberately set an order of
magnitude higher,
+ // and exists only so that a driver which never completes its read future
cannot
+ // wedge this batch forever: fetchInProgress would stay true and every
later trigger
+ // would be skipped, silently killing the batch with no recovery.
+ private static final long DEFAULT_FETCH_TIMEOUT_MS = 300_000;
+ private final long fetchTimeoutMs;
// Guard against overlapping fetch cycles. If a previous fetch hasn't
completed
// when the next trigger fires, the new fetch is skipped to prevent queue
pile-up.
@@ -126,7 +139,7 @@ public class TagBatch implements AutoCloseable {
Map<String, String> tags, Map<String, String> transforms,
Trigger trigger,
TagBatchListener listener, ValueTransformer
valueTransformer,
ValueTransformerRegistry transformerRegistry,
- long maxBackoffMs, long initialBackoffMs) {
+ long maxBackoffMs, long initialBackoffMs, long
fetchTimeoutMs) {
if (batchId == null || batchId.trim().isEmpty()) {
throw new IllegalArgumentException("Batch ID cannot be null or
empty");
}
@@ -167,6 +180,8 @@ public class TagBatch implements AutoCloseable {
this.initialBackoffMs = initialBackoffMs;
this.consecutiveFailures = 0;
this.nextAllowedFetchTimeMs = 0;
+ this.currentBackoffMs = 0;
+ this.fetchTimeoutMs = fetchTimeoutMs;
LOGGER.debug("Created TagBatch '{}' with {} tags ({} with
transformations)",
batchId, tags.size(), this.transforms.size());
@@ -205,6 +220,25 @@ public class TagBatch implements AutoCloseable {
private ValueTransformerRegistry transformerRegistry;
private long maxBackoffMs = DEFAULT_MAX_BACKOFF_MS;
private long initialBackoffMs = DEFAULT_INITIAL_BACKOFF_MS;
+ private long fetchTimeoutMs = DEFAULT_FETCH_TIMEOUT_MS;
+
+ /**
+ * Set the upper bound for a single fetch cycle.
+ * <p>
+ * This is a watchdog, not a request timeout: it only exists so a
driver that never
+ * completes its read future cannot wedge the batch permanently.
Configure the actual
+ * request timeout on the connection string instead (for example
+ * {@code "opcua:tcp://host:4840?request-timeout=10000"}), and leave
this comfortably
+ * above it — the default is 5 minutes.
+ *
+ * @param timeout The timeout, or a value <= 0 to disable the
watchdog entirely
+ * @param unit The time unit of the timeout
+ * @return This builder
+ */
+ public Builder withFetchTimeout(long timeout, TimeUnit unit) {
+ this.fetchTimeoutMs = timeout <= 0 ? 0 : unit.toMillis(timeout);
+ return this;
+ }
/**
* Set the batch ID.
@@ -379,7 +413,8 @@ public class TagBatch implements AutoCloseable {
// Note: listener is optional and can be null
return new TagBatch(batchId, connectionManager, connectionString,
tagAddresses, transforms,
- trigger, listener, valueTransformer, transformerRegistry,
maxBackoffMs, initialBackoffMs);
+ trigger, listener, valueTransformer, transformerRegistry,
maxBackoffMs, initialBackoffMs,
+ fetchTimeoutMs);
}
}
@@ -511,8 +546,15 @@ public class TagBatch implements AutoCloseable {
LOGGER.debug("Batch '{}' - Executing read request", batchId);
CompletableFuture<? extends PlcReadResponse> readFuture =
request.execute();
+ // Bound the cycle so a driver that never completes its future
cannot wedge
+ // this batch forever. This does not cancel or shorten the
driver's own
+ // request timeout — it fires only long after that should have.
+ CompletableFuture<? extends PlcReadResponse> guardedFuture =
fetchTimeoutMs > 0
+ ? readFuture.orTimeout(fetchTimeoutMs,
TimeUnit.MILLISECONDS)
+ : readFuture;
+
// Process the response and close the connection afterward
- return readFuture
+ return guardedFuture
.whenComplete((response, throwable) -> {
try {
if (throwable != null) {
@@ -642,20 +684,24 @@ public class TagBatch implements AutoCloseable {
/**
* Get the tags being fetched.
+ * <p>
+ * Returns a snapshot: the tag map is mutable at runtime, so handing out a
view of the
+ * live map would let a caller iterating it collide with a concurrent
+ * {@link #addTag(String, String)} and see a
ConcurrentModificationException.
*
- * @return An unmodifiable map of tag names to addresses
+ * @return An unmodifiable snapshot of tag names to addresses
*/
- public Map<String, String> getTags() {
- return Collections.unmodifiableMap(tags);
+ public synchronized Map<String, String> getTags() {
+ return Collections.unmodifiableMap(new LinkedHashMap<>(tags));
}
/**
* Get the names of all tags in the batch.
*
- * @return An unmodifiable set of tag names
+ * @return An unmodifiable snapshot of the tag names
*/
public synchronized Set<String> getTagNames() {
- return Collections.unmodifiableSet(tags.keySet());
+ return Collections.unmodifiableSet(new LinkedHashSet<>(tags.keySet()));
}
/**
@@ -860,14 +906,27 @@ public class TagBatch implements AutoCloseable {
/**
* Enter or increase backoff after a failed fetch attempt.
- * Uses exponential backoff: initialBackoffMs * 2^(failures-1), capped at
maxBackoffMs.
+ * Uses exponential backoff: the window doubles per consecutive failure,
capped at
+ * maxBackoffMs.
+ * <p>
+ * The window is derived by doubling the previous one rather than
evaluating
+ * {@code initialBackoffMs * (1L << (failures - 1))}. That expression is
unsafe for a
+ * long outage: the multiplication overflows to a negative value from the
55th
+ * consecutive failure (with the 1s/60s defaults), which puts the next
allowed fetch
+ * time in the past and disables the backoff altogether, and the shift
itself wraps at
+ * 64 (Java masks long shift distances to {@code & 63}), restarting the
ramp at the
+ * initial window. Both turn a long outage back into full-rate hammering
of a PLC that
+ * is already in trouble. Doubling a value that is capped at maxBackoffMs
every step
+ * cannot overflow.
*/
private void enterBackoff() {
consecutiveFailures++;
- long backoffMs = Math.min(initialBackoffMs * (1L <<
(consecutiveFailures - 1)), maxBackoffMs);
- nextAllowedFetchTimeMs = System.currentTimeMillis() + backoffMs;
+ long previous = currentBackoffMs;
+ long next = (previous <= 0) ? initialBackoffMs : previous * 2;
+ currentBackoffMs = Math.min(next, maxBackoffMs);
+ nextAllowedFetchTimeMs = System.currentTimeMillis() + currentBackoffMs;
LOGGER.info("Batch '{}' entering backoff: next attempt in {}ms
(failure {})",
- batchId, backoffMs, consecutiveFailures);
+ batchId, currentBackoffMs, consecutiveFailures);
}
/**
@@ -878,6 +937,7 @@ public class TagBatch implements AutoCloseable {
LOGGER.info("Batch '{}' backoff reset after {} failures", batchId,
consecutiveFailures);
consecutiveFailures = 0;
nextAllowedFetchTimeMs = 0;
+ currentBackoffMs = 0;
}
}
diff --git
a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/BatchConfiguration.java
b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/BatchConfiguration.java
index fcb74e0b8d..c0973247b2 100644
---
a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/BatchConfiguration.java
+++
b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/BatchConfiguration.java
@@ -68,6 +68,14 @@ public class BatchConfiguration {
@JsonProperty("trigger")
private TriggerConfiguration trigger;
+ /**
+ * Watchdog bound for a single fetch cycle, in milliseconds. Null means
"use the
+ * default". This is not the request timeout — set that on the connection
URL, e.g.
+ * {@code "opcua:tcp://host:4840?request-timeout=10000"}.
+ */
+ @JsonProperty("fetchTimeoutMs")
+ private Long fetchTimeoutMs;
+
/**
* Get the batch ID.
*
@@ -164,4 +172,22 @@ public class BatchConfiguration {
public void setTrigger(TriggerConfiguration trigger) {
this.trigger = trigger;
}
+
+ /**
+ * Get the fetch watchdog timeout in milliseconds.
+ *
+ * @return The timeout, or null to use the default
+ */
+ public Long getFetchTimeoutMs() {
+ return fetchTimeoutMs;
+ }
+
+ /**
+ * Set the fetch watchdog timeout in milliseconds.
+ *
+ * @param fetchTimeoutMs The timeout, 0 or less to disable, or null to use
the default
+ */
+ public void setFetchTimeoutMs(Long fetchTimeoutMs) {
+ this.fetchTimeoutMs = fetchTimeoutMs;
+ }
}
diff --git
a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/EventPumpFactory.java
b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/EventPumpFactory.java
index f05fa6656c..a3a121a9fb 100644
---
a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/EventPumpFactory.java
+++
b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/EventPumpFactory.java
@@ -123,6 +123,10 @@ public class EventPumpFactory {
.withTrigger(trigger)
.withTransformerRegistry(registry); // Share registry across
all batches
+ if (batchConfig.getFetchTimeoutMs() != null) {
+ batchBuilder.withFetchTimeout(batchConfig.getFetchTimeoutMs(),
TimeUnit.MILLISECONDS);
+ }
+
// Add transformations
for (Map.Entry<String, TagConfiguration> entry :
batchConfig.getTags().entrySet()) {
if (entry.getValue().hasTransform()) {
diff --git
a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/triggers/TimerTrigger.java
b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/triggers/TimerTrigger.java
index 23d4d8888c..db38990625 100644
---
a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/triggers/TimerTrigger.java
+++
b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/triggers/TimerTrigger.java
@@ -24,6 +24,8 @@ import org.slf4j.LoggerFactory;
import java.util.Timer;
import java.util.TimerTask;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
@@ -48,6 +50,14 @@ public class TimerTrigger implements Trigger {
private final Timer timer;
private final boolean ownTimer;
+ // Listener invocations are handed to this executor rather than run on the
timer
+ // thread. A listener may block (leasing a connection from the connection
manager
+ // does), and blocking the timer thread delays this trigger's own next
tick — and,
+ // when a Timer is shared across triggers, every other batch on it as well.
+ // The queue holds a single pending firing and the rest are discarded: a
listener
+ // that outruns the interval must not build an unbounded backlog.
+ private final ThreadPoolExecutor dispatcher;
+
private volatile TriggerListener listener;
private volatile TimerTask currentTask;
private volatile boolean running = false;
@@ -102,6 +112,15 @@ public class TimerTrigger implements Trigger {
this.ownTimer = true;
}
+ this.dispatcher = new ThreadPoolExecutor(1, 1, 0L,
TimeUnit.MILLISECONDS,
+ new ArrayBlockingQueue<>(1),
+ runnable -> {
+ Thread thread = new Thread(runnable, "TimerTrigger-dispatch-"
+ System.identityHashCode(this));
+ thread.setDaemon(true);
+ return thread;
+ },
+ new ThreadPoolExecutor.DiscardPolicy());
+
LOGGER.debug("Created TimerTrigger with interval={}ms,
initialDelay={}ms", intervalMs, initialDelayMs);
}
@@ -119,25 +138,36 @@ public class TimerTrigger implements Trigger {
this.listener = listener;
- // Create the timer task
+ // Create the timer task. It only hands the firing off to the
dispatcher, so the
+ // timer thread itself never runs listener code and never blocks.
currentTask = new TimerTask() {
@Override
public void run() {
- try {
- LOGGER.trace("Timer trigger firing");
- TimerTrigger.this.listener.onTrigger(TimerTrigger.this);
- } catch (Exception e) {
- if (LOGGER.isTraceEnabled()) {
- LOGGER.error("Error in trigger listener", e);
- } else {
- LOGGER.error("Error in trigger listener: {}",
e.getMessage());
- }
+ if (!running) {
+ return;
}
+ dispatcher.execute(() -> {
+ if (!running) {
+ return;
+ }
+ try {
+ LOGGER.trace("Timer trigger firing");
+
TimerTrigger.this.listener.onTrigger(TimerTrigger.this);
+ } catch (Exception e) {
+ if (LOGGER.isTraceEnabled()) {
+ LOGGER.error("Error in trigger listener", e);
+ } else {
+ LOGGER.error("Error in trigger listener: {}",
e.getMessage());
+ }
+ }
+ });
}
};
- // Schedule the task
- timer.scheduleAtFixedRate(currentTask, initialDelayMs, intervalMs);
+ // Fixed delay rather than fixed rate: if a fetch cycle overruns the
interval,
+ // fixed-rate scheduling fires the missed ticks back-to-back in a
burst the moment
+ // it catches up, which just piles more pressure on a PLC that is
already slow.
+ timer.schedule(currentTask, initialDelayMs, intervalMs);
running = true;
LOGGER.debug("Started TimerTrigger");
@@ -179,6 +209,7 @@ public class TimerTrigger implements Trigger {
if (ownTimer) {
timer.cancel();
}
+ dispatcher.shutdown();
closed = true;
LOGGER.debug("Closed TimerTrigger");
diff --git
a/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/TagBatchTest.java
b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/TagBatchTest.java
index 561ca51090..b7be49686f 100644
---
a/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/TagBatchTest.java
+++
b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/TagBatchTest.java
@@ -68,6 +68,143 @@ class TagBatchTest {
connectionString = "test://localhost";
}
+ /**
+ * A long outage must not run the backoff off the end of its arithmetic.
The previous
+ * formula (initialBackoffMs * (1L << failures-1)) overflowed to a
negative window
+ * from the 55th consecutive failure — putting the next allowed fetch time
in the past
+ * and disabling the backoff entirely — and wrapped at 64 back to the
initial window.
+ */
+ @Test
+ void backoffStaysCappedAcrossALongOutage() throws Exception {
+ long initialBackoffMs = 1;
+ long maxBackoffMs = 4;
+
+ PlcConnectionManager failingManager = mock(PlcConnectionManager.class);
+ when(failingManager.getConnection(any())).thenThrow(new
RuntimeException("Connection refused"));
+
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("outage")
+ .withConnectionManager(failingManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(new TimerTrigger(1, TimeUnit.HOURS))
+ .withListener((b, r) -> {})
+ .withInitialBackoffMs(initialBackoffMs)
+ .withMaxBackoffMs(maxBackoffMs)
+ .build();
+
+ // Drive well past both failure counts where the old arithmetic broke
(55 and 65).
+ for (int failure = 1; failure <= 70; failure++) {
+ batch.fetchTags();
+ long windowMs = batch.getNextAllowedFetchTimeMs() -
System.currentTimeMillis();
+
+ assertEquals(failure, batch.getConsecutiveFailures(),
+ "every attempt must be counted — a collapsed window would let
extra ones through");
+ assertTrue(windowMs > 0,
+ "backoff window must stay positive at failure " + failure + "
(was " + windowMs + "ms)");
+ assertTrue(windowMs <= maxBackoffMs,
+ "backoff window must stay capped at failure " + failure + "
(was " + windowMs + "ms)");
+
+ // Wait out the window so the next attempt is actually made.
+ Thread.sleep(maxBackoffMs + 1);
+ }
+
+ batch.close();
+ }
+
+ /**
+ * A driver that never completes its read future must not wedge the batch:
without the
+ * watchdog, fetchInProgress stays true and every later trigger is skipped
forever.
+ */
+ @Test
+ void fetchWatchdogRecoversWhenTheReadNeverCompletes() throws Exception {
+ PlcReadRequest.Builder hangingBuilder =
mock(PlcReadRequest.Builder.class);
+ PlcReadRequest hangingRequest = mock(PlcReadRequest.class);
+ when(hangingBuilder.addTagAddress(any(),
any())).thenReturn(hangingBuilder);
+ when(hangingBuilder.build()).thenReturn(hangingRequest);
+ // Never completes — models a driver whose own request timeout failed
to fire.
+ when(hangingRequest.execute()).thenAnswer(inv -> new
CompletableFuture<PlcReadResponse>());
+
+ PlcConnectionManager hangingManager = mock(PlcConnectionManager.class);
+ when(hangingManager.getConnection(any())).thenReturn(new
StubPlcConnection(hangingBuilder));
+
+ List<Throwable> errors = Collections.synchronizedList(new
ArrayList<>());
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("hanging")
+ .withConnectionManager(hangingManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(new TimerTrigger(1, TimeUnit.HOURS))
+ .withFetchTimeout(150, TimeUnit.MILLISECONDS)
+ .withListener(new TagBatch.TagBatchListener() {
+ @Override
+ public void onTagsFetched(TagBatch b, PlcReadResponse
response) {
+ }
+
+ @Override
+ public void onError(TagBatch b, Throwable error) {
+ errors.add(error);
+ }
+ })
+ .build();
+
+ batch.fetchTags();
+ verify(hangingManager, times(1)).getConnection(any());
+
+ // Wait for the watchdog to fire and release the in-progress guard.
+ long deadline = System.currentTimeMillis() + 5_000;
+ while (errors.isEmpty() && System.currentTimeMillis() < deadline) {
+ Thread.sleep(10);
+ }
+ assertFalse(errors.isEmpty(), "watchdog must report the stalled fetch
to the listener");
+
+ // The batch must accept work again rather than skipping every later
trigger.
+ // A failed fetch enters backoff, so wait that out first.
+ Thread.sleep(Math.max(0, batch.getNextAllowedFetchTimeMs() -
System.currentTimeMillis()) + 50);
+ batch.fetchTags();
+ verify(hangingManager, times(2)).getConnection(any());
+
+ batch.close();
+ }
+
+ /**
+ * getTags() previously handed out an unmodifiable view of the live tag
map, so a caller
+ * iterating it while another thread added a tag got a
ConcurrentModificationException.
+ */
+ @Test
+ void getTagsReturnsASnapshotNotALiveView() {
+ Trigger trigger = new TimerTrigger(1, TimeUnit.HOURS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("snapshot")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+
+ Map<String, String> tagSnapshot = batch.getTags();
+ Set<String> nameSnapshot = batch.getTagNames();
+
+ // Structurally modify the underlying map, then iterate what we handed
out.
+ batch.addTag("tag2", "MAIN.tag2");
+
+ assertDoesNotThrow(() -> {
+ for (Map.Entry<String, String> entry : tagSnapshot.entrySet()) {
+ assertNotNull(entry.getValue());
+ }
+ for (String name : nameSnapshot) {
+ assertNotNull(name);
+ }
+ }, "snapshots must not be invalidated by concurrent tag mutation");
+
+ assertEquals(1, tagSnapshot.size(), "snapshot must not reflect later
mutations");
+ assertEquals(1, nameSnapshot.size());
+ assertEquals(2, batch.getTagCount(), "the batch itself must see the
new tag");
+
+ batch.close();
+ }
+
@Test
void testRemoveSingleTag() {
// Arrange
diff --git
a/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/triggers/TimerTriggerTest.java
b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/triggers/TimerTriggerTest.java
index 340ce61a93..736861fbd7 100644
---
a/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/triggers/TimerTriggerTest.java
+++
b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/triggers/TimerTriggerTest.java
@@ -21,9 +21,11 @@ package org.apache.plc4x.java.tools.eventpump.triggers;
import org.junit.jupiter.api.Test;
+import java.util.Timer;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import static org.junit.jupiter.api.Assertions.*;
@@ -219,6 +221,90 @@ class TimerTriggerTest {
trigger.close();
}
+ /**
+ * Listener code must not run on the timer thread: a listener may block
(leasing a
+ * connection does), and blocking the timer thread delays this trigger's
next tick and,
+ * with a shared Timer, every other batch scheduled on it.
+ */
+ @Test
+ void listenerDoesNotRunOnTheTimerThread() throws Exception {
+ CountDownLatch fired = new CountDownLatch(1);
+ AtomicReference<String> threadName = new AtomicReference<>();
+
+ TimerTrigger trigger = new TimerTrigger(50, TimeUnit.MILLISECONDS);
+ trigger.start(t -> {
+ threadName.set(Thread.currentThread().getName());
+ fired.countDown();
+ });
+
+ assertTrue(fired.await(2, TimeUnit.SECONDS), "Trigger should fire");
+ assertTrue(threadName.get().contains("dispatch"),
+ "listener must run on the dispatcher, but ran on '" +
threadName.get() + "'");
+
+ trigger.close();
+ }
+
+ /**
+ * A shared Timer must keep serving other triggers even while one
trigger's listener is
+ * blocked.
+ */
+ @Test
+ void aBlockedListenerDoesNotStallOtherTriggersOnASharedTimer() throws
Exception {
+ Timer sharedTimer = new Timer("shared-test-timer", true);
+ CountDownLatch blockedStarted = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ CountDownLatch otherFired = new CountDownLatch(3);
+
+ TimerTrigger blocking = new TimerTrigger(50, 0, TimeUnit.MILLISECONDS,
sharedTimer);
+ TimerTrigger other = new TimerTrigger(50, 0, TimeUnit.MILLISECONDS,
sharedTimer);
+
+ blocking.start(t -> {
+ blockedStarted.countDown();
+ try {
+ release.await(5, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ other.start(t -> otherFired.countDown());
+
+ assertTrue(blockedStarted.await(2, TimeUnit.SECONDS), "blocking
listener should have started");
+ assertTrue(otherFired.await(3, TimeUnit.SECONDS),
+ "the second trigger must keep firing while the first one's
listener is blocked");
+
+ release.countDown();
+ blocking.close();
+ other.close();
+ sharedTimer.cancel();
+ }
+
+ /**
+ * A listener slower than the interval must not build up a backlog of
pending firings.
+ */
+ @Test
+ void slowListenerDoesNotAccumulateABacklog() throws Exception {
+ AtomicInteger fireCount = new AtomicInteger(0);
+ TimerTrigger trigger = new TimerTrigger(20, TimeUnit.MILLISECONDS);
+
+ trigger.start(t -> {
+ fireCount.incrementAndGet();
+ try {
+ Thread.sleep(200);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+
+ Thread.sleep(1_000);
+ trigger.close();
+
+ int fired = fireCount.get();
+ // ~50 ticks are scheduled in a second, but a 200ms listener can only
serve ~5.
+ // Excess firings are discarded rather than queued.
+ assertTrue(fired <= 10, "expected the backlog to be discarded, but the
listener ran " + fired + " times");
+ assertTrue(fired >= 1, "the listener should still have run");
+ }
+
@Test
void testGetters() {
// Arrange
