fix: should retry on 429 (#91)

amplitude · Jun 28, 2023 · bb9ce66 · bb9ce66
1 parent 69a7ca8
commit bb9ce66
Show file tree

Hide file tree

Showing 4 changed files with 46 additions and 18 deletions.
diff --git a/src/main/java/com/amplitude/HttpTransport.java b/src/main/java/com/amplitude/HttpTransport.java
@@ -179,8 +179,6 @@ private EventsRetryResult retryEventsOnce(String userId, String deviceId, List<E
           shouldRetry = false;
           triggerEventCallbacks(events, response.code, response.error);
         }
-        // Reduce the payload to reduce risk of throttling
-        shouldReduceEventCount = true;
         break;
       case PAYLOAD_TOO_LARGE:
         shouldRetry = true;

diff --git a/src/test/java/com/amplitude/AmplitudeMultiThreadTest.java b/src/test/java/com/amplitude/AmplitudeMultiThreadTest.java
@@ -76,7 +76,7 @@ public void testMultipleThreadLogAndSendEventFailure()
     Amplitude amplitude = Amplitude.getInstance("Multiple threads log event failure.");
     amplitude.init(apiKey);
     HttpCall httpCall = getMockHttpCall(amplitude);
-    Response payloadTooLargeResponse = ResponseUtil.getRateLimitResponse(false);
+    Response payloadTooLargeResponse = ResponseUtil.getPayloadTooLargeResponse();
     when(httpCall.makeRequest(anyList())).thenAnswer(invocation -> payloadTooLargeResponse);
     int total = 1000;
     CountDownCallbacks callbacks = new CountDownCallbacks(new CountDownLatch(total));
@@ -111,7 +111,7 @@ public void testMultipleThreadLogAndSendEventWithSuccessAndFailure()
     Response successResponse = ResponseUtil.getSuccessResponse();
     Response payloadTooLargeResponse = ResponseUtil.getPayloadTooLargeResponse();
     Response invalidResponse = ResponseUtil.getInvalidResponse(false);
-    Response rateLimitResponse = ResponseUtil.getRateLimitResponse(false);
+    Response rateLimitResponse = ResponseUtil.getRateLimitResponse(true);
     List<Response> responseList =
         Arrays.asList(successResponse, payloadTooLargeResponse, invalidResponse, rateLimitResponse);
     Random rand = new Random();

diff --git a/src/test/java/com/amplitude/AmplitudeTest.java b/src/test/java/com/amplitude/AmplitudeTest.java
@@ -353,7 +353,6 @@ public void testShouldWait()
     HttpCall httpCall = getMockHttpCall(amplitude, false);
     CyclicBarrier barrier = new CyclicBarrier(2);
     CountDownLatch latch = new CountDownLatch(1);
-    CountDownLatch callbackLatch = new CountDownLatch(2);
     Response response = new Response();
     response.status = Status.RATELIMIT;
     response.code = 429;
@@ -382,22 +381,14 @@ public void testShouldWait()
     assertFalse(amplitude.shouldWait(event));
     amplitude.setRecordThrottledId(true);
     assertFalse(amplitude.shouldWait(event));
-    amplitude.setCallbacks(
-        new AmplitudeCallbacks() {
-          @Override
-          public void onLogEventServerResponse(Event event, int status, String message) {
-            callbackLatch.countDown();
-          }
-        });
     amplitude.logEvent(event);
     amplitude.logEvent(event2);
     amplitude.flushEvents();
     assertTrue(shouldWaitResultWithTimeout(amplitude, event, 1L, TimeUnit.SECONDS));
     assertFalse(amplitude.shouldWait(event2));
     barrier.await();
     assertTrue(latch.await(1L, TimeUnit.SECONDS));
-    assertTrue(callbackLatch.await(1L, TimeUnit.SECONDS));
-    assertFalse(shouldWaitResultWithTimeout(amplitude, event, 1L, TimeUnit.SECONDS));
+    assertTrue(shouldWaitResultWithTimeout(amplitude, event, 1L, TimeUnit.SECONDS));
   }
 
   @Test

diff --git a/src/test/java/com/amplitude/HttpTransportTest.java b/src/test/java/com/amplitude/HttpTransportTest.java
@@ -104,12 +104,10 @@ public void onLogEventServerResponse(Event event, int status, String message) {
     assertTrue(latch2.await(1L, TimeUnit.SECONDS));
     verify(httpCall, times(5)).makeRequest(anyList());
     for (int i = 0; i < events.size(); i++) {
-      if (i < (events.size() / 4)) {
+      if (i < (events.size() / 2)) {
         assertEquals(200, resultMap.get(events.get(i)));
-      } else if (i < (events.size() / 2)) {
-        assertEquals(413, resultMap.get(events.get(i)));
       } else {
-        assertEquals(429, resultMap.get(events.get(i)));
+        assertEquals(413, resultMap.get(events.get(i)));
       }
     }
   }
@@ -260,6 +258,47 @@ public void onLogEventServerResponse(Event event, int status, String message) {
       assertEquals(429, resultMap.get(event));
     }
   }
+
+  @Test
+  public void testRetryEventWithUserThrottled()
+      throws AmplitudeInvalidAPIKeyException, InterruptedException {
+    Response rateLimitResponse = ResponseUtil.getRateLimitResponse(false);
+    Response successResponse = ResponseUtil.getSuccessResponse();
+    HttpCall httpCall = mock(HttpCall.class);
+    CountDownLatch latch = new CountDownLatch(2);
+    CountDownLatch latch2 = new CountDownLatch(10);
+    when(httpCall.makeRequest(anyList()))
+        .thenAnswer(
+            invocation -> {
+              latch.countDown();
+              return rateLimitResponse;
+            })
+        .thenAnswer(
+            invocation -> {
+              latch.countDown();
+              return successResponse;
+            });
+
+    List<Event> events = EventsGenerator.generateEvents(10);
+    Map<Event, Integer> resultMap = new HashMap<>();
+    AmplitudeCallbacks callbacks =
+        new AmplitudeCallbacks() {
+          @Override
+          public void onLogEventServerResponse(Event event, int status, String message) {
+            resultMap.put(event, status);
+            latch2.countDown();
+          }
+        };
+    httpTransport.setHttpCall(httpCall);
+    httpTransport.setCallbacks(callbacks);
+    httpTransport.retryEvents(events, rateLimitResponse);
+    assertTrue(latch.await(1L, TimeUnit.SECONDS));
+    assertTrue(latch2.await(1L, TimeUnit.SECONDS));
+    verify(httpCall, times(2)).makeRequest(anyList());
+    for (Event event : events) {
+      assertEquals(200, resultMap.get(event));
+    }
+  }
 
   @Test
   public void testRetryEventWithTimeout()