FFM-6442 - Java SDK - java.lang.IllegalStateException logged by okhttp (#134)

* FFM-6442 - Java SDK - java.lang.IllegalStateException logged by okhttp

What
- Fixed RetryInterceptor to close response & remove old fix
- Fixed 403 re-authentication logic to correctly restart auth service
- Added new tests for returning and asserting 4xx codes
- ConnectionException to now stores HTTP code and reason for better logging + removed some excessive logging

Why
java.lang.IllegalStateException "cannot make a new request because the previous response is still open" is seen when the endpoint returns a 4xx error.
Due to connection not being closed correctly in retry interceptor.

Testing
New unit tests added in CfClientTest for testing 4xx reponses + manual testing

Author: andybharness

README.md

@@ -69,7 +69,7 @@ The first step is to install the FF SDK as a dependency in your application usin
 
 Refer to the [Harness Feature Flag Java Server SDK](https://mvnrepository.com/artifact/io.harness/ff-java-server-sdk) to identify the latest version for your build automation tool.
 
-This section lists dependencies for Maven and Gradle and uses the 1.1.9 version as an example:
+This section lists dependencies for Maven and Gradle and uses the 1.1.10 version as an example:
 
 #### Maven
 
@@ -78,14 +78,14 @@ Add the following Maven dependency in your project's pom.xml file:
 <dependency>
     <groupId>io.harness</groupId>
     <artifactId>ff-java-server-sdk</artifactId>
-    <version>1.1.9</version>
+    <version>1.1.10</version>
 </dependency>
 ```
 
 #### Gradle
 
 ```
-implementation group: 'io.harness', name: 'ff-java-server-sdk', version: '1.1.9'
+implementation group: 'io.harness', name: 'ff-java-server-sdk', version: '1.1.10'
 ```
 
 ### Code Sample

examples/pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>io.harness.featureflags</groupId>
     <artifactId>examples</artifactId>
-    <version>1.1.10-SNAPSHOT</version>
+    <version>1.1.10</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.10-SNAPSHOT</version>
+            <version>1.1.10</version>
         </dependency>
 
         <dependency>

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>io.harness</groupId>
     <artifactId>ff-java-server-sdk</artifactId>
-    <version>1.1.10-SNAPSHOT</version>
+    <version>1.1.10</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

@@ -1,5 +1,7 @@
 package io.harness.cf.client.api;
 
+import static com.google.common.util.concurrent.Service.State.*;
+
 import com.google.common.base.Strings;
 import com.google.common.util.concurrent.Service;
 import com.google.gson.JsonObject;
@@ -132,7 +134,17 @@ protected void onUnauthorized() {
     if (options.isAnalyticsEnabled()) {
       metricsProcessor.stop();
     }
-    authService.startAsync();
+
+    if (authService.state() == TERMINATED || authService.state() == FAILED) {
+      authService.close();
+      authService = new AuthService(this.connector, options.getPollIntervalInSeconds(), this);
+    }
+
+    if (authService.state() != RUNNING) {
+      authService.startAsync();
+    }
+
+    log.info("Finished re-auth, auth service state={}", authService.state());
   }
 
   @Override

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

@@ -72,7 +72,7 @@ public CompletableFuture<List<Segment>> retrieveSegments() {
       completableFuture.complete(segments);
     } catch (ConnectorException e) {
       log.error(
-          "Exception was raised when fetching flags data with the message {}", e.getMessage());
+          "Exception was raised when fetching flags data with the message {}", e.getMessage(), e);
       completableFuture.completeExceptionally(e);
     }
     return completableFuture;

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

@@ -1,7 +1,26 @@
 package io.harness.cf.client.connector;
 
 public class ConnectorException extends Exception {
+
+  private final int httpCode;
+  private final String httpReason;
+
   public ConnectorException(String message) {
     super(message);
+    this.httpCode = 0;
+    this.httpReason = "";
+  }
+
+  public ConnectorException(String message, int httpCode, String httpMsg) {
+    super(message);
+    this.httpCode = httpCode;
+    this.httpReason = httpMsg;
+  }
+
+  @Override
+  public String getMessage() {
+    final String httpInfo =
+        (httpCode == 0) ? "" : String.format(" - httpCode=%d httpReason=%s", httpCode, httpReason);
+    return super.getMessage() + httpInfo;
   }
 }

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

@@ -8,14 +8,14 @@
 import io.harness.cf.client.dto.Claim;
 import io.harness.cf.client.logger.LogUtil;
 import io.harness.cf.model.*;
-import java.lang.ref.WeakReference;
+import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.*;
 import lombok.NonNull;
 import lombok.extern.slf4j.Slf4j;
+import okhttp3.Interceptor;
 import okhttp3.Request;
 import okhttp3.Response;
-import okhttp3.internal.Util;
 import org.slf4j.MDC;
 
 @Slf4j
@@ -25,8 +25,6 @@ public class HarnessConnector implements Connector, AutoCloseable {
   private final MetricsApi metricsApi;
   private final String apiKey;
   private final HarnessConfig options;
-  private WeakReference<Response> responseReference;
-  private WeakReference<Response> metricsResponseReference;
 
   private String token;
   private String environment;
@@ -48,12 +46,12 @@ public HarnessConnector(@NonNull String apiKey) {
   public HarnessConnector(@NonNull final String apiKey, @NonNull final HarnessConfig options) {
     this.apiKey = apiKey;
     this.options = options;
-    this.api = new ClientApi(makeApiClient());
-    this.metricsApi = new MetricsApi(makeMetricsApiClient());
+    this.api = new ClientApi(makeApiClient(2000));
+    this.metricsApi = new MetricsApi(makeMetricsApiClient(2000));
     log.info("Connector initialized, with options " + options);
   }
 
-  protected ApiClient makeApiClient() {
+  ApiClient makeApiClient(int retryBackOfDelay) {
     final ApiClient apiClient = new ApiClient();
     apiClient.setBasePath(options.getConfigUrl());
     apiClient.setConnectTimeout(options.getConnectionTimeout());
@@ -66,45 +64,28 @@ protected ApiClient makeApiClient() {
         apiClient
             .getHttpClient()
             .newBuilder()
-            .addInterceptor(
-                chain -> {
-                  final Request request =
-                      chain
-                          .request()
-                          .newBuilder()
-                          .addHeader("X-Request-ID", getRequestID())
-                          .build();
-                  log.info("interceptor: requesting url {}", request.url().url());
-                  Response response;
-                  if (responseReference != null) {
-                    response = responseReference.get();
-                    if (response != null) {
-                      if (response.isSuccessful()) {
-                        log.debug(
-                            "interceptor: closing the response url={}", response.request().url());
-                      } else {
-                        log.warn(
-                            "interceptor: closing the faulty response url={}",
-                            response.request().url());
-                      }
-                      Util.closeQuietly(response);
-                    }
-                  }
-                  response = chain.proceed(request);
-                  responseReference = new WeakReference<>(response);
-                  if (response.code() == 403 && onUnauthorized != null) {
-                    onUnauthorized.run();
-                  }
-
-                  return response;
-                })
-            .addInterceptor(new RetryInterceptor(3, 2000))
+            .addInterceptor(this::interceptor)
+            .addInterceptor(new RetryInterceptor(3, retryBackOfDelay))
             .build());
     log.info("apiClient definition complete");
     return apiClient;
   }
 
-  protected ApiClient makeMetricsApiClient() {
+  private Response interceptor(Interceptor.Chain chain) throws IOException {
+    final Request request =
+        chain.request().newBuilder().addHeader("X-Request-ID", getRequestID()).build();
+    log.info("interceptor: requesting url {}", request.url().url());
+
+    Response response = chain.proceed(request);
+
+    if (response.code() == 403 && onUnauthorized != null) {
+      onUnauthorized.run();
+    }
+
+    return response;
+  }
+
+  ApiClient makeMetricsApiClient(int retryBackoffDelay) {
     final int maxTimeout = 30 * 60 * 1000;
     final ApiClient apiClient = new ApiClient();
     apiClient.setBasePath(options.getEventUrl());
@@ -117,41 +98,21 @@ protected ApiClient makeMetricsApiClient() {
         apiClient
             .getHttpClient()
             .newBuilder()
-            .addInterceptor(
-                chain -> {
-                  final Request request =
-                      chain
-                          .request()
-                          .newBuilder()
-                          .addHeader("X-Request-ID", getRequestID())
-                          .build();
-                  log.info("metrics interceptor: requesting url {}", request.url().url());
-                  Response response;
-                  if (metricsResponseReference != null) {
-                    response = metricsResponseReference.get();
-                    if (response != null) {
-                      if (response.isSuccessful()) {
-                        log.debug(
-                            "metrics interceptor: closing the response url={}",
-                            response.request().url());
-                      } else {
-                        log.warn(
-                            "metrics interceptor: closing the faulty response url={}",
-                            response.request().url());
-                      }
-                      Util.closeQuietly(response);
-                    }
-                  }
-                  response = chain.proceed(request);
-                  metricsResponseReference = new WeakReference<>(response);
-                  return response;
-                })
-            .addInterceptor(new RetryInterceptor(3, 2000))
+            .addInterceptor(this::metricsInterceptor)
+            .addInterceptor(new RetryInterceptor(3, retryBackoffDelay))
             .build());
     log.info("metricsApiClient definition complete");
     return apiClient;
   }
 
+  private Response metricsInterceptor(Interceptor.Chain chain) throws IOException {
+    final Request request =
+        chain.request().newBuilder().addHeader("X-Request-ID", getRequestID()).build();
+    log.info("metrics interceptor: requesting url {}", request.url().url());
+
+    return chain.proceed(request);
+  }
+
   protected String getRequestID() {
     String requestId = MDC.get(REQUEST_ID_KEY);
     if (requestId == null) {
@@ -176,10 +137,11 @@ public String authenticate() throws ConnectorException {
       if (apiException.getCode() == 401 || apiException.getCode() == 403) {
         String errorMsg = String.format("Invalid apiKey %s. SDK will serve default values", apiKey);
         log.error(errorMsg);
-        throw new ConnectorException(errorMsg);
+        throw new ConnectorException(errorMsg, apiException.getCode(), apiException.getMessage());
       }
       log.error("Failed to get auth token", apiException);
-      throw new ConnectorException(apiException.getMessage());
+      throw new ConnectorException(
+          apiException.getMessage(), apiException.getCode(), apiException.getMessage());
     } catch (Throwable ex) {
       log.error("Unexpected exception", ex);
       throw ex;
@@ -224,15 +186,17 @@ public List<FeatureConfig> getFlags() throws ConnectorException {
           featureConfig.size(),
           this.environment,
           this.cluster);
-      log.info("Got the following features: " + featureConfig);
+      if (log.isTraceEnabled()) {
+        log.trace("Got the following features: " + featureConfig);
+      }
       return featureConfig;
     } catch (ApiException e) {
       log.error(
           "Exception was raised while fetching the flags on env {} and cluster {}",
           this.environment,
           this.cluster,
           e);
-      throw new ConnectorException(e.getMessage());
+      throw new ConnectorException(e.getMessage(), e.getCode(), e.getMessage());
     } finally {
       MDC.remove(REQUEST_ID_KEY);
     }
@@ -260,7 +224,7 @@ public FeatureConfig getFlag(@NonNull final String identifier) throws ConnectorE
           this.environment,
           this.cluster,
           e);
-      throw new ConnectorException(e.getMessage());
+      throw new ConnectorException(e.getMessage(), e.getCode(), e.getMessage());
     } finally {
       MDC.remove(REQUEST_ID_KEY);
     }
@@ -283,11 +247,13 @@ public List<Segment> getSegments() throws ConnectorException {
       return allSegments;
     } catch (ApiException e) {
       log.error(
-          "Exception was raised while fetching the target groups on env {} and cluster {}",
+          "Exception was raised while fetching the target groups on env {} and cluster {} : httpCode={} message={}",
           this.environment,
           this.cluster,
+          e.getCode(),
+          e.getMessage(),
           e);
-      throw new ConnectorException(e.getMessage());
+      throw new ConnectorException(e.getMessage(), e.getCode(), e.getMessage());
     } finally {
       MDC.remove(REQUEST_ID_KEY);
     }
@@ -317,7 +283,7 @@ public Segment getSegment(@NonNull final String identifier) throws ConnectorExce
           this.environment,
           this.cluster,
           e);
-      throw new ConnectorException(e.getMessage());
+      throw new ConnectorException(e.getMessage(), e.getCode(), e.getMessage());
     } finally {
       MDC.remove(REQUEST_ID_KEY);
     }
@@ -340,7 +306,7 @@ public void postMetrics(@NonNull final Metrics metrics) throws ConnectorExceptio
           this.environment,
           this.cluster,
           e);
-      throw new ConnectorException(e.getMessage());
+      throw new ConnectorException(e.getMessage(), e.getCode(), e.getMessage());
     } finally {
       MDC.remove(REQUEST_ID_KEY);
     }
@@ -365,28 +331,29 @@ public Service stream(@NonNull final Updater updater) {
 
   @Override
   public void close() {
-    log.info("closing connector");
-    if (responseReference != null) {
-      final Response response = responseReference.get();
-      if (response != null) {
-        Util.closeQuietly(response);
-      }
-      responseReference = null;
-    }
-    if (metricsResponseReference != null) {
-      final Response response = metricsResponseReference.get();
-      if (response != null) {
-        Util.closeQuietly(response);
-      }
-      metricsResponseReference = null;
-    }
+    log.debug("closing connector");
     api.getApiClient().getHttpClient().connectionPool().evictAll();
-    log.info("All apiClient connections evicted");
+    log.debug("All apiClient connections evicted");
     metricsApi.getApiClient().getHttpClient().connectionPool().evictAll();
-    log.info("All metricsApiClient connections evicted");
+    log.debug("All metricsApiClient connections evicted");
     if (eventSource != null) {
       eventSource.close();
     }
-    log.info("connector closed!");
+    log.debug("connector closed!");
+  }
+
+  /* package private - should not be used outside of tests */
+
+  HarnessConnector(
+      @NonNull final String apiKey, @NonNull final HarnessConfig options, int retryBackOffDelay) {
+    this.apiKey = apiKey;
+    this.options = options;
+    this.api = new ClientApi(makeApiClient(retryBackOffDelay));
+    this.metricsApi = new MetricsApi(makeMetricsApiClient(retryBackOffDelay));
+    log.info(
+        "Connector initialized, with options "
+            + options
+            + " and retry backoff delay "
+            + retryBackOffDelay);
   }
 }