feat(openfeature): add direct flagevaluation fallback

communication/src/main/java/datadog/communication/BackendApiFactory.java

@@ -64,13 +64,19 @@ public BackendApi createDirectIntakeApi(Intake intake, boolean responseCompressi
         responseCompression);
   }
 
-  /** Creates an API client that sends data through a compatible local EVP proxy. */
+  /** Creates an API client that uses the specified retry policy with a compatible local proxy. */
   public @Nullable BackendApi createEvpProxyApi(Intake intake) {
     return createEvpProxyApi(intake, true);
   }
 
   /** Creates an API client that sends data through a compatible local EVP proxy. */
   public @Nullable BackendApi createEvpProxyApi(Intake intake, boolean responseCompression) {
+    return createEvpProxyApi(intake, responseCompression, retryPolicyFactory());
+  }
+
+  /** Creates an API client that sends data through a compatible local EVP proxy. */
+  public @Nullable BackendApi createEvpProxyApi(
+      Intake intake, boolean responseCompression, HttpRetryPolicy.Factory retryPolicyFactory) {
     DDAgentFeaturesDiscovery featuresDiscovery =
         sharedCommunicationObjects.featuresDiscovery(config);
     featuresDiscovery.discoverIfOutdated();
@@ -91,7 +97,7 @@ public BackendApi createDirectIntakeApi(Intake intake, boolean responseCompressi
         traceId,
         evpProxyUrl,
         subdomain,
-        retryPolicyFactory(),
+        retryPolicyFactory,
         sharedCommunicationObjects.agentHttpClient,
         responseCompression);
   }

communication/src/main/java/datadog/communication/IntakeApi.java

@@ -25,6 +25,7 @@ public class IntakeApi implements BackendApi {
   private static final String ACCEPT_ENCODING_HEADER = "Accept-Encoding";
   private static final String CONTENT_ENCODING_HEADER = "Content-Encoding";
   private static final String GZIP_ENCODING = "gzip";
+  private static final String IDENTITY_ENCODING = "identity";
 
   private final String apiKey;
   private final String traceId;
@@ -73,9 +74,10 @@ public <T> T post(
       requestBuilder.addHeader(CONTENT_ENCODING_HEADER, GZIP_ENCODING);
     }
 
-    if (responseCompression) {
-      requestBuilder.addHeader(ACCEPT_ENCODING_HEADER, GZIP_ENCODING);
-    }
+    // OkHttp adds Accept-Encoding: gzip when this header is absent. Always set the header so a
+    // caller can disable response compression on the wire.
+    requestBuilder.addHeader(
+        ACCEPT_ENCODING_HEADER, responseCompression ? GZIP_ENCODING : IDENTITY_ENCODING);
 
     Request request = requestBuilder.build();
     try (okhttp3.Response response =

communication/src/test/java/datadog/communication/BackendApiFactoryTest.java

@@ -4,9 +4,11 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 import datadog.communication.ddagent.DDAgentFeaturesDiscovery;
 import datadog.communication.ddagent.SharedCommunicationObjects;
+import datadog.communication.http.HttpRetryPolicy;
 import datadog.metrics.api.Monitoring;
 import datadog.trace.api.Config;
 import datadog.trace.api.ProtocolVersion;
@@ -61,6 +63,38 @@ void advertisedEvpProxyEndpointSupportsDisabledResponseCompression() throws Exce
     }
   }
 
