[Dataflow Streaming] Make SideInputCache bytes and expiry configurable (

apache#29871)

Co-authored-by: Arun Pandian <pandiana@google.com>

runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowPipelineDebugOptions.java

@@ -239,15 +239,27 @@ public Dataflow create(PipelineOptions options) {
   /**
    * The size of the worker's in-memory cache, in megabytes.
    *
-   * <p>Currently, this cache is used for storing read values of side inputs. as well as the state
-   * for streaming jobs.
+   * <p>Currently, this cache is used for storing read values of side inputs in batch as well as the
+   * user state for streaming jobs.
    */
   @Description("The size of the worker's in-memory cache, in megabytes.")
   @Default.Integer(100)
   Integer getWorkerCacheMb();
 
   void setWorkerCacheMb(Integer value);
 
+  @Description("The size of the streaming worker's side input cache, in megabytes.")
+  @Default.Integer(100)
+  Integer getStreamingSideInputCacheMb();
+
+  void setStreamingSideInputCacheMb(Integer value);
+
+  @Description("The expiry for streaming worker's side input cache entries, in milliseconds.")
+  @Default.Integer(60 * 1000) // 1 minute
+  Integer getStreamingSideInputCacheExpirationMillis();
+
+  void setstreamingSideInputCacheExpirationMillis(Integer value);
+
   /**
    * The amount of time before UnboundedReaders are considered idle and closed during streaming
    * execution.

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java

@@ -424,7 +424,7 @@ public void run() {
     this.metricTrackingWindmillServer =
         new MetricTrackingWindmillServerStub(windmillServer, memoryMonitor, windmillServiceEnabled);
     this.metricTrackingWindmillServer.start();
-    this.sideInputStateFetcher = new SideInputStateFetcher(metricTrackingWindmillServer);
+    this.sideInputStateFetcher = new SideInputStateFetcher(metricTrackingWindmillServer, options);
     this.clientId = clientIdGenerator.nextLong();
 
     for (MapTask mapTask : mapTasks) {

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/sideinput/SideInputCache.java

@@ -23,6 +23,7 @@
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import org.apache.beam.runners.dataflow.options.DataflowPipelineDebugOptions;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
 import org.apache.beam.sdk.values.TupleTag;
 import org.apache.beam.sdk.values.TypeDescriptor;
@@ -40,20 +41,20 @@
 @CheckReturnValue
 final class SideInputCache {
 
-  private static final long MAXIMUM_CACHE_WEIGHT = 100000000; /* 100 MB */
-  private static final long CACHE_ENTRY_EXPIRY_MINUTES = 1L;
+  private static final long BYTES_PER_MB = 1024 * 1024;
 
   private final Cache<Key<?>, SideInput<?>> sideInputCache;
 
   SideInputCache(Cache<Key<?>, SideInput<?>> sideInputCache) {
     this.sideInputCache = sideInputCache;
   }
 
-  static SideInputCache create() {
+  static SideInputCache create(DataflowPipelineDebugOptions options) {
     return new SideInputCache(
         CacheBuilder.newBuilder()
-            .maximumWeight(MAXIMUM_CACHE_WEIGHT)
-            .expireAfterWrite(CACHE_ENTRY_EXPIRY_MINUTES, TimeUnit.MINUTES)
+            .maximumWeight(options.getStreamingSideInputCacheMb() * BYTES_PER_MB)
+            .expireAfterWrite(
+                options.getStreamingSideInputCacheExpirationMillis(), TimeUnit.MILLISECONDS)
             .weigher((Weigher<Key<?>, SideInput<?>>) (id, entry) -> entry.size())
             .build());
   }

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/sideinput/SideInputStateFetcher.java

@@ -29,6 +29,7 @@
 import java.util.concurrent.Callable;
 import javax.annotation.concurrent.NotThreadSafe;
 import org.apache.beam.runners.core.InMemoryMultimapSideInputView;
+import org.apache.beam.runners.dataflow.options.DataflowPipelineDebugOptions;
 import org.apache.beam.runners.dataflow.worker.MetricTrackingWindmillServerStub;
 import org.apache.beam.runners.dataflow.worker.WindmillTimeUtils;
 import org.apache.beam.runners.dataflow.worker.windmill.Windmill;
@@ -61,8 +62,9 @@ public class SideInputStateFetcher {
   private final MetricTrackingWindmillServerStub server;
   private long bytesRead = 0L;
 
-  public SideInputStateFetcher(MetricTrackingWindmillServerStub server) {
-    this(server, SideInputCache.create());
+  public SideInputStateFetcher(
+      MetricTrackingWindmillServerStub server, DataflowPipelineDebugOptions options) {
+    this(server, SideInputCache.create(options));
   }
 
   SideInputStateFetcher(MetricTrackingWindmillServerStub server, SideInputCache sideInputCache) {

runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/streaming/sideinput/SideInputStateFetcherTest.java

@@ -32,13 +32,15 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
+import org.apache.beam.runners.dataflow.options.DataflowPipelineDebugOptions;
 import org.apache.beam.runners.dataflow.worker.MetricTrackingWindmillServerStub;
 import org.apache.beam.runners.dataflow.worker.windmill.Windmill;
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.coders.ListCoder;
 import org.apache.beam.sdk.coders.StringUtf8Coder;
 import org.apache.beam.sdk.coders.VarLongCoder;
 import org.apache.beam.sdk.coders.VoidCoder;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
 import org.apache.beam.sdk.testing.TestPipeline;
 import org.apache.beam.sdk.transforms.Create;
 import org.apache.beam.sdk.transforms.Sum;
@@ -79,7 +81,9 @@ public void setUp() {
 
   @Test
   public void testFetchGlobalDataBasic() throws Exception {
-    SideInputStateFetcher fetcher = new SideInputStateFetcher(server);
+    SideInputStateFetcher fetcher =
+        new SideInputStateFetcher(
+            server, PipelineOptionsFactory.as(DataflowPipelineDebugOptions.class));
 
     ByteStringOutputStream stream = new ByteStringOutputStream();
     ListCoder.of(StringUtf8Coder.of())
@@ -147,7 +151,9 @@ public void testFetchGlobalDataBasic() throws Exception {
 
   @Test
   public void testFetchGlobalDataNull() throws Exception {
-    SideInputStateFetcher fetcher = new SideInputStateFetcher(server);
+    SideInputStateFetcher fetcher =
+        new SideInputStateFetcher(
+            server, PipelineOptionsFactory.as(DataflowPipelineDebugOptions.class));
 
     ByteStringOutputStream stream = new ByteStringOutputStream();
     ListCoder.of(VoidCoder.of())
@@ -302,7 +308,9 @@ public void testFetchGlobalDataCacheOverflow() throws Exception {
 
   @Test
   public void testEmptyFetchGlobalData() throws Exception {
-    SideInputStateFetcher fetcher = new SideInputStateFetcher(server);
+    SideInputStateFetcher fetcher =
+        new SideInputStateFetcher(
+            server, PipelineOptionsFactory.as(DataflowPipelineDebugOptions.class));
 
     ByteString encodedIterable = ByteString.EMPTY;