FFM-5111 - Java SDK - Polls non stop if /stream endpoint returns 5xx error (#132)

* FFM-5111 - Java SDK - Polls non stop if /stream endpoint returns 5xx error

What
If a /stream endpoint returns a 501 HTTP error code (unimplemented) then the SSE library attempts reconnect endlessly

Why
The code is assuming 501 is a transient http error - so keeps retrying it rather than backing off for good

Testing
Added new mock web server dispatcher to mimic a 501 response when /stream is connected to.
New assertion were added to ensure SDK doesn't attempt to reconnect once when a 501 is returned.

Author: andybharness

examples/pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>io.harness.featureflags</groupId>
     <artifactId>examples</artifactId>
-    <version>1.1.9</version>
+    <version>1.1.10-SNAPSHOT</version>
 
     <properties>
         <maven.compiler.source>8</maven.compiler.source>
@@ -33,7 +33,7 @@
         <dependency>
             <groupId>io.harness</groupId>
             <artifactId>ff-java-server-sdk</artifactId>
-            <version>1.1.9</version>
+            <version>1.1.10-SNAPSHOT</version>
         </dependency>
 
         <dependency>

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>io.harness</groupId>
     <artifactId>ff-java-server-sdk</artifactId>
-    <version>1.1.9</version>
+    <version>1.1.10-SNAPSHOT</version>
     <packaging>jar</packaging>
     <name>Harness Feature Flag Java Server SDK</name>
     <description>Harness Feature Flag Java Server SDK</description>

src/main/java/io/harness/cf/client/api/InnerClient.java

@@ -235,9 +235,9 @@ public void onDisconnected() {
       pollerStartedAt = new Date();
     } else {
       log.warn(
-          "Poller was not restarted [closing={} terminated={} pollStartTime+interval={} now={} ]",
+          "Poller was not restarted [closing={} state={} pollStartTime+interval={} now={} ]",
           closing,
-          pollProcessor.state() == Service.State.TERMINATED,
+          pollProcessor.state(),
           instant,
           now);
     }
@@ -396,4 +396,8 @@ public void close() {
   BaseConfig getOptions() {
     return options;
   }
+
+  PollingProcessor getPollProcessor() {
+    return pollProcessor;
+  }
 }

src/main/java/io/harness/cf/client/connector/EventSource.java

@@ -110,13 +110,20 @@ public boolean onRetryError(
         throwable.getClass().getSimpleName(),
         throwable.getMessage());
     log.trace("onRetryError exception", throwable);
+
     updater.onError();
     if (response != null) {
-      return response.code() == 429 || response.code() >= 500;
+      return shouldRetryForHttpErrorCode(response.code());
     }
     return true;
   }
 
