From 1d62a653ed69c4fb66cc4d6062bbcca09f97cea0 Mon Sep 17 00:00:00 2001
From: "balazs.nemeth@doctusoft.com" <balazs.nemeth@doctusoft.com>
Date: Wed, 10 May 2023 03:14:35 +0200
Subject: [PATCH] #26395 Disabling possibly unnecessary prefetching during
 GroupIntoBatches by using an experimental flag

---
 .../beam/sdk/transforms/GroupIntoBatches.java | 22 +++++++++++++++----
 1 file changed, 18 insertions(+), 4 deletions(-)

diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupIntoBatches.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupIntoBatches.java
index cb8e226fc1dd..13315b054c21 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupIntoBatches.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupIntoBatches.java
@@ -26,6 +26,8 @@
 import javax.annotation.Nullable;
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.options.ExperimentalOptions;
+import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.state.BagState;
 import org.apache.beam.sdk.state.CombiningState;
 import org.apache.beam.sdk.state.StateSpec;
@@ -110,6 +112,10 @@
 public class GroupIntoBatches<K, InputT>
     extends PTransform<PCollection<KV<K, InputT>>, PCollection<KV<K, Iterable<InputT>>>> {
 
+  /** Experiment to "avoid possibly unnecessary prefetching". */
+  public static final String AVOID_POSSIBLY_UNNECESSARY_PREFETCHING =
+      "unsafely_attempt_to_process_unbounded_data_in_batch_mode";
+
   /**
    * Wrapper class for batching parameters supplied by users. Shared by both {@link
    * GroupIntoBatches} and {@link GroupIntoBatches.WithShardedKey}.
@@ -343,7 +349,8 @@ public PCollection<KV<K, Iterable<InputT>>> expand(PCollection<KV<K, InputT>> in
                 weigher,
                 params.getMaxBufferingDuration(),
                 allowedLateness,
-                valueCoder)));
+                valueCoder,
+                input.getPipeline().getOptions())));
   }
 
   @VisibleForTesting
@@ -418,7 +425,8 @@ private static class GroupIntoBatchesDoFn<K, InputT>
         @Nullable SerializableFunction<InputT, Long> weigher,
         Duration maxBufferingDuration,
         Duration allowedLateness,
-        Coder<InputT> inputValueCoder) {
+        Coder<InputT> inputValueCoder,
+        PipelineOptions options) {
       this.batchSize = batchSize;
       this.batchSizeBytes = batchSizeBytes;
       this.weigher = weigher;
@@ -457,8 +465,14 @@ public long apply(long left, long right) {
       this.timerTsSpec = StateSpecs.value();
       this.minBufferedTsSpec = StateSpecs.combining(minCombineFn);
 
-      // Prefetch every 20% of batchSize elements. Do not prefetch if batchSize is too little
-      this.prefetchFrequency = ((batchSize / 5) <= 1) ? Long.MAX_VALUE : (batchSize / 5);
+      // Prefetch every 20% of batchSize elements. Do not prefetch ...
+      if (ExperimentalOptions.hasExperiment(options, AVOID_POSSIBLY_UNNECESSARY_PREFETCHING)) {
+        // ... if we have disabled prefetching
+        this.prefetchFrequency = Long.MAX_VALUE;
+      } else {
+        // ... or if the batchSize is too little
+        this.prefetchFrequency = ((batchSize / 5) <= 1) ? Long.MAX_VALUE : (batchSize / 5);
+      }
     }
 
     @Override