fix(feature-flags): complete initial poll during activation

Author: leoromanovsky

products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/AgentlessConfigurationSource.java

@@ -56,7 +56,9 @@ final class AgentlessConfigurationSource implements ConfigurationSourceService {
   private final RatelimitedLogger ratelimitedLogger;
   private final Object lifecycleLock = new Object();
   private final AtomicBoolean polling = new AtomicBoolean();
+  private final AtomicReference<Thread> pollingThread = new AtomicReference<>();
   private volatile boolean closed;
+  private volatile boolean started;
   private volatile ScheduledFuture<?> scheduledPoll;
   private volatile String etag;
 
@@ -114,22 +116,38 @@ private AgentlessConfigurationSource(final Config config, final HttpUrl endpoint
   @Override
   public void init() {
     synchronized (lifecycleLock) {
-      if (closed || scheduledPoll != null) {
+      if (closed || started) {
         return;
       }
-      scheduledPoll =
-          executor.scheduleWithFixedDelay(
-              this::pollOnceSafely, 0, pollIntervalMillis, TimeUnit.MILLISECONDS);
+      started = true;
+    }
+
+    // Complete the first poll cycle on the activation thread. This lets OpenFeature provider
+    // initialization observe a successful retry before it checks whether configuration is ready.
+    // No request occurs before application code activates the provider.
+    pollOnceSafely();
+
+    synchronized (lifecycleLock) {
+      if (!closed) {
+        scheduledPoll =
+            executor.scheduleWithFixedDelay(
+                this::pollOnceSafely,
+                pollIntervalMillis,
+                pollIntervalMillis,
+                TimeUnit.MILLISECONDS);
+      }
     }
   }
 
   boolean pollOnce() {
     if (closed || !polling.compareAndSet(false, true)) {
       return false;
     }
+    pollingThread.set(Thread.currentThread());
     try {
       return fetchAndApply();
     } finally {
+      pollingThread.compareAndSet(Thread.currentThread(), null);
       polling.set(false);
     }
   }
@@ -142,13 +160,18 @@ public void close() {
         return;
       }
       closed = true;
+      started = false;
       poll = scheduledPoll;
       scheduledPoll = null;
     }
     if (poll != null) {
       poll.cancel(true);
     }
     client.cancel();
+    final Thread activePollingThread = pollingThread.get();
+    if (activePollingThread != null) {
+      activePollingThread.interrupt();
+    }
     executor.shutdownNow();
   }
 

products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessConfigurationSourceTest.java

@@ -847,7 +847,7 @@ void rejectsOverlappingPolls() throws Exception {
   }
 
   @Test
-  void initSchedulesPollAndCloseCancelsFuture() throws Exception {
+  void initCompletesFirstPollAndCloseCancelsScheduledFuture() throws Exception {
     final FakeClient client = new FakeClient(response(200, "etag-a", emptyConfig()));
     final AgentlessConfigurationSource service =
         new AgentlessConfigurationSource(
@@ -859,12 +859,41 @@ void initSchedulesPollAndCloseCancelsFuture() throws Exception {
     FeatureFlaggingGateway.addConfigListener(listener);
 
     service.init();
-    awaitCalls(client, 1);
+    assertEquals(1, client.calls.get());
     service.close();
 
     verify(listener).accept(any(ServerConfiguration.class));
   }
 
+  @Test
+  void initCompletesInitialRetryCycleBeforeReturning() throws Exception {
+    final List<okhttp3.Request> requests = new ArrayList<>();
+    final AgentlessConfigurationSource.OkHttpUfcHttpClient client =
+        scriptedClient(
+            requests,
+            delay -> {},
+            () -> 1.0,
+            response(500, null, null),
+            response(200, "etag-a", emptyConfig()));
+    final AgentlessConfigurationSource service =
+        new AgentlessConfigurationSource(
+            HttpUrl.get("http://localhost" + CONFIG_PATH),
+            config(),
+            60_000,
+            client,
+            Executors.newSingleThreadScheduledExecutor());
+    FeatureFlaggingGateway.addConfigListener(listener);
+
+    try {
+      service.init();
+
+      assertEquals(2, requests.size());
+      verify(listener).accept(any(ServerConfiguration.class));
+    } finally {
+      service.close();
+    }
+  }
+
   @Test
   void repeatedInitStartsOnlyOnePoller() throws Exception {
     final FakeClient client = new FakeClient(response(200, "etag-a", emptyConfig()));
@@ -979,14 +1008,20 @@ void closeInterruptsRetryBackoff() throws Exception {
     final AgentlessConfigurationSource service =
         new AgentlessConfigurationSource(
             HttpUrl.get("http://localhost" + CONFIG_PATH), config(), 30_000, client, executor);
+    final ExecutorService runner = Executors.newSingleThreadExecutor();
 
-    service.init();
-    assertTrue(backoffStarted.await(1, TimeUnit.SECONDS));
+    try {
+      final Future<?> initialization = runner.submit(service::init);
+      assertTrue(backoffStarted.await(1, TimeUnit.SECONDS));
 
-    service.close();
+      service.close();
 
-    assertTrue(executor.awaitTermination(1, TimeUnit.SECONDS));
-    assertEquals(1, requests.size());
+      initialization.get(1, TimeUnit.SECONDS);
+      assertTrue(executor.awaitTermination(1, TimeUnit.SECONDS));
+      assertEquals(1, requests.size());
+    } finally {
+      runner.shutdownNow();
+    }
   }
 
   @Test