feat: polling in BaseEppoClient (#82)

* Add Fetch Configuration Task to the common sdk
* feat: allow polling in base eppo client
* better java. better jitter
* bump to v3.7.0

Author: typotter

build.gradle

@@ -6,7 +6,7 @@ plugins {
 }
 
 group = 'cloud.eppo'
-version = '3.6.1-SNAPSHOT'
+version = '3.7.0'
 ext.isReleaseVersion = !version.endsWith("SNAPSHOT")
 
 java {

src/main/java/cloud/eppo/BaseEppoClient.java

@@ -1,5 +1,7 @@
 package cloud.eppo;
 
+import static cloud.eppo.Constants.DEFAULT_JITTER_INTERVAL_RATIO;
+import static cloud.eppo.Constants.DEFAULT_POLLING_INTERVAL_MILLIS;
 import static cloud.eppo.Utils.throwIfEmptyOrNull;
 
 import cloud.eppo.api.*;
@@ -15,6 +17,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Timer;
 import java.util.concurrent.CompletableFuture;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
@@ -37,6 +40,7 @@ public class BaseEppoClient {
   private boolean isGracefulMode;
   private final IAssignmentCache assignmentCache;
   private final IAssignmentCache banditAssignmentCache;
+  private Timer pollTimer;
 
   @Nullable protected CompletableFuture<Boolean> getInitialConfigFuture() {
     return initialConfigFuture;
@@ -122,6 +126,55 @@ protected void loadConfiguration() {
     }
   }
 
+  protected void stopPolling() {
+    if (pollTimer != null) {
+      pollTimer.cancel();
+    }
+  }
+
+  /** Start polling using the default interval and jitter. */
+  protected void startPolling() {
+    startPolling(DEFAULT_POLLING_INTERVAL_MILLIS);
+  }
+
+  /**
+   * Start polling using the provided polling interval and default jitter of 10%
+   *
+   * @param pollingIntervalMs The base number of milliseconds to wait between configuration fetches.
+   */
+  protected void startPolling(long pollingIntervalMs) {
+    startPolling(pollingIntervalMs, pollingIntervalMs / DEFAULT_JITTER_INTERVAL_RATIO);
+  }
+
+  /**
+   * Start polling using the provided interval and jitter.
+   *
+   * @param pollingIntervalMs The base number of milliseconds to wait between configuration fetches.
+   * @param pollingJitterMs The max number of milliseconds to offset each polling interval. The SDK
+   *     selects a random number between 0 and pollingJitterMS to offset the polling interval by.
+   */
+  protected void startPolling(long pollingIntervalMs, long pollingJitterMs) {
+    stopPolling();
+    log.debug("Started polling at " + pollingIntervalMs + "," + pollingJitterMs);
+
+    // Set up polling for UFC
+    pollTimer = new Timer(true);
+    FetchConfigurationTask fetchConfigurationsTask =
+        new FetchConfigurationTask(
+            () -> {
+              log.debug("[Eppo SDK] Polling callback");
+              this.loadConfiguration();
+            },
+            pollTimer,
+            pollingIntervalMs,
+            pollingJitterMs);
+
+    // We don't want to fetch right away, so we schedule the next fetch.
+    // Graceful mode is implicit here because `FetchConfigurationsTask` catches and
+    // logs errors without rethrowing.
+    fetchConfigurationsTask.scheduleNext();
+  }
+
   protected CompletableFuture<Void> loadConfigurationAsync() {
     CompletableFuture<Void> future = new CompletableFuture<>();
 

src/main/java/cloud/eppo/Constants.java

@@ -2,33 +2,19 @@
 
 /** Constants Class */
 public class Constants {
-  /** Base URL */
+  /** API Endpoint Settings */
+  public static final String BANDIT_ENDPOINT = "/flag-config/v1/bandits";
+
+  public static final String FLAG_CONFIG_ENDPOINT = "/flag-config/v1/config";
   public static final String DEFAULT_BASE_URL = "https://fscdn.eppo.cloud/api";
 
   static String appendApiPathToHost(String host) {
     return host + "/api";
   }
 
-  public static final int REQUEST_TIMEOUT_MILLIS = 1000;
-
   /** Poller Settings */
   private static final long MILLISECOND_IN_ONE_SECOND = 1000;
 
-  public static final long TIME_INTERVAL_IN_MILLIS =
-      30 * MILLISECOND_IN_ONE_SECOND; // time interval
-  public static final long JITTER_INTERVAL_IN_MILLIS = 5 * MILLISECOND_IN_ONE_SECOND;
-
-  /** Cache Settings */
-  public static final int MAX_CACHE_ENTRIES = 1000;
-
-  /** RAC settings */
-  public static final String RAC_ENDPOINT = "/randomized_assignment/v3/config";
-
-  public static final String BANDIT_ENDPOINT = "/flag-config/v1/bandits";
-  public static final String FLAG_CONFIG_ENDPOINT = "/flag-config/v1/config";
-
-  /** Caching Settings */
-  public static final String EXPERIMENT_CONFIGURATION_CACHE_KEY = "experiment-configuration";
-
-  public static final String BANDIT_PARAMETER_CACHE_KEY = "bandit-parameter";
+  public static final long DEFAULT_POLLING_INTERVAL_MILLIS = 30 * MILLISECOND_IN_ONE_SECOND;
+  public static final long DEFAULT_JITTER_INTERVAL_RATIO = 10;
 }

src/main/java/cloud/eppo/FetchConfigurationTask.java

@@ -0,0 +1,48 @@
+package cloud.eppo;
+
+import java.util.Timer;
+import java.util.TimerTask;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class FetchConfigurationTask extends TimerTask {
+  private static final Logger log = LoggerFactory.getLogger(FetchConfigurationTask.class);
+  private final Runnable runnable;
+  private final Timer timer;
+  private final long intervalInMillis;
+  private final long jitterInMillis;
+  private final long maxJitter;
+
+  public FetchConfigurationTask(
+      Runnable runnable, Timer timer, long intervalInMillis, long jitterInMillis) {
+    assert (jitterInMillis > 0);
+
+    this.runnable = runnable;
+    this.timer = timer;
+    this.intervalInMillis = intervalInMillis;
+    this.maxJitter = intervalInMillis / 2;
+    this.jitterInMillis = jitterInMillis;
+  }
+
+  public void scheduleNext() {
+    // Limit jitter to half the interval. Also, prevents user-provided jitter from under-running the
+    // delay below 0.
+    long jitter =
+        Math.min(maxJitter, Math.round(Math.floor((Math.random() * this.jitterInMillis))));
+    long delay = this.intervalInMillis - jitter;
+    FetchConfigurationTask nextTask =
+        new FetchConfigurationTask(runnable, timer, intervalInMillis, jitterInMillis);
+    timer.schedule(nextTask, delay);
+  }
+
+  @Override
+  public void run() {
+    // TODO: retry on failed fetches
+    try {
+      runnable.run();
+    } catch (Exception e) {
+      log.error("[Eppo SDK] Error fetching experiment configuration", e);
+    }
+    scheduleNext();
+  }
+}

src/test/java/cloud/eppo/BaseEppoClientTest.java

@@ -334,7 +334,7 @@ public void testErrorGracefulModeOff() {
   @Test
   public void testInvalidConfigJSON() {
 
-    mockHttpResponse(TEST_BASE_URL, "{}");
+    mockHttpResponse("{}");
 
     initClient(false, false);
 
@@ -569,4 +569,139 @@ public void testAssignmentLogErrorNonFatal() {
     ArgumentCaptor<Assignment> assignmentLogCaptor = ArgumentCaptor.forClass(Assignment.class);
     verify(mockAssignmentLogger, times(1)).logAssignment(assignmentLogCaptor.capture());
   }
+
+  @Test
+  public void testPolling() {
+    EppoHttpClient httpClient = mockHttpResponse(BOOL_FLAG_CONFIG);
+
+    BaseEppoClient client =
+        eppoClient =
+            new BaseEppoClient(
+                DUMMY_FLAG_API_KEY,
+                "java",
+                "100.1.0",
+                null,
+                TEST_BASE_URL,
+                mockAssignmentLogger,
+                null,
+                null,
+                false,
+                false,
+                true,
+                null,
+                null,
+                null);
+
+    client.loadConfiguration();
+    client.startPolling(20);
+
+    // Method will be called immediately on init
+    verify(httpClient, times(1)).get(anyString());
+    assertTrue(eppoClient.getBooleanAssignment("bool_flag", "subject1", false));
+
+    // Sleep for 25 ms to allow another polling cycle to complete
+    sleepUninterruptedly(25);
+
+    // Now, the method should have been called twice
+    verify(httpClient, times(2)).get(anyString());
+
+    eppoClient.stopPolling();
+    assertTrue(eppoClient.getBooleanAssignment("bool_flag", "subject1", false));
+
+    sleepUninterruptedly(25);
+
+    // No more calls since stopped
+    verify(httpClient, times(2)).get(anyString());
+
+    // Set up a different config to be served
+    when(httpClient.get(anyString())).thenReturn(DISABLED_BOOL_FLAG_CONFIG.getBytes());
+    client.startPolling(20);
+
+    // True until the next config is fetched.
+    assertTrue(eppoClient.getBooleanAssignment("bool_flag", "subject1", false));
+
+    sleepUninterruptedly(25);
+
+    assertFalse(eppoClient.getBooleanAssignment("bool_flag", "subject1", false));
+
+    eppoClient.stopPolling();
+  }
+
+  @SuppressWarnings("SameParameterValue")
+  private void sleepUninterruptedly(long sleepMs) {
+    try {
+      Thread.sleep(sleepMs);
+    } catch (InterruptedException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private static final String BOOL_FLAG_CONFIG =
+      ("{\n"
+          + "  \"createdAt\": \"2024-04-17T19:40:53.716Z\",\n"
+          + "  \"format\": \"SERVER\",\n"
+          + "  \"environment\": {\n"
+          + "    \"name\": \"Test\"\n"
+          + "  },\n"
+          + "  \"flags\": {\n"
+          + "    \"bool_flag\": {\n"
+          + "      \"key\": \"bool_flag\",\n"
+          + "      \"enabled\": true,\n"
+          + "      \"variationType\": \"BOOLEAN\",\n"
+          + "      \"variations\": {\n"
+          + "        \"on\": {\n"
+          + "          \"key\": \"on\",\n"
+          + "          \"value\": true\n"
+          + "        }\n"
+          + "      },\n"
+          + "      \"allocations\": [\n"
+          + "        {\n"
+          + "          \"key\": \"on\",\n"
+          + "          \"doLog\": true,\n"
+          + "          \"splits\": [\n"
+          + "            {\n"
+          + "              \"variationKey\": \"on\",\n"
+          + "              \"shards\": []\n"
+          + "            }\n"
+          + "          ]\n"
+          + "        }\n"
+          + "      ],\n"
+          + "      \"totalShards\": 10000\n"
+          + "    }\n"
+          + "  }\n"
+          + "}");
+  private static final String DISABLED_BOOL_FLAG_CONFIG =
+      ("{\n"
+          + "  \"createdAt\": \"2024-04-17T19:40:53.716Z\",\n"
+          + "  \"format\": \"SERVER\",\n"
+          + "  \"environment\": {\n"
+          + "    \"name\": \"Test\"\n"
+          + "  },\n"
+          + "  \"flags\": {\n"
+          + "    \"bool_flag\": {\n"
+          + "      \"key\": \"bool_flag\",\n"
+          + "      \"enabled\": false,\n"
+          + "      \"variationType\": \"BOOLEAN\",\n"
+          + "      \"variations\": {\n"
+          + "        \"on\": {\n"
+          + "          \"key\": \"on\",\n"
+          + "          \"value\": true\n"
+          + "        }\n"
+          + "      },\n"
+          + "      \"allocations\": [\n"
+          + "        {\n"
+          + "          \"key\": \"on\",\n"
+          + "          \"doLog\": true,\n"
+          + "          \"splits\": [\n"
+          + "            {\n"
+          + "              \"variationKey\": \"on\",\n"
+          + "              \"shards\": []\n"
+          + "            }\n"
+          + "          ]\n"
+          + "        }\n"
+          + "      ],\n"
+          + "      \"totalShards\": 10000\n"
+          + "    }\n"
+          + "  }\n"
+          + "}");
 }

src/test/java/cloud/eppo/helpers/TestUtils.java

@@ -12,7 +12,7 @@
 public class TestUtils {
 
   @SuppressWarnings("SameParameterValue")
-  public static void mockHttpResponse(String host, String responseBody) {
+  public static EppoHttpClient mockHttpResponse(String responseBody) {
     // Create a mock instance of EppoHttpClient
     EppoHttpClient mockHttpClient = mock(EppoHttpClient.class);
 
@@ -25,6 +25,7 @@ public static void mockHttpResponse(String host, String responseBody) {
     mockAsyncResponse.complete(responseBody.getBytes());
 
     setBaseClientHttpClientOverrideField(mockHttpClient);
+    return mockHttpClient;
   }
 
   public static void mockHttpError() {