SDC-12744. Classification Previews don't always get cleaned up

Changes the expiration for the preview cache from 30 minutes to 5 - there's no
need to keep them around for that long.  Also adds a background thread to call
the cache cleanup method every 5 minutes because Guava isn't actually removing
expired previews otherwise.  We normally invalidate specific previews from the
cache when retrieving their output, so this isn't typically a problem; but if
the output isn't retrieved for some reason, the preview isn't cleaned up and
removed otherwise.

Change-Id: Ib8eee76c7693b480410ed66f8314b6130e7bf6ea
Reviewed-on: https://review.streamsets.net/c/datacollector/+/27176
Tested-by: StreamSets CI <streamsets-ci-spam@streamsets.com>
Reviewed-by: Pasindu Perera <pasindu@streamsets.com>
Reviewed-by: Tony McKeehan <anthony@streamsets.com>

container/src/main/java/com/streamsets/datacollector/execution/manager/standalone/StandaloneAndClusterPipelineManager.java

@@ -63,6 +63,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
@@ -229,7 +230,7 @@ public void runTask() {
       runtimeInfo.getMetrics(),
       "manager-previewer-cache",
         CacheBuilder.newBuilder()
-          .expireAfterAccess(30, TimeUnit.MINUTES).removalListener((RemovalListener<String, Previewer>) removal -> {
+          .expireAfterAccess(5, TimeUnit.MINUTES).removalListener((RemovalListener<String, Previewer>) removal -> {
             Previewer previewer = removal.getValue();
             LOG.warn(
                 "Evicting idle previewer '{}::{}'::'{}' in status '{}'",
@@ -251,6 +252,12 @@ public void runTask() {
           }).build()
     );
 
+    // Create a background thread to cleanup the previewerCache because guava doesn't always want to do it on its own
+    Executors.newSingleThreadScheduledExecutor().scheduleWithFixedDelay(() -> {
+      LOG.debug("Triggering previewer cache cleanup");
+      previewerCache.cleanUp();
+    }, 5, 5, TimeUnit.MINUTES);
+
     runnerCache = new MetricsCache<>(
       runtimeInfo.getMetrics(),
       "manager-runner-cache",