diff --git a/CHANGELOG.md b/CHANGELOG.md
index 95d2bdfe..1c2fdb84 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,11 @@
+# Unreleased
+- [Change] Batch upload tasks are handed to the network executor with `execute()` rather than `submit()`, so `shutdownNow()` returns them and the batches inside can be reported. A `Callback` or `Log` implementation that throws is now caught and logged rather than being absorbed by the discarded `Future`; it no longer reaches the thread's uncaught-exception handler either. Note that a caller supplying a `ForkJoinPool` or `ScheduledThreadPoolExecutor` through `Analytics.Builder.networkExecutor` still gets no shutdown callbacks: the former returns nothing from `shutdownNow()`, the latter wraps tasks regardless.
+- [Note] `maxRateLimitDuration` is measured against the system clock, so a large clock adjustment during a rate-limit episode can shorten or extend it.
+- [Fix] Batches abandoned at shutdown now report failure through `Callback`. Note this is a failure notification for a batch that may in fact have been delivered: a batch interrupted after its request went out is reported as failed because the client cannot know the outcome. Delivery has always been at-least-once; this makes the uncertainty visible rather than silent, so a `Callback` that counts failures will see some that were not lost. A batch interrupted while waiting out a `Retry-After` or a backoff returned without notifying anyone, and batches still queued when the executor stopped were counted in a log line and otherwise discarded. Both were unreachable while the network executor was never stopped; both became reachable with the fix below.
+- [Fix] `shutdown()` now stops the network executor rather than leaving it running. It was asked to stop and then given up to 75 seconds to finish on its own; if it had not, `shutdown()` returned anyway. Because `ExecutorService.shutdown()` does not interrupt running tasks, a thread waiting out a `Retry-After` kept running, and as these threads are not daemon threads, the JVM would not exit. `shutdown()` reported success while this happened. It now interrupts the executor once the timeout elapses, including when shutdown is itself interrupted.
+- [Change] `maxRateLimitDuration` now defaults to 30 minutes, was 12 hours. Retries prompted by a `Retry-After` header do not consume the retry count, so this duration is the only limit on how long they continue; at 12 hours a server that kept sending the header could hold a batch for that long, and because uploads run on a single thread, hold every other batch behind it. Pass a longer value to `Analytics.Builder.maxRateLimitDuration` to restore the previous limit.
+- [Fix] A `Retry-After` that will not fit in what is left of `maxRateLimitDuration` ends the episode rather than being shortened to fit. The budget was tested and then the full `Retry-After` slept on top of it, so an episode could run past its limit by up to one wait; shortening it instead would resume inside the window the server named, which it has already declined to serve, and the budget would be spent by then regardless.
+
# Version 3.5.5 (June 30, 2026)
- [New](https://github.com/segmentio/analytics-java/pull/531) Unified HTTP response handling and retry behavior
- Retryable statuses (429, 408, 410, 460, 5xx except 501/505/511) check Retry-After header first, fall back to exponential backoff
diff --git a/analytics/src/main/java/com/segment/analytics/Analytics.java b/analytics/src/main/java/com/segment/analytics/Analytics.java
index 9065627c..2903e42e 100644
--- a/analytics/src/main/java/com/segment/analytics/Analytics.java
+++ b/analytics/src/main/java/com/segment/analytics/Analytics.java
@@ -455,7 +455,14 @@ public Analytics build() {
maxTotalBackoffDurationMs = 43200 * 1000L; // 12 hours
}
if (maxRateLimitDurationMs == 0) {
- maxRateLimitDurationMs = 43200 * 1000L; // 12 hours
+ // Retry-After retries are deliberately uncounted, so this duration is the
+ // only thing bounding them. The network executor is single-threaded, so it
+ // also bounds how long one stuck batch holds up every other one.
+ //
+ // Deliberately several times the Retry-After cap. When the two are equal a
+ // single maximal Retry-After consumes the whole budget, and because the
+ // elapsed check runs before the wait the batch is dropped after one attempt.
+ maxRateLimitDurationMs = 1800 * 1000L; // 30 minutes
}
HttpLoggingInterceptor interceptor =
diff --git a/analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java b/analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java
index bc2362e5..e64f3e4b 100644
--- a/analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java
+++ b/analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java
@@ -53,8 +53,14 @@ public class AnalyticsClient {
private static final String instanceId = UUID.randomUUID().toString();
private static final int WAIT_FOR_THREAD_COMPLETE_S = 5;
private static final int TERMINATION_TIMEOUT_S = 1;
- private static final int NETWORK_TERMINATION_TIMEOUT_S =
- 75; // base Retry-After cap is 60s + headroom
+ // Deliberately shorter than a maximal Retry-After wait. Shutdown does not wait out
+ // the retry schedule; it interrupts it, and this is only the grace period before it
+ // does.
+ private static final int NETWORK_TERMINATION_TIMEOUT_S = 75;
+ // A guard against an absurd header, not a second budget: waiting less than the server
+ // asked for does not make the next attempt more likely to succeed, it just sends more
+ // requests at something already rate-limiting us. How long we keep trying is
+ // maxRateLimitDuration's job.
private static final long MAX_RATE_LIMITED_SECONDS = 300L;
static {
@@ -259,24 +265,32 @@ public void flush() {
}
}
- synchronized void setRateLimitState(long retryAfterSeconds) {
+ /** Returns the clock reading it used, so a caller can measure from the same instant. */
+ synchronized long setRateLimitState(long retryAfterSeconds) {
long now = System.currentTimeMillis();
if (rateLimitStartTime == 0) {
rateLimitStartTime = now;
}
rateLimitWaitUntil = now + (retryAfterSeconds * 1000);
rateLimited = true;
+ return now;
}
/**
- * Sets rate-limit state and atomically checks whether maxRateLimitDuration has been exceeded.
- * Returns true if the duration has been exceeded and the batch should be dropped.
+ * Sets rate-limit state and returns how much of {@code maxRateLimitDuration} is left, in
+ * milliseconds. Zero or less means the budget is spent.
+ *
+ * Returning the remaining time rather than a boolean lets one clock reading serve both the
+ * budget test and the wait that follows it. Testing and then sleeping a full Retry-After on top
+ * would otherwise overshoot the budget by up to that much.
*/
- synchronized boolean setRateLimitStateAndCheckDuration(
+ synchronized long setRateLimitStateAndRemaining(
long retryAfterSeconds, long maxRateLimitDurationMs) {
- setRateLimitState(retryAfterSeconds);
- return rateLimitStartTime > 0
- && System.currentTimeMillis() - rateLimitStartTime > maxRateLimitDurationMs;
+ long now = setRateLimitState(retryAfterSeconds);
+ if (rateLimitStartTime <= 0) {
+ return maxRateLimitDurationMs;
+ }
+ return maxRateLimitDurationMs - (now - rateLimitStartTime);
}
synchronized void clearRateLimitState() {
@@ -337,8 +351,25 @@ private void waitForLooperCompletion() {
}
}
+ /**
+ * Reports batches that were queued and never attempted.
+ *
+ *
Only reaches tasks the executor hands back as they were submitted. A {@code ForkJoinPool}
+ * returns an empty list from {@code shutdownNow()} whatever is queued, and a {@code
+ * ScheduledThreadPoolExecutor} wraps even {@code execute()}, so a caller supplying either through
+ * {@code Analytics.Builder#networkExecutor} gets no callbacks here and a task count that reads
+ * zero.
+ */
+ private void notifyDroppedBatches(List dropped) {
+ for (Runnable task : dropped) {
+ if (task instanceof BatchUploadTask) {
+ ((BatchUploadTask) task)
+ .notifyDropped(new IOException("Dropped at shutdown without being attempted"));
+ }
+ }
+ }
+
public void shutdownAndWait(ExecutorService executor, String name) {
- boolean isLooperExecutor = name != null && name.equalsIgnoreCase("looper");
boolean isNetworkExecutor = name != null && name.equalsIgnoreCase("network");
int timeoutSeconds = isNetworkExecutor ? NETWORK_TERMINATION_TIMEOUT_S : TERMINATION_TIMEOUT_S;
try {
@@ -348,49 +379,64 @@ public void shutdownAndWait(ExecutorService executor, String name) {
log.print(VERBOSE, "%s executor terminated normally.", name);
return;
}
- if (isLooperExecutor) { // Handle looper - network should finish on its own
- // not terminated within timeout -> force shutdown
- log.print(
- VERBOSE,
- "%s did not terminate in %d seconds; requesting shutdownNow().",
- name,
- TERMINATION_TIMEOUT_S);
- List dropped = executor.shutdownNow(); // interrupts running tasks
- log.print(
- VERBOSE,
- "%s shutdownNow returned %d queued tasks that never started.",
- name,
- dropped.size());
-
- // optional short wait to give interrupted tasks a chance to exit
- boolean terminatedAfterForce =
- executor.awaitTermination(TERMINATION_TIMEOUT_S, TimeUnit.SECONDS);
- log.print(
- VERBOSE,
- "%s executor %s after shutdownNow().",
- name,
- terminatedAfterForce ? "terminated" : "still running (did not terminate)");
- if (!terminatedAfterForce) {
- // final warning — investigate tasks that ignore interrupts
- log.print(
- ERROR,
- "%s executor still did not terminate; tasks may be ignoring interrupts.",
- name);
- }
+ // Both executors are force-stopped, the network one included. shutdown() does
+ // not interrupt a running task, its task can be a whole rate-limit budget deep
+ // in a sleep, and these threads are non-daemon — so leaving it to finish on its
+ // own lets shutdown() return while a thread holds the JVM open.
+ //
+ // The interrupt only reaches a thread parked in a sleep. OkHttp's reads are
+ // governed by SO_TIMEOUT, so a thread inside the HTTP call is bounded by the
+ // client's own timeouts instead, and by nothing at all if a caller supplies an
+ // OkHttpClient without them.
+ log.print(
+ VERBOSE,
+ "%s did not terminate in %d seconds; requesting shutdownNow().",
+ name,
+ timeoutSeconds);
+ List dropped = executor.shutdownNow(); // interrupts running tasks
+ log.print(
+ VERBOSE,
+ "%s shutdownNow returned %d queued tasks that never started.",
+ name,
+ dropped.size());
+
+ // Submitted and never run, so their callbacks are still owed. Counting them in
+ // a log line is not the same as telling the caller the messages did not go.
+ notifyDroppedBatches(dropped);
+
+ // optional short wait to give interrupted tasks a chance to exit
+ boolean terminatedAfterForce =
+ executor.awaitTermination(TERMINATION_TIMEOUT_S, TimeUnit.SECONDS);
+ log.print(
+ VERBOSE,
+ "%s executor %s after shutdownNow().",
+ name,
+ terminatedAfterForce ? "terminated" : "still running (did not terminate)");
+
+ if (!terminatedAfterForce) {
+ // final warning — investigate tasks that ignore interrupts
+ log.print(
+ ERROR, "%s executor still did not terminate; tasks may be ignoring interrupts.", name);
}
} catch (InterruptedException e) {
// Preserve interrupt status and attempt forceful shutdown
log.print(ERROR, e, "Interrupted while stopping %s executor.", name);
+ // Same reasoning as above: this applied to the looper only, leaving the network
+ // executor running after an interrupted shutdown.
+ List dropped = executor.shutdownNow();
+ log.print(
+ VERBOSE,
+ "%s shutdownNow invoked after interrupt; %d tasks returned.",
+ name,
+ dropped.size());
+ // These are owed a callback just as much as the ones dropped above; an
+ // interrupted shutdown is still a shutdown. Reported before the interrupt flag
+ // goes back on: callbacks run on this thread, and with the flag already set any
+ // interruptible call inside one -- a queue put, an await, a Future.get -- throws
+ // InterruptedException the moment it starts.
+ notifyDroppedBatches(dropped);
Thread.currentThread().interrupt();
- if (isLooperExecutor) {
- List dropped = executor.shutdownNow();
- log.print(
- VERBOSE,
- "%s shutdownNow invoked after interrupt; %d tasks returned.",
- name,
- dropped.size());
- }
}
}
@@ -468,7 +514,11 @@ public void run() {
batch.batch().size(),
batch.sequence());
try {
- networkExecutor.submit(
+ // execute, not submit: submit wraps the task in a FutureTask, and the
+ // work queue then holds that wrapper, so shutdownNow() hands back
+ // FutureTasks and the batches inside them cannot be identified or
+ // reported. The Future was discarded anyway.
+ networkExecutor.execute(
BatchUploadTask.create(AnalyticsClient.this, batch, maximumRetries));
} catch (RejectedExecutionException e) {
log.print(
@@ -541,6 +591,11 @@ static BatchUploadTask create(AnalyticsClient client, Batch batch, int maxRetrie
this.maxRetries = maxRetries;
}
+ /** Reports a batch that was discarded from the queue without ever being attempted. */
+ void notifyDropped(Exception exception) {
+ notifyCallbacksWithException(batch, exception);
+ }
+
private void notifyCallbacksWithException(Batch batch, Exception exception) {
for (Message message : batch.batch()) {
for (Callback callback : client.callbacks) {
@@ -682,6 +737,22 @@ private static Long parseRetryAfterSeconds(String headerValue) {
@Override
public void run() {
+ // Handed to the executor with execute() rather than submit(), so that
+ // shutdownNow() gives back the task itself and the batch inside it can be
+ // reported. That also means nothing catches what escapes here: under submit()
+ // a FutureTask absorbed it into a result nobody read, and the worker survived.
+ // Callback and Log are supplied by the caller and are invoked below outside any
+ // try, so one that throws would now kill and replace the pool's worker and reach
+ // the application's uncaught-exception handler -- from a library that could not
+ // previously raise one. Keep that property.
+ try {
+ runUploadLoop();
+ } catch (Throwable t) {
+ client.log.print(ERROR, t, "Batch %s upload task failed unexpectedly.", batch.sequence());
+ }
+ }
+
+ private void runUploadLoop() {
int totalAttempts = 0; // counts every HTTP attempt (for header and error message)
int backoffAttempts = 0; // counts attempts that consume backoff-based retries
int maxBackoffAttempts = maxRetries + 1; // preserve existing semantics
@@ -697,11 +768,11 @@ public void run() {
}
if (result.strategy == RetryStrategy.RATE_LIMITED) {
- // Atomically set rate-limit state and check whether maxRateLimitDuration is exceeded.
- boolean durationExceeded =
- client.setRateLimitStateAndCheckDuration(
+ // Atomically set rate-limit state and take what is left of maxRateLimitDuration.
+ long remainingMs =
+ client.setRateLimitStateAndRemaining(
result.retryAfterSeconds, client.maxRateLimitDurationMs);
- if (durationExceeded) {
+ if (remainingMs <= 0) {
client.clearRateLimitState();
break;
}
@@ -711,8 +782,18 @@ public void run() {
break;
}
+ long retryAfterMs = TimeUnit.SECONDS.toMillis(result.retryAfterSeconds);
+ if (retryAfterMs > remainingMs) {
+ // A wait that will not fit ends the episode. Shortening it would resume
+ // inside the window the server named -- one it has already said it will
+ // not serve -- and the budget is spent by then, so that attempt would be
+ // the last either way.
+ client.clearRateLimitState();
+ break;
+ }
+
try {
- TimeUnit.SECONDS.sleep(result.retryAfterSeconds);
+ TimeUnit.MILLISECONDS.sleep(retryAfterMs);
} catch (InterruptedException e) {
client.log.print(
DEBUG,
@@ -720,6 +801,9 @@ public void run() {
batch.sequence());
client.clearRateLimitState();
Thread.currentThread().interrupt();
+ // Every exit from this loop reports the batch. Returning without this
+ // loses it silently, with no callback at all.
+ notifyCallbacksWithException(batch, new IOException("Interrupted during shutdown", e));
return;
}
// Retry-After does not count against maxRetries.
@@ -743,6 +827,7 @@ public void run() {
client.log.print(
DEBUG, "Thread interrupted while backing off for batch %s.", batch.sequence());
Thread.currentThread().interrupt();
+ notifyCallbacksWithException(batch, new IOException("Interrupted during shutdown", e));
return;
}
}
diff --git a/analytics/src/test/java/com/segment/analytics/internal/AnalyticsClientTest.java b/analytics/src/test/java/com/segment/analytics/internal/AnalyticsClientTest.java
index 8c573173..9029393b 100644
--- a/analytics/src/test/java/com/segment/analytics/internal/AnalyticsClientTest.java
+++ b/analytics/src/test/java/com/segment/analytics/internal/AnalyticsClientTest.java
@@ -15,7 +15,6 @@
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.when;
import static org.mockito.MockitoAnnotations.openMocks;
@@ -41,12 +40,15 @@
import java.util.Map;
import java.util.Queue;
import java.util.Random;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import okhttp3.ResponseBody;
import org.junit.Before;
import org.junit.Test;
@@ -104,6 +106,26 @@ AnalyticsClient newClient() {
return newClient(DEFAULT_MAX_TOTAL_BACKOFF_DURATION_MS, DEFAULT_MAX_RATE_LIMIT_DURATION_MS);
}
+ AnalyticsClient newClient(List callbacks) {
+ return new AnalyticsClient(
+ messageQueue,
+ null,
+ segmentService,
+ 50,
+ TimeUnit.HOURS.toMillis(1),
+ 0,
+ MAX_BATCH_SIZE,
+ log,
+ threadFactory,
+ networkExecutor,
+ callbacks,
+ isShutDown,
+ writeKey,
+ new Gson(),
+ DEFAULT_MAX_TOTAL_BACKOFF_DURATION_MS,
+ DEFAULT_MAX_RATE_LIMIT_DURATION_MS);
+ }
+
AnalyticsClient newClient(long maxTotalBackoffDurationMs, long maxRateLimitDurationMs) {
return new AnalyticsClient(
messageQueue,
@@ -207,7 +229,7 @@ public void rateLimitedDeferralPreservesOverflowMessage() throws InterruptedExce
// rate-limit and submits batch with msg1. msg2 remains in queue.
assertThat(localQueue).contains(overflowMessage);
// Batch with msg1 was submitted on StopMessage (shutdown always flushes)
- verify(networkExecutor).submit(any(Runnable.class));
+ verify(networkExecutor).execute(any(Runnable.class));
}
@Test
@@ -262,7 +284,7 @@ public Message take() throws InterruptedException {
looper.run();
// First POISON deferred, second POISON submitted after rate limit cleared
- verify(networkExecutor, times(1)).submit(any(Runnable.class));
+ verify(networkExecutor, times(1)).execute(any(Runnable.class));
}
/** Wait until the queue is drained. */
@@ -272,12 +294,12 @@ static void wait(Queue queue) {
}
/**
- * Verify that a {@link BatchUploadTask} was submitted to the executor, and return the {@link
+ * Verify that a {@link BatchUploadTask} was handed to the executor, and return the {@link
* BatchUploadTask#batch} it was uploading..
*/
static Batch captureBatch(ExecutorService executor) {
final ArgumentCaptor runnableArgumentCaptor = ArgumentCaptor.forClass(Runnable.class);
- verify(executor, timeout(1000)).submit(runnableArgumentCaptor.capture());
+ verify(executor, timeout(1000)).execute(runnableArgumentCaptor.capture());
final BatchUploadTask task = (BatchUploadTask) runnableArgumentCaptor.getValue();
return task.batch;
}
@@ -389,7 +411,7 @@ public void dontFlushUntilReachesMaxSize() throws InterruptedException {
wait(messageQueue);
- verify(networkExecutor, never()).submit(any(Runnable.class));
+ verify(networkExecutor, never()).execute(any(Runnable.class));
}
/**
@@ -443,7 +465,7 @@ public void flushHowManyTimesNecessaryToStayWithinLimit() throws InterruptedExce
* message batch until the message list is empty, that was forcing the code to make one last
* batch of 1 msg in size bumping the number of times a batch would be submitted from 3 to 4
*/
- verify(networkExecutor, times(3)).submit(any(Runnable.class));
+ verify(networkExecutor, times(3)).execute(any(Runnable.class));
}
/**
@@ -471,7 +493,7 @@ public void flushWhenMultipleMessagesReachesMaxSize() throws InterruptedExceptio
wait(messageQueue);
client.shutdown();
while (!isShutDown.get()) {}
- verify(networkExecutor, times(2)).submit(any(Runnable.class));
+ verify(networkExecutor, times(2)).execute(any(Runnable.class));
}
@Test
@@ -486,7 +508,7 @@ public void enqueueBeforeMaxDoesNotTriggerFlush() {
wait(messageQueue);
// Verify that the executor didn't see anything.
- verify(networkExecutor, never()).submit(any(Runnable.class));
+ verify(networkExecutor, never()).execute(any(Runnable.class));
}
static Batch batchFor(Message message) {
@@ -907,7 +929,167 @@ public void shutdownWithNoMessageInTheQueue() throws InterruptedException {
verify(messageQueue).put(STOP);
verify(networkExecutor).shutdown();
verify(networkExecutor).awaitTermination(75, TimeUnit.SECONDS);
- verifyNoMoreInteractions(networkExecutor);
+ }
+
+ @Test
+ public void rateLimitRemainingShrinksAndGoesNonPositive() throws InterruptedException {
+ // The retry loop waits min(Retry-After, remaining), so the budget cannot be
+ // overshot by a full Retry-After the way it was when this returned a boolean and
+ // the wait was unclamped.
+ AnalyticsClient client = newClient();
+
+ long first = client.setRateLimitStateAndRemaining(1L, 200L);
+ assertThat(first).isGreaterThan(0L);
+ assertThat(first).isLessThanOrEqualTo(200L);
+
+ Thread.sleep(250);
+
+ // The episode has outlived the budget, so the caller breaks rather than waiting.
+ long second = client.setRateLimitStateAndRemaining(1L, 200L);
+ assertThat(second).isLessThanOrEqualTo(0L);
+ }
+
+ @Test
+ public void aThrowingCallbackDoesNotEscapeTheUploadTask() throws Exception {
+ // Handing the task to execute() rather than submit() removed the FutureTask that
+ // used to absorb anything thrown here. Callback is caller-supplied and is invoked
+ // outside any try inside the retry loop, so without a guard a callback that throws
+ // would kill and replace the pool's worker and reach the application's
+ // uncaught-exception handler -- from a library that previously could not raise one.
+ final AtomicReference uncaught = new AtomicReference<>();
+ Callback throwing =
+ new Callback() {
+ @Override
+ public void success(Message message) {}
+
+ @Override
+ public void failure(Message message, Throwable throwable) {
+ throw new IllegalStateException("callback blew up");
+ }
+ };
+
+ AnalyticsClient client = newClient(Collections.singletonList(throwing));
+ TrackMessage trackMessage = TrackMessage.builder("foo").userId("bar").build();
+ BatchUploadTask task =
+ new BatchUploadTask(client, BACKO, batchFor(trackMessage), DEFAULT_RETRIES);
+
+ // A non-retryable status ends the batch and reports it through Callback.failure.
+ when(segmentService.upload(isNull(), any(Batch.class)))
+ .thenReturn(Calls.response(Response.error(400, ResponseBody.create(null, "bad"))));
+
+ Thread worker = new Thread(task);
+ worker.setUncaughtExceptionHandler(
+ new Thread.UncaughtExceptionHandler() {
+ @Override
+ public void uncaughtException(Thread t, Throwable e) {
+ uncaught.set(e);
+ }
+ });
+ worker.start();
+ worker.join(5_000);
+
+ assertThat(uncaught.get())
+ .as("a throwing callback must not escape the task and reach the handler")
+ .isNull();
+ }
+
+ @Test
+ public void aRealExecutorHandsBackTheBatchTasksItQueued() throws InterruptedException {
+ // The defect this guards cannot be seen through a mock. submit() wraps a Runnable
+ // in a FutureTask and queues the wrapper, so shutdownNow() handed back FutureTasks
+ // and the batches inside them could not be identified, let alone reported. A
+ // Mockito mock does no wrapping, so it reports whatever it was given either way.
+ ThreadPoolExecutor real =
+ new ThreadPoolExecutor(
+ 1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue());
+ final CountDownLatch occupied = new CountDownLatch(1);
+ final CountDownLatch release = new CountDownLatch(1);
+ real.execute(
+ new Runnable() {
+ @Override
+ public void run() {
+ occupied.countDown();
+ try {
+ release.await();
+ } catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ });
+ occupied.await();
+
+ AnalyticsClient client = newClient();
+ TrackMessage trackMessage = TrackMessage.builder("foo").userId("bar").build();
+ // Queued behind the occupied thread, so it never starts.
+ real.execute(new BatchUploadTask(client, BACKO, batchFor(trackMessage), DEFAULT_RETRIES));
+
+ List dropped = real.shutdownNow();
+ release.countDown();
+
+ assertThat(dropped).hasSize(1);
+ assertThat(dropped.get(0))
+ .as("shutdownNow must hand back the batch task itself, not a wrapper around it")
+ .isInstanceOf(BatchUploadTask.class);
+ }
+
+ @Test
+ public void shutdownReportsQueuedBatchesItDiscards() throws InterruptedException {
+ // shutdownNow() hands back tasks that were submitted and never ran. Those batches
+ // are discarded, so their callers are owed a failure — a log line counting them is
+ // not a substitute.
+ AnalyticsClient client = newClient();
+ TrackMessage trackMessage = TrackMessage.builder("foo").userId("bar").build();
+ BatchUploadTask queued =
+ new BatchUploadTask(client, BACKO, batchFor(trackMessage), DEFAULT_RETRIES);
+
+ when(networkExecutor.awaitTermination(anyLong(), any(TimeUnit.class))).thenReturn(false);
+ when(networkExecutor.shutdownNow()).thenReturn(Collections.singletonList(queued));
+
+ client.shutdown();
+
+ verify(callback).failure(eq(trackMessage), any(IOException.class));
+ }
+
+ @Test
+ public void interruptingARetryWaitReportsTheBatch() throws InterruptedException {
+ // Every exit from the retry loop reports the batch, including this one. A batch
+ // interrupted while waiting out a Retry-After must not vanish silently.
+ AnalyticsClient client = newClient();
+ TrackMessage trackMessage = TrackMessage.builder("foo").userId("bar").build();
+ BatchUploadTask task =
+ new BatchUploadTask(client, BACKO, batchFor(trackMessage), DEFAULT_RETRIES);
+
+ // A 429 with a long Retry-After parks the task in the sleep this test interrupts.
+ when(segmentService.upload(isNull(), any(Batch.class)))
+ .thenReturn(Calls.response(errorWithRetryAfter(429, "60")));
+
+ Thread worker = new Thread(task);
+ worker.start();
+ // Give it time to reach the sleep, then interrupt as shutdown now does.
+ Thread.sleep(500);
+ worker.interrupt();
+ worker.join(5_000);
+
+ assertThat(worker.isAlive()).isFalse();
+ verify(callback, timeout(1_000)).failure(eq(trackMessage), any(IOException.class));
+ }
+
+ @Test
+ public void shutdownForcesTheNetworkExecutorThatWillNotTerminate() throws InterruptedException {
+ // The network executor must be interrupted, not merely asked to stop: its task can
+ // be a whole rate-limit budget deep in a sleep, shutdown() does not interrupt
+ // running tasks, and these threads are non-daemon — so leaving it alone lets
+ // shutdown() return while a thread holds the JVM open.
+ AnalyticsClient client = newClient();
+
+ // The mock reports it did not terminate within the timeout.
+ when(networkExecutor.awaitTermination(anyLong(), any(TimeUnit.class))).thenReturn(false);
+
+ client.shutdown();
+
+ verify(networkExecutor).shutdown();
+ verify(networkExecutor).awaitTermination(75, TimeUnit.SECONDS);
+ verify(networkExecutor).shutdownNow();
}
@Test
@@ -921,7 +1103,7 @@ public void shutdownWithMessagesInTheQueue(MessageBuilderTest builder)
verify(messageQueue).put(STOP);
verify(networkExecutor).shutdown();
verify(networkExecutor).awaitTermination(75, TimeUnit.SECONDS);
- verify(networkExecutor).submit(any(AnalyticsClient.BatchUploadTask.class));
+ verify(networkExecutor).execute(any(AnalyticsClient.BatchUploadTask.class));
}
@Test
@@ -1110,7 +1292,7 @@ public void enqueueSingleMessageAboveLimitWhenNotShutdown(MessageBuilderTest bui
// Message is above MSG/BATCH size limit so it should not be put in queue
verify(messageQueue, never()).put(message);
// And since it was never in the queue, it was never submitted in batch
- verify(networkExecutor, never()).submit(any(AnalyticsClient.BatchUploadTask.class));
+ verify(networkExecutor, never()).execute(any(AnalyticsClient.BatchUploadTask.class));
}
@Test
@@ -1135,7 +1317,7 @@ public void enqueueVerifyRegularMessagesSpecialCharactersBelowLimit(MessageBuild
client.shutdown();
while (!isShutDown.get()) {}
- verify(networkExecutor, times(1)).submit(any(AnalyticsClient.BatchUploadTask.class));
+ verify(networkExecutor, times(1)).execute(any(AnalyticsClient.BatchUploadTask.class));
}
/**
@@ -1184,7 +1366,7 @@ public void submitBatchBelowThreshold() throws InterruptedException, IllegalArgu
client.shutdown();
while (!isShutDown.get()) {}
- verify(networkExecutor, times(1)).submit(any(Runnable.class));
+ verify(networkExecutor, times(1)).execute(any(Runnable.class));
}
/**
@@ -1227,7 +1409,7 @@ public void submitBatchAboveThreshold() throws InterruptedException, IllegalArgu
client.shutdown();
while (!isShutDown.get()) {}
- verify(networkExecutor, times(8)).submit(any(Runnable.class));
+ verify(networkExecutor, times(8)).execute(any(Runnable.class));
}
@Test
@@ -1264,7 +1446,7 @@ public void submitManySmallMessagesBatchAboveThreshold() throws InterruptedExcep
client.shutdown();
while (!isShutDown.get()) {}
- verify(networkExecutor, times(21)).submit(any(Runnable.class));
+ verify(networkExecutor, times(21)).execute(any(Runnable.class));
}
@Test
@@ -1356,8 +1538,8 @@ public void rateLimitStateSetOn429() {
BatchUploadTask batchUploadTask = new BatchUploadTask(client, BACKO, batch, DEFAULT_RETRIES);
batchUploadTask.run();
- // Verify setRateLimitStateAndCheckDuration was called (state was actually set on 429)
- verify(client).setRateLimitStateAndCheckDuration(eq(1L), anyLong());
+ // Verify setRateLimitStateAndRemaining was called (state was actually set on 429)
+ verify(client).setRateLimitStateAndRemaining(eq(1L), anyLong());
assertThat(client.isRateLimited()).isFalse();
verify(segmentService, times(2)).upload(isNull(), eq(batch));
verify(callback).success(trackMessage);
@@ -1699,19 +1881,20 @@ public void retryAfterCappedAtMaxRateLimitedSeconds() {
TrackMessage trackMessage = TrackMessage.builder("foo").userId("bar").build();
Batch batch = batchFor(trackMessage);
- // Use Retry-After: 1 (small so the test doesn't sleep 300s) to verify the cap behavior
- // indirectly — the key assertion is that maxRateLimitDuration kicks in.
+ // A 1 second Retry-After against a 1ms budget: the wait cannot fit.
Response rateLimited = errorWithRetryAfter(429, "1");
- when(segmentService.upload(isNull(), eq(batch)))
- .thenReturn(Calls.response(rateLimited))
- .thenReturn(Calls.response(rateLimited));
+ when(segmentService.upload(isNull(), eq(batch))).thenReturn(Calls.response(rateLimited));
BatchUploadTask batchUploadTask =
new BatchUploadTask(shortClient, BACKO, batch, DEFAULT_RETRIES);
batchUploadTask.run();
- // 2 attempts: first one sleeps 1s, second one exceeds maxRateLimitDuration → dropped
- verify(segmentService, times(2)).upload(isNull(), eq(batch));
+ // One attempt. Shortening the wait to the 1ms that remains would resume inside the
+ // window the server named and the budget would be spent, so the batch ends here.
+ // Previously this slept the full second and made a second, doomed request — and
+ // because the budget was 1ms, whether it did so at all depended on whether the
+ // clock ticked between two reads, which made this test intermittently fail.
+ verify(segmentService, times(1)).upload(isNull(), eq(batch));
verify(callback).failure(eq(trackMessage), any(IOException.class));
}