Implement unified HTTP response handling per SDD

- Remove 408/503 from Retry-After eligibility (only 429 uses Retry-After)
- Add 429 pipeline blocking: shared rate-limit state on AnalyticsClient
  (rateLimited, rateLimitWaitUntil, rateLimitStartTime), Looper checks
  before submitting BatchUploadTasks
- Add maxTotalBackoffDuration / maxRateLimitDuration config with 12h defaults,
  exposed via Analytics.Builder
- Restructure BatchUploadTask.run() retry loop: RATE_LIMITED strategy sets
  shared state and sleeps, BACKOFF strategy tracks firstFailureTime for
  duration guard, success clears rate-limit state
- Add tests: 408/503 use BACKOFF, 429 sets/clears rate-limit state,
  maxTotalBackoffDuration drops batch, maxRateLimitDuration drops batch

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: MichaelGHSeg

analytics/src/main/java/com/segment/analytics/Analytics.java

@@ -152,6 +152,8 @@ public static class Builder {
     private int queueCapacity;
     private boolean forceTlsV1 = false;
     private GsonBuilder gsonBuilder;
+    private long maxTotalBackoffDurationMs;
+    private long maxRateLimitDurationMs;
 
     Builder(String writeKey) {
       if (writeKey == null || writeKey.trim().length() == 0) {
@@ -356,6 +358,32 @@ public Builder forceTlsVersion1() {
       return this;
     }
 
+    /**
+     * Set the maximum total duration for backoff-based retries before giving up on a batch. Default
+     * is 12 hours.
+     */
+    public Builder maxTotalBackoffDuration(long duration, TimeUnit unit) {
+      long seconds = unit.toSeconds(duration);
+      if (seconds < 1) {
+        throw new IllegalArgumentException("maxTotalBackoffDuration must be at least 1 second.");
+      }
+      this.maxTotalBackoffDurationMs = unit.toMillis(duration);
+      return this;
+    }
+
+    /**
+     * Set the maximum total duration for rate-limit (429) retries before giving up on a batch.
+     * Default is 12 hours.
+     */
+    public Builder maxRateLimitDuration(long duration, TimeUnit unit) {
+      long seconds = unit.toSeconds(duration);
+      if (seconds < 1) {
+        throw new IllegalArgumentException("maxRateLimitDuration must be at least 1 second.");
+      }
+      this.maxRateLimitDurationMs = unit.toMillis(duration);
+      return this;
+    }
+
     /** Create a {@link Analytics} client. */
     public Analytics build() {
       if (gsonBuilder == null) {
@@ -421,6 +449,12 @@ public Analytics build() {
       } else {
         callbacks = Collections.unmodifiableList(callbacks);
       }
+      if (maxTotalBackoffDurationMs == 0) {
+        maxTotalBackoffDurationMs = 43200 * 1000L; // 12 hours
+      }
+      if (maxRateLimitDurationMs == 0) {
+        maxRateLimitDurationMs = 43200 * 1000L; // 12 hours
+      }
 
       HttpLoggingInterceptor interceptor =
           new HttpLoggingInterceptor(
@@ -474,7 +508,9 @@ public void log(String message) {
               networkExecutor,
               callbacks,
               writeKey,
-              gson);
+              gson,
+              maxTotalBackoffDurationMs,
+              maxRateLimitDurationMs);
 
       return new Analytics(analyticsClient, messageTransformers, messageInterceptors, log);
     }

analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java

@@ -46,7 +46,7 @@ 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 long MAX_RETRY_AFTER_SECONDS = 300L;
+  private static final long MAX_RATE_LIMITED_SECONDS = 300L;
 
   static {
     Map<String, String> library = new LinkedHashMap<>();
@@ -72,7 +72,12 @@ public class AnalyticsClient {
   private final ScheduledExecutorService flushScheduler;
   private final AtomicBoolean isShutDown;
   private final String writeKey;
+  final long maxTotalBackoffDurationMs;
+  final long maxRateLimitDurationMs;
   private volatile Future<?> looperFuture;
+  private volatile boolean rateLimited;
+  private volatile long rateLimitWaitUntil;
+  private volatile long rateLimitStartTime;
 
   public static AnalyticsClient create(
       HttpUrl uploadUrl,
@@ -87,7 +92,9 @@ public static AnalyticsClient create(
       ExecutorService networkExecutor,
       List<Callback> callbacks,
       String writeKey,
-      Gson gsonInstance) {
+      Gson gsonInstance,
+      long maxTotalBackoffDurationMs,
+      long maxRateLimitDurationMs) {
     return new AnalyticsClient(
         new LinkedBlockingQueue<Message>(queueCapacity),
         uploadUrl,
@@ -102,7 +109,9 @@ public static AnalyticsClient create(
         callbacks,
         new AtomicBoolean(false),
         writeKey,
-        gsonInstance);
+        gsonInstance,
+        maxTotalBackoffDurationMs,
+        maxRateLimitDurationMs);
   }
 
   public AnalyticsClient(
@@ -119,7 +128,9 @@ public AnalyticsClient(
       List<Callback> callbacks,
       AtomicBoolean isShutDown,
       String writeKey,
-      Gson gsonInstance) {
+      Gson gsonInstance,
+      long maxTotalBackoffDurationMs,
+      long maxRateLimitDurationMs) {
     this.messageQueue = messageQueue;
     this.uploadUrl = uploadUrl;
     this.service = service;
@@ -133,6 +144,8 @@ public AnalyticsClient(
     this.isShutDown = isShutDown;
     this.writeKey = writeKey;
     this.gsonInstance = gsonInstance;
+    this.maxTotalBackoffDurationMs = maxTotalBackoffDurationMs;
+    this.maxRateLimitDurationMs = maxRateLimitDurationMs;
 
     this.currentQueueSizeInBytes = 0;
 
@@ -216,6 +229,30 @@ public void flush() {
     }
   }
 
+  void setRateLimitState(long retryAfterSeconds) {
+    long now = System.currentTimeMillis();
+    if (rateLimitStartTime == 0) {
+      rateLimitStartTime = now;
+    }
+    rateLimitWaitUntil = now + (retryAfterSeconds * 1000);
+    rateLimited = true;
+  }
+
+  void clearRateLimitState() {
+    rateLimited = false;
+    rateLimitWaitUntil = 0;
+    rateLimitStartTime = 0;
+  }
+
+  boolean isRateLimited() {
+    if (!rateLimited) return false;
+    if (System.currentTimeMillis() >= rateLimitWaitUntil) {
+      rateLimited = false;
+      return false;
+    }
+    return true;
+  }
+
   public void shutdown() {
     if (isShutDown.compareAndSet(false, true)) {
       final long start = System.currentTimeMillis();
@@ -365,38 +402,45 @@ public void run() {
           Boolean isOverflow = messages.size() >= size;
 
           if (!messages.isEmpty() && (isOverflow || isBlockingSignal || batchSizeLimitReached)) {
-            Batch batch = Batch.create(CONTEXT, new ArrayList<>(messages), writeKey);
-            log.print(
-                VERBOSE,
-                "Batching %s message(s) into batch %s.",
-                batch.batch().size(),
-                batch.sequence());
-            try {
-              networkExecutor.submit(
-                  BatchUploadTask.create(AnalyticsClient.this, batch, maximumRetries));
-            } catch (RejectedExecutionException e) {
+            // Skip submission if rate-limited (unless this is a StopMessage — always flush on
+            // shutdown)
+            if (isRateLimited() && message != StopMessage.STOP) {
+              log.print(DEBUG, "Rate-limited. Deferring batch submission.");
+              // Don't clear messages — they'll be picked up on the next flush trigger
+            } else {
+              Batch batch = Batch.create(CONTEXT, new ArrayList<>(messages), writeKey);
               log.print(
-                  ERROR,
-                  e,
-                  "Failed to submit batch %s to network executor during shutdown. Batch will be lost.",
+                  VERBOSE,
+                  "Batching %s message(s) into batch %s.",
+                  batch.batch().size(),
                   batch.sequence());
-              // Notify callbacks about the failure
-              for (Message msg : batch.batch()) {
-                for (Callback callback : callbacks) {
-                  callback.failure(msg, e);
+              try {
+                networkExecutor.submit(
+                    BatchUploadTask.create(AnalyticsClient.this, batch, maximumRetries));
+              } catch (RejectedExecutionException e) {
+                log.print(
+                    ERROR,
+                    e,
+                    "Failed to submit batch %s to network executor during shutdown. Batch will be lost.",
+                    batch.sequence());
+                // Notify callbacks about the failure
+                for (Message msg : batch.batch()) {
+                  for (Callback callback : callbacks) {
+                    callback.failure(msg, e);
+                  }
                 }
               }
-            }
 
-            currentBatchSize.set(0);
-            messages.clear();
-            if (batchSizeLimitReached) {
-              // If this is true that means the last message that would make us go over the limit
-              // was not added,
-              // add it to the now cleared messages list so its not lost
-              messages.add(message);
+              currentBatchSize.set(0);
+              messages.clear();
+              if (batchSizeLimitReached) {
+                // If this is true that means the last message that would make us go over the limit
+                // was not added,
+                // add it to the now cleared messages list so its not lost
+                messages.add(message);
+              }
+              batchSizeLimitReached = false;
             }
-            batchSizeLimitReached = false;
           }
         }
       } catch (InterruptedException e) {
@@ -442,7 +486,7 @@ private void notifyCallbacksWithException(Batch batch, Exception exception) {
     private enum RetryStrategy {
       NONE,
       BACKOFF,
-      RETRY_AFTER
+      RATE_LIMITED
     }
 
     private static final class UploadResult {
@@ -496,13 +540,14 @@ UploadResult upload(int attempt) {
                 batch.sequence(),
                 status,
                 retryAfterSeconds);
-            return new UploadResult(RetryStrategy.RETRY_AFTER, retryAfterSeconds);
+            return new UploadResult(RetryStrategy.RATE_LIMITED, retryAfterSeconds);
           }
           client.log.print(
               DEBUG,
-              "Status %s did not have a valid Retry-After header for batch %s.",
+              "Status %s did not have a valid Retry-After header for batch %s. Using backoff.",
               status,
               batch.sequence());
+          return new UploadResult(RetryStrategy.RATE_LIMITED, 0);
         }
 
         if (isStatusRetryWithBackoff(status)) {
@@ -536,7 +581,7 @@ UploadResult upload(int attempt) {
     }
 
     private static boolean isStatusRetryAfterEligible(int status) {
-      return status == 429 || status == 408 || status == 503;
+      return status == 429;
     }
 
     private static Long parseRetryAfterSeconds(String headerValue) {
@@ -552,8 +597,8 @@ private static Long parseRetryAfterSeconds(String headerValue) {
         if (seconds <= 0L) {
           return null;
         }
-        if (seconds > MAX_RETRY_AFTER_SECONDS) {
-          return MAX_RETRY_AFTER_SECONDS;
+        if (seconds > MAX_RATE_LIMITED_SECONDS) {
+          return MAX_RATE_LIMITED_SECONDS;
         }
         return seconds;
       } catch (NumberFormatException ignored) {
@@ -566,31 +611,66 @@ public void run() {
       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
+      long firstFailureTime = 0;
 
       while (true) {
         totalAttempts++;
         UploadResult result = upload(totalAttempts);
 
         if (result.strategy == RetryStrategy.NONE) {
+          client.clearRateLimitState();
           return;
         }
 
-        if (result.strategy == RetryStrategy.RETRY_AFTER) {
-          try {
-            TimeUnit.SECONDS.sleep(result.retryAfterSeconds);
-          } catch (InterruptedException e) {
-            client.log.print(
-                DEBUG,
-                "Thread interrupted while waiting for Retry-After for batch %s.",
-                batch.sequence());
-            Thread.currentThread().interrupt();
-            return;
+        if (result.strategy == RetryStrategy.RATE_LIMITED) {
+          // Set global rate-limit state (blocks Looper from submitting more)
+          client.setRateLimitState(result.retryAfterSeconds);
+
+          // Check maxRateLimitDuration
+          if (client.rateLimitStartTime > 0
+              && System.currentTimeMillis() - client.rateLimitStartTime
+                  > client.maxRateLimitDurationMs) {
+            client.clearRateLimitState();
+            break;
+          }
+
+          // Sleep for Retry-After then retry this batch
+          if (result.retryAfterSeconds > 0) {
+            try {
+              TimeUnit.SECONDS.sleep(result.retryAfterSeconds);
+            } catch (InterruptedException e) {
+              client.log.print(
+                  DEBUG,
+                  "Thread interrupted while waiting for Retry-After for batch %s.",
+                  batch.sequence());
+              Thread.currentThread().interrupt();
+              return;
+            }
+          } else {
+            // No valid Retry-After header — use backoff-style wait
+            backoffAttempts++;
+            if (backoffAttempts >= maxBackoffAttempts) {
+              break;
+            }
+            try {
+              backo.sleep(backoffAttempts - 1);
+            } catch (InterruptedException e) {
+              client.log.print(
+                  DEBUG, "Thread interrupted while backing off for batch %s.", batch.sequence());
+              Thread.currentThread().interrupt();
+              return;
+            }
           }
-          // Do not count Retry-After based retries against maxRetries.
+          // Retry-After with valid header does not count against maxRetries.
           continue;
         }
 
         // BACKOFF strategy
+        if (firstFailureTime == 0) firstFailureTime = System.currentTimeMillis();
+        if (System.currentTimeMillis() - firstFailureTime > client.maxTotalBackoffDurationMs) {
+          break;
+        }
+
         backoffAttempts++;
         if (backoffAttempts >= maxBackoffAttempts) {
           break;