[FFM-10760] - Use a single ExecutorService for UpdateProcessor (#182)

* [FFM-10760] - Use a single ExecutorService for UpdateProcessor

What
- Change UpdateProcessor to use a single thread pool for the life time of the service
- Updated `processFlag()` and `processSegment()` to catch `Throwable` to prevent executor threads dying
- Drop some verbose logs to trace to make debug level more readable

Why
We are getting reports in the field that the following error message is being logged

`Update processor is terminating/restarting. Update skipped: <EVENT JSON>`

Looking at the code this can only happen if the SSE stream drops and we attempt to call `restart()` while events are still arriving from the old stream. We should modify the code to avoid creating/destroying the thread pool on each disconnect and instead use a single thread pool for the lifetime of the UpdateProcessor which is only destroyed when `close()` is called. This will allow the removal of the following defensive check and will help avoid dropping SSE messages that did make it before the disconnect:

```
   if (executor.isShutdown() || executor.isTerminated()) {
      log.warn("Update processor is terminating/restarting. Update skipped: {}", message);
      return;
   }
```

Author: andybharness

examples/src/main/java/io/harness/ff/examples/GettingStarted.java

@@ -9,19 +9,18 @@
 
 public class GettingStarted {
     // API Key - set this as an env variable
-    private static String apiKey = getEnvOrDefault("FF_API_KEY", "");
+    private static final String apiKey = getEnvOrDefault("FF_API_KEY", "");
 
     // Flag Identifier
-    private static String flagName = getEnvOrDefault("FF_FLAG_NAME", "harnessappdemodarkmode");
+    private static final String flagName = getEnvOrDefault("FF_FLAG_NAME", "harnessappdemodarkmode");
 
     private static final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
 
     public static void main(String[] args) {
         System.out.println("Harness SDK Getting Started");
 
-        try {
-            //Create a Feature Flag Client
-            CfClient cfClient = new CfClient(apiKey, BaseConfig.builder().build());
+        //Create a Feature Flag Client
+        try (CfClient cfClient = new CfClient(apiKey, BaseConfig.builder().build())) {
             cfClient.waitForInitialization();
 
             // Create a target (different targets can get different results based on rules.  This includes a custom attribute 'location')
@@ -46,7 +45,6 @@ public static void main(String[] args) {
             // Close the SDK
             System.out.println("Cleaning up...");
             scheduler.shutdownNow();
-            cfClient.close();
 
         } catch (Exception e) {
             e.printStackTrace();

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

@@ -20,37 +20,37 @@ public class CaffeineCache implements Cache {
 
   public CaffeineCache(int size) {
     cache = Caffeine.newBuilder().maximumSize(size).build();
-    log.debug("CaffeineCache initialized with size {}", size);
+    log.trace("CaffeineCache initialized with size {}", size);
   }
 
   @Override
   public void set(@NonNull String key, @NonNull Object value) {
     cache.put(key, value);
-    log.debug("New value in the cache with key {} and value {}", key, value);
+    log.trace("New value in the cache with key '{}' and value '{}'", key, value);
   }
 
   @Override
   @Nullable
   public Object get(@NonNull String key) {
     Object value = cache.getIfPresent(key);
     if (value != null) {
-      log.debug("Key {} found in cache with value {}", key, value);
+      log.trace("Key '{}' found in cache with value '{}'", key, value);
     } else {
-      log.debug("Key {} not found in cache", key);
+      log.trace("Key '{}' not found in cache", key);
     }
     return value;
   }
 
   @Override
   public void delete(@NonNull String key) {
     cache.invalidate(key);
-    log.debug("Key {} removed from cache", key);
+    log.trace("Key {} removed from cache", key);
   }
 
   @Override
   public List<String> keys() {
     List<String> keys = new ArrayList<>(cache.asMap().keySet());
-    log.debug("Keys in cache {}", keys);
+    log.trace("Keys in cache {}", keys);
     return keys;
   }
 }

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

@@ -60,7 +60,7 @@ protected Optional<Variation> findVariation(
     }
     Optional<Variation> variation =
         variations.stream().filter(v -> v.getIdentifier().equals(identifier)).findFirst();
-    log.debug("Variation {} found in variations {}", identifier, variations);
+    log.trace("Variation {} found in variations {}", identifier, variations);
     return variation;
   }
 

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

@@ -9,7 +9,6 @@
 import io.harness.cf.model.Segment;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
 import lombok.NonNull;
 import lombok.extern.slf4j.Slf4j;
 
@@ -18,7 +17,7 @@ class UpdateProcessor implements AutoCloseable {
   private final Connector connector;
   private final Repository repository;
   private final Updater updater;
-  private final ExecutorService executor = Executors.newFixedThreadPool(100);
+  private final ExecutorService executor = Executors.newFixedThreadPool(10);
 
   private Service stream;
 
@@ -42,6 +41,9 @@ public void start() {
     }
 
     try {
+      if (stream != null) {
+        stream.close();
+      }
       stream = connector.stream(this.updater);
       stream.start();
       running = true;
@@ -65,24 +67,14 @@ public void stop() {
         stream.stop();
         running = false;
       }
-      executor.shutdown();
-      boolean result = executor.awaitTermination(3, TimeUnit.SECONDS);
-      if (result) {
-        log.debug("All tasks done");
-      } else {
-        log.warn("UpdateProcessor: timeout while wait threads to finish!");
-      }
+
     } catch (InterruptedException e) {
       log.error("Exception was raised when stopping update tasks", e);
       Thread.currentThread().interrupt();
     }
   }
 
   public void update(@NonNull final Message message) {
-    if (executor.isShutdown() || executor.isTerminated()) {
-      log.warn("Update processor is terminating/restarting. Update skipped: {}", message);
-      return;
-    }
 
     if (message.getDomain().equals("flag")) {
       log.debug("execute processFlag with message {}", message);
@@ -103,13 +95,13 @@ protected Runnable processFlag(@NonNull final Message message) {
           final FeatureConfig config = connector.getFlag(message.getIdentifier());
           if (config != null) {
             repository.setFlag(message.getIdentifier(), config);
-            log.debug("Set new segment with key {} and value {}", message.getIdentifier(), config);
+            log.trace("Set new segment with key {} and value {}", message.getIdentifier(), config);
           }
         } else if (message.getEvent().equals("delete")) {
           log.debug("Delete flag with key {}", message.getIdentifier());
           repository.deleteFlag(message.getIdentifier());
         }
-      } catch (ConnectorException e) {
+      } catch (Throwable e) {
         log.error(
             "Exception was raised when fetching flag '{}' with the message {}",
             message.getIdentifier(),
@@ -124,14 +116,14 @@ protected Runnable processSegment(@NonNull final Message message) {
         if (message.getEvent().equals("create") || message.getEvent().equals("patch")) {
           final Segment segment = connector.getSegment(message.getIdentifier());
           if (segment != null) {
-            log.debug("Set new segment with key {} and value {}", message.getIdentifier(), segment);
+            log.trace("Set new segment with key {} and value {}", message.getIdentifier(), segment);
             repository.setSegment(message.getIdentifier(), segment);
           }
         } else if (message.getEvent().equals("delete")) {
           log.debug("Delete segment with key {}", message.getIdentifier());
           repository.deleteSegment(message.getIdentifier());
         }
-      } catch (ConnectorException e) {
+      } catch (Throwable e) {
         log.error(
             "Exception was raised when fetching segment '{}' with the message {}",
             message.getIdentifier(),
@@ -152,6 +144,8 @@ public void close() {
         Thread.currentThread().interrupt();
       }
     }
+
+    executor.shutdownNow();
     log.debug("UpdateProcessor closed");
   }
 

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

@@ -190,12 +190,18 @@ public void onResponse(@NotNull Call call, @NotNull Response response) throws IO
       log.warn("End of SSE stream");
       updater.onDisconnected("End of SSE stream");
     } catch (Throwable ex) {
-      log.warn("SSE Stream aborted: " + ex.getMessage());
+      log.warn("SSE Stream aborted: " + getExceptionMsg(ex));
       log.trace("SSE Stream aborted trace", ex);
-      updater.onDisconnected(ex.getMessage());
+      updater.onDisconnected(getExceptionMsg(ex));
     }
   }
 
+  private String getExceptionMsg(Throwable ex) {
+    return (ex.getMessage() == null || "null".equals(ex.getMessage()))
+        ? ex.getClass().getSimpleName()
+        : ex.getMessage();
+  }
+
   private static class SSEStreamException extends RuntimeException {
     public SSEStreamException(String msg, Throwable cause) {
       super(msg, cause);

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

@@ -404,7 +404,7 @@ public Service stream(@NonNull final Updater updater) throws ConnectorException
             map,
             updater,
             Math.max(options.getSseReadTimeout(), 1),
-            2_000,
+            ThreadLocalRandom.current().nextInt(5000, 10000),
             options.getTlsTrustedCAs());
     return eventSource;
   }