+  private boolean shouldRetryForHttpErrorCode(int httpCode) {
+    if (httpCode == 501) return false;
+
+    return httpCode == 429 || httpCode >= 500;
+  }
+
   @Override
   public void onClosed(ServerSentEvent serverSentEvent) {
     log.info("EventSource onClosed - disconnected");

src/test/java/io/harness/cf/client/api/CfClientTest.java

@@ -1,11 +1,15 @@
 package io.harness.cf.client.api;
 
+import static com.google.common.util.concurrent.Service.State.RUNNING;
 import static io.harness.cf.client.api.TestUtils.*;
-import static io.harness.cf.client.api.TestUtils.makeMockJsonResponse;
+import static io.harness.cf.client.api.dispatchers.CannedResponses.makeDummyJwtToken;
+import static io.harness.cf.client.api.dispatchers.CannedResponses.makeMockJsonResponse;
 import static org.junit.jupiter.api.Assertions.*;
 
 import com.google.common.collect.ImmutableSet;
 import com.google.gson.JsonObject;
+import io.harness.cf.client.api.dispatchers.TestWebServerDispatcher;
+import io.harness.cf.client.api.dispatchers.UnimplementedStreamDispatcher;
 import io.harness.cf.client.common.Cache;
 import io.harness.cf.client.connector.Connector;
 import io.harness.cf.client.connector.ConnectorException;
@@ -20,20 +24,15 @@
 import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
-import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
 import java.util.stream.Stream;
 import lombok.NonNull;
-import lombok.SneakyThrows;
-import okhttp3.mockwebserver.Dispatcher;
 import okhttp3.mockwebserver.MockResponse;
 import okhttp3.mockwebserver.MockWebServer;
-import okhttp3.mockwebserver.RecordedRequest;
-import org.jetbrains.annotations.NotNull;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -248,35 +247,6 @@ private static void testJsonVariantWithNullTarget(CfClient client) {
     assertEquals("on", value.get("value").getAsString());
   }
 
-  static class TestWebServerDispatcher extends Dispatcher {
-    private final AtomicInteger version = new AtomicInteger(2);
-
-    @Override
-    @SneakyThrows
-    @NotNull
-    public MockResponse dispatch(@NotNull RecordedRequest recordedRequest)
-        throws InterruptedException {
-      System.out.println("DISPATCH GOT ------> " + recordedRequest.getPath());
-
-      switch (Objects.requireNonNull(recordedRequest.getPath())) {
-        case "/api/1.0/client/auth":
-          return makeAuthResponse();
-        case "/api/1.0/client/env/00000000-0000-0000-0000-000000000000/feature-configs?cluster=1":
-          return makeMockJsonResponse(200, makeBasicFeatureJson());
-        case "/api/1.0/client/env/00000000-0000-0000-0000-000000000000/target-segments?cluster=1":
-          return makeMockJsonResponse(200, makeSegmentsJson());
-        case "/api/1.0/stream?cluster=1":
-          return makeMockStreamResponse(
-              200, makeFlagPatchEvent("simplebool", version.getAndIncrement()));
-        case "/api/1.0/client/env/00000000-0000-0000-0000-000000000000/feature-configs/simplebool?cluster=1":
-          return makeMockSingleBoolFlagResponse(200, "simplebool", "off", version.get());
-        default:
-          throw new UnsupportedOperationException(
-              "ERROR: url not mapped " + recordedRequest.getPath());
-      }
-    }
-  }
-
   @Test
   void streamPatchTest() throws Exception {
     BaseConfig config =
@@ -310,6 +280,41 @@ void streamPatchTest() throws Exception {
     }
   }
 
+  @Test
+  void shouldNotReconnectToStreamEndpointIfEndpointReturns501Unimplemented() throws Exception {
+    BaseConfig config =
+        BaseConfig.builder()
+            .pollIntervalInSeconds(1)
+            .analyticsEnabled(false)
+            .streamEnabled(true)
+            .build();
+
+    // This will return 501 when /stream endpoint is connected to
+    UnimplementedStreamDispatcher webserverDispatcher = new UnimplementedStreamDispatcher(1);
+
+    try (MockWebServer mockSvr = new MockWebServer()) {
+      mockSvr.setDispatcher(webserverDispatcher);
+      mockSvr.start();
+
+      try (CfClient client =
+          new CfClient(makeConnector(mockSvr.getHostName(), mockSvr.getPort()), config)) {
+
+        client.waitForInitialization();
+
+        // we should only get one connection to /stream within this window of time
+        webserverDispatcher.waitForAllConnections(15);
+        assertEquals(
+            1,
+            webserverDispatcher.getStreamEndpointCount().get(),
+            "there should only be 1 connection to /stream endpoint");
+        assertEquals(
+            RUNNING,
+            client.getInnerClient().getPollProcessor().state(),
+            "poller not in RUNNING state");
+      }
+    }
+  }
+
   private static void onEvent(String e) {
     if ("simplebool".equals(e)) {
       onEventCounter.incrementAndGet();

src/test/java/io/harness/cf/client/api/TestUtils.java

@@ -1,126 +1,35 @@
 package io.harness.cf.client.api;
 
-import com.google.gson.Gson;
 import io.harness.cf.client.connector.HarnessConfig;
 import io.harness.cf.client.connector.HarnessConnector;
-import io.harness.cf.model.*;
 import java.io.IOException;
 import java.net.URISyntaxException;
-import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
-import java.util.Arrays;
-import java.util.Base64;
-import java.util.Collections;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import okhttp3.mockwebserver.MockResponse;
 
-class TestUtils {
+public class TestUtils {
 
-  @Data
-  @AllArgsConstructor
-  static class Event {
-    String event, domain, identifier;
-    int version;
-  }
-
-  static MockResponse makeMockJsonResponse(int httpCode, String body) {
-    return new MockResponse()
-        .setResponseCode(httpCode)
-        .setBody(body)
-        .addHeader("Content-Type", "application/json; charset=UTF-8");
-  }
-
-  static String makeSegmentsJson() throws IOException, URISyntaxException {
+  public static String makeSegmentsJson() throws IOException, URISyntaxException {
     return getJsonResource("local-test-cases/segments.json");
   }
 
-  static String makeFeatureJson() throws IOException, URISyntaxException {
+  public static String makeFeatureJson() throws IOException, URISyntaxException {
     return getJsonResource("local-test-cases/percentage-rollout-with-zero-weights.json");
   }
 
-  static String makeBasicFeatureJson() throws IOException, URISyntaxException {
+  public static String makeBasicFeatureJson() throws IOException, URISyntaxException {
     return getJsonResource("local-test-cases/basic_bool_string_number_json_variations.json");
   }
 
-  static MockResponse makeAuthResponse() {
-    return makeMockJsonResponse(200, "{\"authToken\": \"" + makeDummyJwtToken() + "\"}");
-  }
-
-  static MockResponse makeMockStreamResponse(int httpCode, Event... events) {
-
-    final StringBuilder builder = new StringBuilder();
-    Arrays.stream(events)
-        .forEach(
-            e -> builder.append("event: *\ndata: ").append(new Gson().toJson(e)).append("\n\n"));
-
-    return new MockResponse()
-        .setResponseCode(httpCode)
-        .setBody(builder.toString())
-        .addHeader("Content-Type", "text/event-stream; charset=UTF-8")
-        .addHeader("Accept-Encoding", "identity");
-  }
-
-  static MockResponse makeMockSingleBoolFlagResponse(
-      int httpCode, String flagName, String state, int version) {
-    final FeatureConfig flag = makeFlag(flagName, state, version);
-    return makeMockJsonResponse(httpCode, new Gson().toJson(flag));
-  }
-
-  static HarnessConnector makeConnector(String host, int port) {
+  public static HarnessConnector makeConnector(String host, int port) {
     final String url = String.format("http://%s:%s/api/1.0", host, port);
     return new HarnessConnector(
         "dummykey", HarnessConfig.builder().readTimeout(1000).configUrl(url).eventUrl(url).build());
   }
 
-  static String makeDummyJwtToken() {
-    final String header = "{\"alg\":\"HS256\",\"typ\":\"JWT\"}";
-    final String payload =
-        "{\"environment\":\"00000000-0000-0000-0000-000000000000\","
-            + "\"environmentIdentifier\":\"Production\","
-            + "\"project\":\"00000000-0000-0000-0000-000000000000\","
-            + "\"projectIdentifier\":\"dev\","
-            + "\"accountID\":\"aaaaa_BBBBB-cccccccccc\","
-            + "\"organization\":\"00000000-0000-0000-0000-000000000000\","
-            + "\"organizationIdentifier\":\"default\","
-            + "\"clusterIdentifier\":\"1\","
-            + "\"key_type\":\"Server\"}";
-    final byte[] hmac256 = new byte[32];
-    return Base64.getEncoder().encodeToString(header.getBytes(StandardCharsets.UTF_8))
-        + "."
-        + Base64.getEncoder().encodeToString(payload.getBytes(StandardCharsets.UTF_8))
-        + "."
-        + Base64.getEncoder().encodeToString(hmac256);
-  }
-
-  static String getJsonResource(String location) throws IOException, URISyntaxException {
+  public static String getJsonResource(String location) throws IOException, URISyntaxException {
     final Path path = Paths.get(EvaluatorTest.class.getClassLoader().getResource(location).toURI());
     return new String(Files.readAllBytes(path));
   }
-
-  static Event makeFlagPatchEvent(String identifier, int version) {
-    return new Event("patch", "flag", identifier, version);
-  }
-
-  static FeatureConfig makeFlag(String flagName, String state, int version) {
-    final FeatureConfig flag = new FeatureConfig();
-    flag.setDefaultServe(new Serve().variation("true"));
-    flag.setEnvironment("DUMMY");
-    flag.setFeature(flagName);
-    flag.setKind(FeatureConfig.KindEnum.BOOLEAN);
-    flag.setOffVariation("false");
-    flag.setPrerequisites(Collections.emptyList());
-    flag.setProject("DUMMYPROJ");
-    flag.setRules(Collections.emptyList());
-    flag.setState(FeatureState.fromValue(state));
-    flag.setVariationToTargetMap(Collections.emptyList());
-    flag.setVariations(
-        Arrays.asList(
-            new Variation("true", "true", "True", "desc"),
-            new Variation("false", "false", "False", "desc")));
-    flag.setVersion((long) version);
-    return flag;
-  }
 }