+  @Test
+  void explicitNoRetryProxyPolicyDoesNotReplayAmbiguousFailure() throws Exception {
+    final MockWebServer agent = new MockWebServer();
+    agent.enqueue(new MockResponse().setResponseCode(500).setBody("ambiguous"));
+    agent.enqueue(new MockResponse().setResponseCode(200).setBody("{}"));
+    agent.start();
+    try {
+      final FakeFeaturesDiscovery discovery = new FakeFeaturesDiscovery(V4_EVP_PROXY_ENDPOINT);
+      final BackendApiFactory factory =
+          new BackendApiFactory(
+              Config.get(), sharedCommunicationObjects(discovery, agent.url("/")));
+      final BackendApi api =
+          factory.createEvpProxyApi(
+              Intake.EVENT_PLATFORM, false, HttpRetryPolicy.Factory.NEVER_RETRY);
+
+      assertNotNull(api);
+      assertThrows(
+          HttpResponseException.class,
+          () ->
+              api.post(
+                  "flagevaluation",
+                  RequestBody.create(JSON, "{}".getBytes(StandardCharsets.UTF_8)),
+                  stream -> null,
+                  null,
+                  false));
+
+      assertEquals(1, agent.getRequestCount());
+    } finally {
+      agent.shutdown();
+    }
+  }
+
   private static SharedCommunicationObjects sharedCommunicationObjects(
       final DDAgentFeaturesDiscovery discovery, final HttpUrl agentUrl) {
     final TestSharedCommunicationObjects sco = new TestSharedCommunicationObjects(discovery);

communication/src/test/java/datadog/communication/IntakeApiTest.java

@@ -0,0 +1,65 @@
+package datadog.communication;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import datadog.communication.http.HttpRetryPolicy;
+import java.io.IOException;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.RequestBody;
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import okhttp3.mockwebserver.RecordedRequest;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class IntakeApiTest {
+
+  private static final MediaType JSON = MediaType.parse("application/json");
+
+  private MockWebServer server;
+  private OkHttpClient client;
+
+  @BeforeEach
+  void setUp() throws IOException {
+    server = new MockWebServer();
+    server.start();
+    client = new OkHttpClient.Builder().build();
+  }
+
+  @AfterEach
+  void tearDown() throws IOException {
+    client.dispatcher().executorService().shutdownNow();
+    client.connectionPool().evictAll();
+    server.shutdown();
+  }
+
+  @Test
+  void requestsGzipResponseCompressionWhenEnabled() throws Exception {
+    assertEquals("gzip", postAndReadAcceptEncoding(true));
+  }
+
+  @Test
+  void requestsIdentityResponseEncodingWhenCompressionIsDisabled() throws Exception {
+    assertEquals("identity", postAndReadAcceptEncoding(false));
+  }
+
+  private String postAndReadAcceptEncoding(final boolean responseCompression) throws Exception {
+    server.enqueue(new MockResponse().setResponseCode(200).setBody("{}"));
+    final IntakeApi api =
+        new IntakeApi(
+            server.url("/api/v2/"),
+            "api-key",
+            "123",
+            HttpRetryPolicy.Factory.NEVER_RETRY,
+            client,
+            responseCompression);
+
+    api.post("flagevaluation", RequestBody.create(JSON, "{}"), responseBody -> null, null, false);
+
+    final RecordedRequest request = server.takeRequest();
+    assertEquals("/api/v2/flagevaluation", request.getPath());
+    return request.getHeader("Accept-Encoding");
+  }
+}

products/feature-flagging/feature-flagging-lib/src/jmh/java/com/datadog/featureflag/FlagEvaluationEnqueueContentionBenchmark.java

@@ -7,6 +7,7 @@
 import datadog.trace.api.Config;
 import datadog.trace.api.featureflag.FeatureFlaggingGateway;
 import datadog.trace.api.featureflag.flagevaluation.FlagEvalEvent;
+import datadog.trace.api.intake.Intake;
 import de.thetaphi.forbiddenapis.SuppressForbidden;
 import java.util.HashMap;
 import java.util.Map;
@@ -90,7 +91,13 @@ public void setUp() {
     final Config config = Config.get();
     final BackendApiFactory factory = new BackendApiFactory(config, null);
     // Capacity well above what the batch-draining consumer should ever let build up.
-    writer = new FlagEvaluationWriterImpl(1 << 20, Long.MAX_VALUE, NANOSECONDS, factory, config);
+    writer =
+        new FlagEvaluationWriterImpl(
+            1 << 20,
+            Long.MAX_VALUE,
+            NANOSECONDS,
+            () -> factory.createBackendApi(Intake.EVENT_PLATFORM, false),
+            config);
   }
 
   /**

products/feature-flagging/feature-flagging-lib/src/jmh/java/com/datadog/featureflag/FlagEvaluationHotPathBenchmark.java

@@ -6,6 +6,7 @@
 import datadog.communication.BackendApiFactory;
 import datadog.trace.api.Config;
 import datadog.trace.api.featureflag.flagevaluation.FlagEvalEvent;
+import datadog.trace.api.intake.Intake;
 import java.util.HashMap;
 import java.util.Map;
 import org.openjdk.jmh.annotations.Benchmark;
@@ -79,10 +80,18 @@ public void setUp() {
     final BackendApiFactory factory = new BackendApiFactory(config, null);
     final Map<String, String> ddContext = new HashMap<>();
     ddContext.put("service", "bench-service");
-    handler = FlagEvaluationWriterImpl.createHandlerForTest(factory, ddContext);
+    handler =
+        FlagEvaluationWriterImpl.createHandlerForTest(
+            () -> factory.createBackendApi(Intake.EVENT_PLATFORM, false), ddContext);
 
     // Capacity large enough that the benchmark never overflows within a measurement window.
-    writer = new FlagEvaluationWriterImpl(1 << 20, Long.MAX_VALUE, NANOSECONDS, factory, config);
+    writer =
+        new FlagEvaluationWriterImpl(
+            1 << 20,
+            Long.MAX_VALUE,
+            NANOSECONDS,
+            () -> factory.createBackendApi(Intake.EVENT_PLATFORM, false),
+            config);
   }
 
   /**

products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/AgentlessExposureBackendApi.java → products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/AgentlessFeatureFlagBackendApi.java

@@ -13,21 +13,26 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-/** Sends exposures through a local EVP proxy, with a safe direct intake fallback. */
-final class AgentlessExposureBackendApi implements BackendApi {
+/** Sends Feature Flag events through a local EVP proxy, with a safe direct intake fallback. */
+final class AgentlessFeatureFlagBackendApi implements BackendApi {
 
-  private static final Logger LOGGER = LoggerFactory.getLogger(AgentlessExposureBackendApi.class);
+  private static final Logger LOGGER =
+      LoggerFactory.getLogger(AgentlessFeatureFlagBackendApi.class);
 
-  private final BackendApi localApi;
+  private final BackendApi proxyApi;
   private final Supplier<BackendApi> directApiSupplier;
+  private final String eventType;
   private volatile BackendApi activeApi;
   private volatile boolean directApiCreationAttempted;
 
-  AgentlessExposureBackendApi(
-      final BackendApi localApi, final Supplier<BackendApi> directApiSupplier) {
-    this.localApi = localApi;
+  AgentlessFeatureFlagBackendApi(
+      final BackendApi proxyApi,
+      final Supplier<BackendApi> directApiSupplier,
+      final String eventType) {
+    this.proxyApi = proxyApi;
     this.directApiSupplier = directApiSupplier;
-    this.activeApi = localApi;
+    this.eventType = eventType;
+    this.activeApi = proxyApi;
   }
 
   @Override
@@ -43,7 +48,7 @@ public <T> T post(
       return selectedApi.post(
           uri, requestBody, responseParser, requestListener, requestCompression);
     } catch (final IOException exception) {
-      if (selectedApi != localApi || !isDefinitiveRejection(exception)) {
+      if (selectedApi != proxyApi || !isDefinitiveRejection(exception)) {
         throw exception;
       }
 
@@ -58,13 +63,13 @@ public <T> T post(
   @Nullable
   private BackendApi getOrCreateDirectApi() {
     final BackendApi selectedApi = activeApi;
-    if (selectedApi != localApi) {
+    if (selectedApi != proxyApi) {
       return selectedApi;
     }
 
     synchronized (this) {
       final BackendApi currentApi = activeApi;
-      if (currentApi != localApi) {
+      if (currentApi != proxyApi) {
         return currentApi;
       }
       if (directApiCreationAttempted) {
@@ -74,7 +79,8 @@ private BackendApi getOrCreateDirectApi() {
       final BackendApi directApi = directApiSupplier.get();
       if (directApi != null) {
         LOGGER.debug(
-            "Switching Feature Flagging exposure delivery from the local EVP proxy to direct intake");
+            "Switching Feature Flagging {} delivery from the local EVP proxy to direct intake",
+            eventType);
         activeApi = directApi;
       }
       directApiCreationAttempted = true;