FFM-5149 - Avoid using read timeout of 0 for SSE connections (#114)

What
SSE client should avoid using a read timeout of 0 since if the socket is not gracefully shutdown we have no way to detect that the socket is dead. Instead enforce a default read time out of 30 mins to force the connection to refresh and remove any stale socket.

Why
This will allow the client to self heal in the event of a broken connection and avoid a situtation where the client gets out of sync with the FF server.

Testing
Manual testing, checked client was able to recover its connection even after the socket was broken on purpose.

Author: andybharness

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

@@ -31,11 +31,13 @@
 import io.jsonwebtoken.Jwts;
 import io.vavr.CheckedRunnable;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import lombok.Getter;
 import lombok.Setter;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
+import okhttp3.OkHttpClient;
 import okhttp3.Request;
 import org.apache.commons.collections4.CollectionUtils;
 import org.jetbrains.annotations.NotNull;
@@ -257,7 +259,13 @@ private void initStreamingMode() {
   }
 
   void startSSE() {
-    OkSse okSse = new OkSse();
+    long readTimeoutInMins = Math.max(config.getSseReadTimeout(), 1);
+    OkHttpClient sseClient =
+        new OkHttpClient.Builder()
+            .readTimeout(readTimeoutInMins, TimeUnit.MINUTES)
+            .retryOnConnectionFailure(true)
+            .build();
+    OkSse okSse = new OkSse(sseClient);
     sse = okSse.newServerSentEvent(sseRequest, listener);
   }
 

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

@@ -54,6 +54,9 @@ public class Config {
   /** If metrics service POST call is taking > this time, we need to know about it */
   @Getter @Builder.Default long metricsServiceAcceptableDuration = 10000;
 
+  /** read timeout in minutes for SSE connections */
+  @Getter @Builder.Default long sseReadTimeout = 30;
+
   public int getFrequency() {
     return Math.max(frequency, Config.MIN_FREQUENCY);
   }

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

@@ -46,7 +46,7 @@ public Poller(
   protected void runOneIteration() {
 
     if (Thread.currentThread().isInterrupted()) {
-
+      log.warn("Polling thread interrupted, skipping iteration");
       return;
     }
     try {

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

@@ -83,7 +83,7 @@ private void processFeature(JsonObject jsonObject) {
           break;
         } else {
           log.error(
-              format("Mismatched versions, payload version [%s] featureConfig version [%s]"),
+              "Mismatched versions, payload version [{}] featureConfig version [{}]",
               version,
               featureConfig.getFeature());
         }
@@ -124,6 +124,8 @@ public boolean onRetryTime(ServerSentEvent serverSentEvent, long l) {
   @Override
   public boolean onRetryError(
       ServerSentEvent serverSentEvent, Throwable throwable, Response response) {
+    log.warn("onRetryError got {}", throwable.getMessage());
+    log.trace("onRetryError throwable is ", throwable);
     return false;
   }
 

src/test/java/io/harness/cf/client/api/mock/MockedCfConfiguration.java

@@ -29,7 +29,8 @@ public MockedCfConfiguration() {
         30000,
         10000,
         true,
-        10000);
+        10000,
+        30);
   }
 
   @Override