feat(communication): support mapped retried HTTP calls

Author: leoromanovsky

communication/src/main/java/datadog/communication/http/HttpRetryPolicy.java

@@ -54,7 +54,13 @@ public class HttpRetryPolicy implements AutoCloseable {
   private final double delayFactor;
   private final boolean suppressInterrupts;
 
-  private HttpRetryPolicy(
+  /**
+   * Creates a retry policy.
+   *
+   * <p>Protected so products with a stricter cross-SDK retry contract can reuse the shared HTTP
+   * retry loop while supplying their own response, exception, and backoff rules.
+   */
+  protected HttpRetryPolicy(
       int retriesLeft, long delay, double delayFactor, boolean suppressInterrupts) {
     this.retriesLeft = retriesLeft;
     this.delay = delay;

communication/src/main/java/datadog/communication/http/OkHttpUtils.java

@@ -20,6 +20,7 @@
 import java.util.List;
 import java.util.Map;
 import javax.annotation.Nullable;
+import okhttp3.Call;
 import okhttp3.ConnectionPool;
 import okhttp3.ConnectionSpec;
 import okhttp3.Credentials;
@@ -405,19 +406,39 @@ public void writeTo(BufferedSink sink) throws IOException {
   public static Response sendWithRetries(
       OkHttpClient httpClient, HttpRetryPolicy.Factory retryPolicyFactory, Request request)
       throws IOException {
+    return sendWithRetries((Call.Factory) httpClient, retryPolicyFactory, request);
+  }
+
+  public static Response sendWithRetries(
+      Call.Factory callFactory, HttpRetryPolicy.Factory retryPolicyFactory, Request request)
+      throws IOException {
+    return sendWithRetries(callFactory, retryPolicyFactory, request, response -> response);
+  }
+
+  public static <T> T sendWithRetries(
+      Call.Factory callFactory,
+      HttpRetryPolicy.Factory retryPolicyFactory,
+      Request request,
+      ResponseMapper<T> responseMapper)
+      throws IOException {
     try (HttpRetryPolicy retryPolicy = retryPolicyFactory.create()) {
       while (true) {
+        Response response = null;
         try {
-          Response response = httpClient.newCall(request).execute();
+          response = callFactory.newCall(request).execute();
           if (response.isSuccessful()) {
-            return response;
+            return responseMapper.map(response);
           }
           if (!retryPolicy.shouldRetry(response)) {
-            return response;
+            return responseMapper.map(response);
           } else {
             closeQuietly(response);
+            response = null;
           }
         } catch (Exception ex) {
+          if (response != null) {
+            closeQuietly(response);
+          }
           if (!retryPolicy.shouldRetry(ex)) {
             throw ex;
           }
@@ -428,6 +449,11 @@ public static Response sendWithRetries(
     }
   }
 
+  @FunctionalInterface
+  public interface ResponseMapper<T> {
+    T map(Response response) throws IOException;
+  }
+
   private static void closeQuietly(Response response) {
     try {
       response.close();

communication/src/test/java/datadog/communication/http/OkHttpUtilsRetryTest.java

@@ -0,0 +1,53 @@
+package datadog.communication.http;
+
+import static datadog.communication.http.OkHttpUtils.sendWithRetries;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.net.ConnectException;
+import java.util.concurrent.atomic.AtomicInteger;
+import okhttp3.Call;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.ResponseBody;
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import org.junit.jupiter.api.Test;
+
+class OkHttpUtilsRetryTest {
+
+  @Test
+  void retriesResponseMappingFailureThroughCallFactory() throws Exception {
+    final MockWebServer server = new MockWebServer();
+    final OkHttpClient client = new OkHttpClient();
+    final AtomicInteger mappingAttempts = new AtomicInteger();
+    server.enqueue(new MockResponse().setBody("first"));
+    server.enqueue(new MockResponse().setBody("second"));
+    server.start();
+
+    try {
+      final Request request = new Request.Builder().url(server.url("/configuration")).build();
+
+      final String body =
+          sendWithRetries(
+              (Call.Factory) client,
+              new HttpRetryPolicy.Factory(1, 0, 1),
+              request,
+              response -> {
+                try (ResponseBody responseBody = response.body()) {
+                  if (mappingAttempts.getAndIncrement() == 0) {
+                    throw new ConnectException("response body could not be mapped");
+                  }
+                  return responseBody.string();
+                }
+              });
+
+      assertEquals("second", body);
+      assertEquals(2, mappingAttempts.get());
+      assertEquals(2, server.getRequestCount());
+    } finally {
+      client.dispatcher().executorService().shutdownNow();
+      client.connectionPool().evictAll();
+      server.shutdown();
+    }
+  }
+}