From 7d734bd010abfecb515e7aacca2f4570dcb2f8d8 Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Wed, 23 Sep 2026 17:43:14 -0400 Subject: [PATCH 1/9] Force-stop the network executor on shutdown, bound rate limiting at 5 min MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit shutdownAndWait force-stopped only the looper. The network executor was given shutdown() and a 75s wait, and if it had not terminated by then the method returned — no shutdownNow(), no interrupt, no log, no exception. shutdown() then logged "Analytics client shut down in %s ms" and returned as though it had drained. ExecutorService.shutdown() does not interrupt running tasks, the task can be a whole rate-limit budget deep in a sleep, and the thread factory in Platform.java never calls setDaemon — so a live thread kept the JVM alive after shutdown() reported success. The retry sleeps have always handled InterruptedException correctly, clearing rate-limit state and returning; nothing was ever sending the interrupt. The same gap applied in the catch block, where an interrupted shutdown also only forced the looper. The log line there also printed TERMINATION_TIMEOUT_S regardless of which executor it waited on, so the network case reported 1 second after waiting 75. Separately, maxRateLimitDuration drops from 12 hours to 5 minutes and Retry-After from 300s to 60s. The 12 hour value assumed a retry count would stop us reaching it, but Retry-After retries are deliberately uncounted, so it was the only limit. At 300s a single sleep consumed a 5 minute budget, so the cap has to sit well under it. NETWORK_TERMINATION_TIMEOUT_S can stay at 75s: its comment claimed headroom over a 60s cap, which is now true. The test asserting verifyNoMoreInteractions(networkExecutor) encoded the old behaviour; it is replaced by one that drives awaitTermination to false and expects shutdownNow(). That test reports "Wanted but not invoked" against the previous code. 173 tests pass under devbox (JDK 11). --- CHANGELOG.md | 4 + .../java/com/segment/analytics/Analytics.java | 7 +- .../analytics/internal/AnalyticsClient.java | 85 ++++++++++--------- .../internal/AnalyticsClientTest.java | 20 ++++- 4 files changed, 74 insertions(+), 42 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 95d2bdfe..3598def1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,7 @@ +# Unreleased +- [Fix] `shutdown()` now force-stops the network executor. It was asked to stop and then left to "finish on its own", but `shutdown()` does not interrupt running tasks, its task can be a whole rate-limit budget deep in a sleep, and these threads are non-daemon — so `shutdown()` returned having logged success while a live thread kept the JVM from exiting. The retry sleeps have always handled `InterruptedException` correctly; nothing was sending the interrupt. The same gap applied when shutdown was itself interrupted. +- [New] `maxRateLimitDuration` defaults to 5 minutes rather than 12 hours, and `Retry-After` is capped at 60s rather than 300s. The 12 hour value was a backstop meant to be unreachable, but `Retry-After` retries are deliberately uncounted, so it was the only limit on that path — and with a single-threaded network executor, one stuck batch stalled every other one for the duration. + # 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..c56142de 100644 --- a/analytics/src/main/java/com/segment/analytics/Analytics.java +++ b/analytics/src/main/java/com/segment/analytics/Analytics.java @@ -455,7 +455,12 @@ public Analytics build() { maxTotalBackoffDurationMs = 43200 * 1000L; // 12 hours } if (maxRateLimitDurationMs == 0) { - maxRateLimitDurationMs = 43200 * 1000L; // 12 hours + // Five minutes, in line with the counted-backoff path's ~13 minute worst case. + // This was 12 hours, meant as a backstop a retry count would stop us reaching — + // but Retry-After retries are deliberately uncounted, so it was the only limit + // on that path, and the network executor is single-threaded, so a stuck batch + // stalled every other one for the duration. + maxRateLimitDurationMs = 300 * 1000L; // 5 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..e72f23f0 100644 --- a/analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java +++ b/analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java @@ -54,8 +54,10 @@ public class AnalyticsClient { 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 - private static final long MAX_RATE_LIMITED_SECONDS = 300L; + 75; // MAX_RATE_LIMITED_SECONDS (60s) plus headroom for the request itself + // Capped well below maxRateLimitDuration so the budget buys several attempts rather + // than one long sleep; at the old 300s a single sleep consumed a 5 minute budget. + private static final long MAX_RATE_LIMITED_SECONDS = 60L; static { Map library = new LinkedHashMap<>(); @@ -338,7 +340,6 @@ private void waitForLooperCompletion() { } 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 +349,53 @@ 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. Only the looper used to be, on the reasoning + // that the network executor would "finish on its own" — but 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 shutdown() returned, logged success + // and left a thread holding the JVM open. The sleeps have always handled + // InterruptedException correctly; nothing was sending the interrupt. + 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()); + + // 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); Thread.currentThread().interrupt(); - if (isLooperExecutor) { - List dropped = executor.shutdownNow(); - log.print( - VERBOSE, - "%s shutdownNow invoked after interrupt; %d tasks returned.", - name, - dropped.size()); - } + // 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()); } } 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..86d50455 100644 --- a/analytics/src/test/java/com/segment/analytics/internal/AnalyticsClientTest.java +++ b/analytics/src/test/java/com/segment/analytics/internal/AnalyticsClientTest.java @@ -907,7 +907,25 @@ public void shutdownWithNoMessageInTheQueue() throws InterruptedException { verify(messageQueue).put(STOP); verify(networkExecutor).shutdown(); verify(networkExecutor).awaitTermination(75, TimeUnit.SECONDS); - verifyNoMoreInteractions(networkExecutor); + } + + @Test + public void shutdownForcesTheNetworkExecutorThatWillNotTerminate() throws InterruptedException { + // This used to assert verifyNoMoreInteractions(networkExecutor) — that the + // executor was asked to stop and then left alone to "finish on its own". 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 shutdown() returned having + // logged success while a thread held 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 From a44cd82e87fd87e6d4a04d528f06b3d24aa7b268 Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Wed, 23 Sep 2026 17:53:22 -0400 Subject: [PATCH 2/9] Rewrite the release notes for a reader seeing them in isolation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Less conversational, and no references to other SDKs — someone reading this library's notes has no context for them. Unlike the other clients, the 12 hour and 300 second values here did ship, in 3.5.5, so these entries do describe a change from one released state to another and say what to pass to restore the previous limit. --- CHANGELOG.md | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3598def1..06b3bb21 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,7 @@ # Unreleased -- [Fix] `shutdown()` now force-stops the network executor. It was asked to stop and then left to "finish on its own", but `shutdown()` does not interrupt running tasks, its task can be a whole rate-limit budget deep in a sleep, and these threads are non-daemon — so `shutdown()` returned having logged success while a live thread kept the JVM from exiting. The retry sleeps have always handled `InterruptedException` correctly; nothing was sending the interrupt. The same gap applied when shutdown was itself interrupted. -- [New] `maxRateLimitDuration` defaults to 5 minutes rather than 12 hours, and `Retry-After` is capped at 60s rather than 300s. The 12 hour value was a backstop meant to be unreachable, but `Retry-After` retries are deliberately uncounted, so it was the only limit on that path — and with a single-threaded network executor, one stuck batch stalled every other one for the duration. +- [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 5 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. +- [Change] `Retry-After` values are now capped at 60 seconds, was 300. A single wait can no longer consume the whole rate-limit budget. # Version 3.5.5 (June 30, 2026) - [New](https://github.com/segmentio/analytics-java/pull/531) Unified HTTP response handling and retry behavior From 0b42b9b9f01ce882a5141505cba999e642753f93 Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Wed, 23 Sep 2026 19:34:16 -0400 Subject: [PATCH 3/9] Report batches abandoned at shutdown instead of dropping them silently MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Making shutdown actually stop the network executor turned two latent paths live, and both lose data without telling anyone. A batch interrupted while waiting out a Retry-After or a backoff returned straight out of the retry loop. Every other exit from that loop — success, retries exhausted, duration exceeded, non-retryable status — reports the batch through Callback. These two did not, and could not be reached before, because nothing interrupted that thread. Both now report an IOException. shutdownNow() hands back the tasks that were submitted and never ran. The code logged how many there were. For the looper that is nothing, since it only ever holds one task, but for the network executor those are real batches, and previously they always eventually ran. They now report failure too. Both tests fail against the code without these changes, reporting "Wanted but not invoked" for the callback. One limitation this does not address, worth stating plainly: the interrupt only helps a thread parked in a sleep. OkHttp's socket reads are governed by SO_TIMEOUT rather than interruption, so a thread inside the HTTP call itself is unaffected and shutdown still waits for the client's own timeouts — 15s each by default, but unbounded if a caller supplies an OkHttpClient without them. 173 tests pass under devbox JDK 11. --- CHANGELOG.md | 1 + .../analytics/internal/AnalyticsClient.java | 22 ++++++++++ .../internal/AnalyticsClientTest.java | 44 +++++++++++++++++++ 3 files changed, 67 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 06b3bb21..aa60dc5b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,5 @@ # Unreleased +- [Fix] Batches abandoned at shutdown now report failure through `Callback`. 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 5 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. - [Change] `Retry-After` values are now capped at 60 seconds, was 300. A single wait can no longer consume the whole rate-limit budget. 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 e72f23f0..35258c51 100644 --- a/analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java +++ b/analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java @@ -368,6 +368,17 @@ public void shutdownAndWait(ExecutorService executor, String name) { name, dropped.size()); + // These were submitted and never ran, so their callbacks are still owed. Before + // the network executor was force-stopped they always eventually ran; now they + // can be discarded here, and counting them in a log line is not the same as + // telling the caller the messages did not go. + for (Runnable task : dropped) { + if (task instanceof BatchUploadTask) { + ((BatchUploadTask) task) + .notifyDropped(new IOException("Dropped at shutdown without being attempted")); + } + } + // optional short wait to give interrupted tasks a chance to exit boolean terminatedAfterForce = executor.awaitTermination(TERMINATION_TIMEOUT_S, TimeUnit.SECONDS); @@ -546,6 +557,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) { @@ -725,6 +741,11 @@ public void run() { batch.sequence()); client.clearRateLimitState(); Thread.currentThread().interrupt(); + // Every other exit from this loop reports the batch. This one is now + // reachable — shutdown interrupts the network executor rather than + // leaving it running — so without this a batch interrupted mid-wait + // would disappear with no callback at all. + notifyCallbacksWithException(batch, new IOException("Interrupted during shutdown", e)); return; } // Retry-After does not count against maxRetries. @@ -748,6 +769,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 86d50455..fd141bfd 100644 --- a/analytics/src/test/java/com/segment/analytics/internal/AnalyticsClientTest.java +++ b/analytics/src/test/java/com/segment/analytics/internal/AnalyticsClientTest.java @@ -909,6 +909,50 @@ public void shutdownWithNoMessageInTheQueue() throws InterruptedException { verify(networkExecutor).awaitTermination(75, TimeUnit.SECONDS); } + @Test + public void shutdownReportsQueuedBatchesItDiscards() throws InterruptedException { + // shutdownNow() hands back tasks that were submitted and never ran. They used to + // be counted in a log line and otherwise forgotten, which was survivable while the + // network executor was never force-stopped. Now that it is, those batches are + // discarded on shutdown and their callers are owed a failure. + 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 other exit from the retry loop reports the batch. The interrupt paths did + // not, and were unreachable for the network executor until shutdown began + // interrupting it — so a batch waiting out a Retry-After vanished 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 { // This used to assert verifyNoMoreInteractions(networkExecutor) — that the From cec4f5d2287ff73cefeb461292685fc2dbc5582e Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Wed, 23 Sep 2026 19:37:11 -0400 Subject: [PATCH 4/9] Clamp the Retry-After wait to the remaining rate-limit budget MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every other client clamps this; java was the one I missed. It tested the budget and then slept the full Retry-After regardless, so an episode could run past its limit by up to one wait. Against twelve hours that was 0.14% and not worth noticing. Against five minutes it is 20% — a 300 second budget running to 360 — which is the overshoot this whole change exists to close. setRateLimitStateAndCheckDuration returned a boolean, which meant the caller had no way to know how much budget was left without reading the clock a second time. It now returns the remaining milliseconds, so one reading serves both the test and the wait, the same discipline applied to ruby and go. The boundary moves from "elapsed > budget" to "elapsed >= budget", matching the other clients. 176 tests pass under devbox JDK 11. --- CHANGELOG.md | 1 + .../analytics/internal/AnalyticsClient.java | 29 ++++++++++++++----- .../internal/AnalyticsClientTest.java | 22 ++++++++++++-- 3 files changed, 42 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index aa60dc5b..8e1ac584 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ - [Fix] Batches abandoned at shutdown now report failure through `Callback`. 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 5 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` wait is now clamped to what is left of `maxRateLimitDuration`. 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 — negligible against the old 12 hour budget, a fifth of the budget against 5 minutes. - [Change] `Retry-After` values are now capped at 60 seconds, was 300. A single wait can no longer consume the whole rate-limit budget. # Version 3.5.5 (June 30, 2026) 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 35258c51..7f78133d 100644 --- a/analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java +++ b/analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java @@ -274,11 +274,22 @@ synchronized void setRateLimitState(long retryAfterSeconds) { * 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. */ - synchronized boolean setRateLimitStateAndCheckDuration( + /** + * 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 overshoots the budget by up to that much — negligible against + * twelve hours, a fifth of the budget against five minutes. + */ + synchronized long setRateLimitStateAndRemaining( long retryAfterSeconds, long maxRateLimitDurationMs) { setRateLimitState(retryAfterSeconds); - return rateLimitStartTime > 0 - && System.currentTimeMillis() - rateLimitStartTime > maxRateLimitDurationMs; + if (rateLimitStartTime <= 0) { + return maxRateLimitDurationMs; + } + return maxRateLimitDurationMs - (System.currentTimeMillis() - rateLimitStartTime); } synchronized void clearRateLimitState() { @@ -718,11 +729,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; } @@ -733,7 +744,9 @@ public void run() { } try { - TimeUnit.SECONDS.sleep(result.retryAfterSeconds); + // Clamped to what is left of the budget, so the wait cannot run past it. + TimeUnit.MILLISECONDS.sleep( + Math.min(TimeUnit.SECONDS.toMillis(result.retryAfterSeconds), remainingMs)); } catch (InterruptedException e) { client.log.print( DEBUG, 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 fd141bfd..0c35b353 100644 --- a/analytics/src/test/java/com/segment/analytics/internal/AnalyticsClientTest.java +++ b/analytics/src/test/java/com/segment/analytics/internal/AnalyticsClientTest.java @@ -909,6 +909,24 @@ public void shutdownWithNoMessageInTheQueue() throws InterruptedException { verify(networkExecutor).awaitTermination(75, TimeUnit.SECONDS); } + @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 shutdownReportsQueuedBatchesItDiscards() throws InterruptedException { // shutdownNow() hands back tasks that were submitted and never ran. They used to @@ -1418,8 +1436,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); From d73ce9b66bfae7f3b3f6dababf93c8d6c0055c1a Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Wed, 23 Sep 2026 21:19:07 -0400 Subject: [PATCH 5/9] Cut the comments back to why, not history MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Applying the team convention to my own work from today. The comments explaining these changes had accumulated into potted histories: why a value had been twelve hours, what a test used to assert, which path used to be unreachable. Six months from now none of that resolves to anything — the diff and the commit messages hold it, and the comment should say why the code is the way it is. What stayed is what a maintainer would undo without it: that Kernel#sleep raises on a negative interval, that Thread#wakeup only interrupts a sleep already in progress, that OkHttp's reads are governed by SO_TIMEOUT so an interrupt does not reach them, and that inverting one assertion would make the duration budget unreachable again. Comments only, no behaviour change. --- .../java/com/segment/analytics/Analytics.java | 8 +++--- .../analytics/internal/AnalyticsClient.java | 27 +++++++++---------- .../internal/AnalyticsClientTest.java | 21 +++++++-------- 3 files changed, 25 insertions(+), 31 deletions(-) diff --git a/analytics/src/main/java/com/segment/analytics/Analytics.java b/analytics/src/main/java/com/segment/analytics/Analytics.java index c56142de..f929fe54 100644 --- a/analytics/src/main/java/com/segment/analytics/Analytics.java +++ b/analytics/src/main/java/com/segment/analytics/Analytics.java @@ -455,11 +455,9 @@ public Analytics build() { maxTotalBackoffDurationMs = 43200 * 1000L; // 12 hours } if (maxRateLimitDurationMs == 0) { - // Five minutes, in line with the counted-backoff path's ~13 minute worst case. - // This was 12 hours, meant as a backstop a retry count would stop us reaching — - // but Retry-After retries are deliberately uncounted, so it was the only limit - // on that path, and the network executor is single-threaded, so a stuck batch - // stalled every other one for the duration. + // 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. maxRateLimitDurationMs = 300 * 1000L; // 5 minutes } 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 7f78133d..d33d2e75 100644 --- a/analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java +++ b/analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java @@ -361,12 +361,15 @@ public void shutdownAndWait(ExecutorService executor, String name) { return; } - // Both executors are force-stopped. Only the looper used to be, on the reasoning - // that the network executor would "finish on its own" — but 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 shutdown() returned, logged success - // and left a thread holding the JVM open. The sleeps have always handled - // InterruptedException correctly; nothing was sending the interrupt. + // 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().", @@ -379,10 +382,8 @@ public void shutdownAndWait(ExecutorService executor, String name) { name, dropped.size()); - // These were submitted and never ran, so their callbacks are still owed. Before - // the network executor was force-stopped they always eventually ran; now they - // can be discarded here, and counting them in a log line is not the same as - // telling the caller the messages did not go. + // 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. for (Runnable task : dropped) { if (task instanceof BatchUploadTask) { ((BatchUploadTask) task) @@ -754,10 +755,8 @@ public void run() { batch.sequence()); client.clearRateLimitState(); Thread.currentThread().interrupt(); - // Every other exit from this loop reports the batch. This one is now - // reachable — shutdown interrupts the network executor rather than - // leaving it running — so without this a batch interrupted mid-wait - // would disappear with no callback at all. + // 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; } 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 0c35b353..293e9494 100644 --- a/analytics/src/test/java/com/segment/analytics/internal/AnalyticsClientTest.java +++ b/analytics/src/test/java/com/segment/analytics/internal/AnalyticsClientTest.java @@ -929,10 +929,9 @@ public void rateLimitRemainingShrinksAndGoesNonPositive() throws InterruptedExce @Test public void shutdownReportsQueuedBatchesItDiscards() throws InterruptedException { - // shutdownNow() hands back tasks that were submitted and never ran. They used to - // be counted in a log line and otherwise forgotten, which was survivable while the - // network executor was never force-stopped. Now that it is, those batches are - // discarded on shutdown and their callers are owed a failure. + // 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 = @@ -948,9 +947,8 @@ public void shutdownReportsQueuedBatchesItDiscards() throws InterruptedException @Test public void interruptingARetryWaitReportsTheBatch() throws InterruptedException { - // Every other exit from the retry loop reports the batch. The interrupt paths did - // not, and were unreachable for the network executor until shutdown began - // interrupting it — so a batch waiting out a Retry-After vanished silently. + // 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 = @@ -973,11 +971,10 @@ public void interruptingARetryWaitReportsTheBatch() throws InterruptedException @Test public void shutdownForcesTheNetworkExecutorThatWillNotTerminate() throws InterruptedException { - // This used to assert verifyNoMoreInteractions(networkExecutor) — that the - // executor was asked to stop and then left alone to "finish on its own". 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 shutdown() returned having - // logged success while a thread held the JVM open. + // 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. From 05a1d693ba07cb72f95619af10461048184f7fdc Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Thu, 24 Sep 2026 09:54:16 -0400 Subject: [PATCH 6/9] Honour Retry-After up to 300s rather than capping it at 60 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Capping at 60s meant waiting less than the server asked for, which does not make the next attempt more likely to succeed — it just sends more requests at something already rate-limiting us. Against a Retry-After of 180s inside a 5 minute budget it turns 3 requests into 6; against 300s it turns 2 into 6. The cap is a guard against an absurd header, not a second budget. How long we keep trying is max_rate_limit_duration's job, and the clamp to the remaining budget already stops a single wait running past it, so the cap now rarely binds at all. It also bought nothing for the client this was partly aimed at: with no background thread, a shorter cap turns one long wait into several short ones for the same total blocking time and more requests. Tests that pinned 60 are updated, and each SDK gains one asserting that a Retry-After inside the cap is used as given rather than shortened. --- CHANGELOG.md | 1 - .../analytics/internal/AnalyticsClient.java | 14 +++++++++----- 2 files changed, 9 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8e1ac584..b7093e98 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,7 +3,6 @@ - [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 5 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` wait is now clamped to what is left of `maxRateLimitDuration`. 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 — negligible against the old 12 hour budget, a fifth of the budget against 5 minutes. -- [Change] `Retry-After` values are now capped at 60 seconds, was 300. A single wait can no longer consume the whole rate-limit budget. # Version 3.5.5 (June 30, 2026) - [New](https://github.com/segmentio/analytics-java/pull/531) Unified HTTP response handling and retry behavior 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 d33d2e75..c49180ae 100644 --- a/analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java +++ b/analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java @@ -53,11 +53,15 @@ 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; // MAX_RATE_LIMITED_SECONDS (60s) plus headroom for the request itself - // Capped well below maxRateLimitDuration so the budget buys several attempts rather - // than one long sleep; at the old 300s a single sleep consumed a 5 minute budget. - private static final long MAX_RATE_LIMITED_SECONDS = 60L; + // 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 { Map library = new LinkedHashMap<>(); From be9e48807d185d0c4ab7670545d8a886fe350cbc Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Fri, 25 Sep 2026 14:33:20 -0400 Subject: [PATCH 7/9] Raise the rate-limit budget to 30 minutes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The budget and the Retry-After cap were both 300s, and at parity the rate-limit path degenerates. A response with no usable Retry-After waits the cap by default, the elapsed check runs before the wait, so that one wait spends the whole budget and the batch is dropped having been tried once. A legitimate Retry-After of 300 does the same. The cap also stops binding: whatever is left of the budget is always the smaller term, so the cap can never be the value that clamps. Thirty minutes restores the relationship the two knobs are meant to have — the cap bounds one wait, the budget bounds the episode — and leaves room for several attempts. It costs nothing in normal operation, since the budget only binds when the server has been rate-limiting us for a long time, and in that case keeping the data is the point. --- CHANGELOG.md | 4 ++-- .../src/main/java/com/segment/analytics/Analytics.java | 6 +++++- .../com/segment/analytics/internal/AnalyticsClient.java | 7 +------ 3 files changed, 8 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b7093e98..ef97ce2e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,8 +1,8 @@ # Unreleased - [Fix] Batches abandoned at shutdown now report failure through `Callback`. 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 5 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` wait is now clamped to what is left of `maxRateLimitDuration`. 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 — negligible against the old 12 hour budget, a fifth of the budget against 5 minutes. +- [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` wait is now clamped to what is left of `maxRateLimitDuration`. 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. # Version 3.5.5 (June 30, 2026) - [New](https://github.com/segmentio/analytics-java/pull/531) Unified HTTP response handling and retry behavior diff --git a/analytics/src/main/java/com/segment/analytics/Analytics.java b/analytics/src/main/java/com/segment/analytics/Analytics.java index f929fe54..2903e42e 100644 --- a/analytics/src/main/java/com/segment/analytics/Analytics.java +++ b/analytics/src/main/java/com/segment/analytics/Analytics.java @@ -458,7 +458,11 @@ public Analytics build() { // 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. - maxRateLimitDurationMs = 300 * 1000L; // 5 minutes + // + // 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 c49180ae..dd35f8ab 100644 --- a/analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java +++ b/analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java @@ -274,18 +274,13 @@ synchronized void setRateLimitState(long retryAfterSeconds) { rateLimited = true; } - /** - * 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 overshoots the budget by up to that much — negligible against - * twelve hours, a fifth of the budget against five minutes. + * Retry-After on top would otherwise overshoot the budget by up to that much. */ synchronized long setRateLimitStateAndRemaining( long retryAfterSeconds, long maxRateLimitDurationMs) { From f066d51c431fe38a831d3f0af1953649adbd02a6 Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Fri, 25 Sep 2026 15:51:00 -0400 Subject: [PATCH 8/9] Make the shutdown notification actually reach the queued batches The notification added for batches discarded at shutdown could never fire. Batches reached the network executor via submit(), which wraps a Runnable in a FutureTask and queues the wrapper, so shutdownNow() handed back FutureTasks and the `instanceof BatchUploadTask` test was never true. The log line counted them and nothing else happened -- exactly the behaviour the changelog says was fixed. execute() queues the task itself; its Future was discarded anyway, and it throws the same RejectedExecutionException already handled at the call site. The test could not have caught this: it stubs a mock executor to return a bare BatchUploadTask, which no real executor does. There is now a test driving a real ThreadPoolExecutor, which is what makes the wrapping visible. It documents the JDK contract rather than guarding our call site; the guard is the set of verifications that now assert execute(), which fail if the call reverts. Two related gaps closed: The interrupted-shutdown path called shutdownNow() without notifying anything, so an interrupt during the network executor's 75 second wait still discarded queued batches silently. Both paths now share one helper. setRateLimitStateAndRemaining took a second clock reading, so the budget test and the clamp did not share one -- the thing its own javadoc said they did. setRateLimitState now returns the reading it used. Immaterial in production at millisecond granularity, but a test running with a 1ms budget is flaky as a result, and ruby has the same defect where it turns CI red. Left alone and documented instead: a ForkJoinPool returns an empty list from shutdownNow() whatever is queued, and a ScheduledThreadPoolExecutor wraps even execute(), so a caller supplying either through Builder#networkExecutor still gets no callbacks. Noted on the helper. 177 tests pass, spotless clean. --- .../analytics/internal/AnalyticsClient.java | 56 ++++++++++----- .../internal/AnalyticsClientTest.java | 70 +++++++++++++++---- 2 files changed, 93 insertions(+), 33 deletions(-) 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 dd35f8ab..72a7a8f2 100644 --- a/analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java +++ b/analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java @@ -265,30 +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 returns how much of {@code maxRateLimitDuration} is left, - * in milliseconds. Zero or less means the budget is spent. + * 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. + *

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 long setRateLimitStateAndRemaining( long retryAfterSeconds, long maxRateLimitDurationMs) { - setRateLimitState(retryAfterSeconds); + long now = setRateLimitState(retryAfterSeconds); if (rateLimitStartTime <= 0) { return maxRateLimitDurationMs; } - return maxRateLimitDurationMs - (System.currentTimeMillis() - rateLimitStartTime); + return maxRateLimitDurationMs - (now - rateLimitStartTime); } synchronized void clearRateLimitState() { @@ -349,6 +351,24 @@ 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 isNetworkExecutor = name != null && name.equalsIgnoreCase("network"); int timeoutSeconds = isNetworkExecutor ? NETWORK_TERMINATION_TIMEOUT_S : TERMINATION_TIMEOUT_S; @@ -383,12 +403,7 @@ public void shutdownAndWait(ExecutorService executor, String name) { // 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. - for (Runnable task : dropped) { - if (task instanceof BatchUploadTask) { - ((BatchUploadTask) task) - .notifyDropped(new IOException("Dropped at shutdown without being attempted")); - } - } + notifyDroppedBatches(dropped); // optional short wait to give interrupted tasks a chance to exit boolean terminatedAfterForce = @@ -402,9 +417,7 @@ public void shutdownAndWait(ExecutorService executor, String name) { if (!terminatedAfterForce) { // final warning — investigate tasks that ignore interrupts log.print( - ERROR, - "%s executor still did not terminate; tasks may be ignoring interrupts.", - name); + ERROR, "%s executor still did not terminate; tasks may be ignoring interrupts.", name); } } catch (InterruptedException e) { // Preserve interrupt status and attempt forceful shutdown @@ -418,6 +431,9 @@ public void shutdownAndWait(ExecutorService executor, String name) { "%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. + notifyDroppedBatches(dropped); } } @@ -495,7 +511,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( 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 293e9494..7fed1e17 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,10 +40,12 @@ 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 okhttp3.ResponseBody; @@ -207,7 +208,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 +263,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 +273,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 +390,7 @@ public void dontFlushUntilReachesMaxSize() throws InterruptedException { wait(messageQueue); - verify(networkExecutor, never()).submit(any(Runnable.class)); + verify(networkExecutor, never()).execute(any(Runnable.class)); } /** @@ -443,7 +444,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 +472,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 +487,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) { @@ -927,6 +928,45 @@ public void rateLimitRemainingShrinksAndGoesNonPositive() throws InterruptedExce assertThat(second).isLessThanOrEqualTo(0L); } + @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 @@ -998,7 +1038,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 @@ -1187,7 +1227,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 @@ -1212,7 +1252,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)); } /** @@ -1261,7 +1301,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)); } /** @@ -1304,7 +1344,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 @@ -1341,7 +1381,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 From fbfca0369c94e8513be120b02fa638116ae4e1d1 Mon Sep 17 00:00:00 2001 From: Michael Grosse Huelsewiesche Date: Fri, 25 Sep 2026 17:14:56 -0400 Subject: [PATCH 9/9] Guard the task body, end an episode rather than resume inside the window Four review points. execute() removed the FutureTask that used to absorb anything thrown out of the upload task. Callback and Log are caller-supplied and are invoked from inside the retry loop outside any try, so 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. The loop now runs inside a try/catch(Throwable) that logs, restoring the property the old code had by accident. A test drives a throwing Callback on a real thread with a recording handler and fails without the guard. The dropped-batch notification on the interrupted-shutdown path ran after the interrupt flag was restored, so every callback it invoked started with the flag set and any interruptible call inside one -- a queue put, an await, a Future.get -- threw immediately. It now reports first and restores the flag afterwards. A Retry-After that will not fit the remaining budget ends the episode instead of being shortened, for the reason raised in review: resuming inside the window the server named sends a request it has already declined to serve, and the budget is spent by then so it would be the final attempt regardless. That also settles a test that was measurably flaky. retryAfterCappedAtMaxRateLimitedSeconds runs with a 1ms budget, where whether the second attempt happened at all depended on whether the millisecond ticked between two clock reads. It is now one attempt, deterministically. CHANGELOG additions for two things it did not mention: the at-least-once consequence of reporting a batch that may in fact have been delivered, and that the budget is measured against the system clock. 178 tests pass, spotless clean. --- CHANGELOG.md | 6 +- .../analytics/internal/AnalyticsClient.java | 37 +++++++-- .../internal/AnalyticsClientTest.java | 80 +++++++++++++++++-- 3 files changed, 109 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ef97ce2e..1c2fdb84 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,8 +1,10 @@ # Unreleased -- [Fix] Batches abandoned at shutdown now report failure through `Callback`. 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. +- [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` wait is now clamped to what is left of `maxRateLimitDuration`. 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. +- [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 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 72a7a8f2..e64f3e4b 100644 --- a/analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java +++ b/analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java @@ -422,7 +422,6 @@ public void shutdownAndWait(ExecutorService executor, String name) { } catch (InterruptedException e) { // Preserve interrupt status and attempt forceful shutdown log.print(ERROR, e, "Interrupted while stopping %s executor.", name); - Thread.currentThread().interrupt(); // Same reasoning as above: this applied to the looper only, leaving the network // executor running after an interrupted shutdown. List dropped = executor.shutdownNow(); @@ -432,8 +431,12 @@ public void shutdownAndWait(ExecutorService executor, String name) { name, dropped.size()); // These are owed a callback just as much as the ones dropped above; an - // interrupted shutdown is still a shutdown. + // 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(); } } @@ -734,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 @@ -763,10 +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 { - // Clamped to what is left of the budget, so the wait cannot run past it. - TimeUnit.MILLISECONDS.sleep( - Math.min(TimeUnit.SECONDS.toMillis(result.retryAfterSeconds), remainingMs)); + TimeUnit.MILLISECONDS.sleep(retryAfterMs); } catch (InterruptedException e) { client.log.print( DEBUG, 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 7fed1e17..9029393b 100644 --- a/analytics/src/test/java/com/segment/analytics/internal/AnalyticsClientTest.java +++ b/analytics/src/test/java/com/segment/analytics/internal/AnalyticsClientTest.java @@ -48,6 +48,7 @@ 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; @@ -105,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, @@ -928,6 +949,50 @@ public void rateLimitRemainingShrinksAndGoesNonPositive() throws InterruptedExce 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 @@ -1816,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)); }