Revert "Allow dropLataData in GBK for SamzaRunner (apache#28461)"

This reverts commit 6765bf9.
linkedin · Feb 28, 2024 · d334b91 · d334b91
1 parent 0b94f58
commit d334b91
Show file tree

Hide file tree

Showing 3 changed files with 2 additions and 139 deletions.
diff --git a/runners/samza/src/main/java/org/apache/beam/runners/samza/SamzaPipelineOptions.java b/runners/samza/src/main/java/org/apache/beam/runners/samza/SamzaPipelineOptions.java
@@ -173,10 +173,4 @@ public ExecutorService create(PipelineOptions options) {
           new ThreadFactoryBuilder().setNameFormat("Process Element Thread-%d").build());
     }
   }
-
-  @Description("Enable/disable late data dropping in GroupByKey/Combine transforms")
-  @Default.Boolean(false)
-  boolean getDropLateData();
-
-  void setDropLateData(boolean dropLateData);
 }
diff --git a/runners/samza/src/main/java/org/apache/beam/runners/samza/runtime/GroupByKeyOp.java b/runners/samza/src/main/java/org/apache/beam/runners/samza/runtime/GroupByKeyOp.java
@@ -180,19 +180,11 @@ public TimerInternals timerInternals() {
             DoFnSchemaInformation.create(),
             Collections.emptyMap());
 
-    final DoFnRunner<KeyedWorkItem<K, InputT>, KV<K, OutputT>> dropLateDataRunner =
-        pipelineOptions.getDropLateData()
-            ? DoFnRunners.lateDataDroppingRunner(
-                doFnRunner, keyedInternals.timerInternals(), windowingStrategy)
-            : doFnRunner;
-
     final SamzaExecutionContext executionContext =
         (SamzaExecutionContext) context.getApplicationContainerContext();
-    final DoFnRunner<KeyedWorkItem<K, InputT>, KV<K, OutputT>> doFnRunnerWithMetrics =
+    this.fnRunner =
         DoFnRunnerWithMetrics.wrap(
-            dropLateDataRunner, executionContext.getMetricsContainer(), transformFullName);
-
-    this.fnRunner = new DoFnRunnerWithKeyedInternals<>(doFnRunnerWithMetrics, keyedInternals);
+            doFnRunner, executionContext.getMetricsContainer(), transformFullName);
   }
 
   @Override

diff --git a/runners/samza/src/test/java/org/apache/beam/runners/samza/runtime/GroupByKeyOpTest.java b/runners/samza/src/test/java/org/apache/beam/runners/samza/runtime/GroupByKeyOpTest.java