Revert "Add Redistribute translation to Samza runner"

This reverts commit 8f1d3da.
kennknowles · Apr 30, 2024 · ab9af72 · ab9af72
1 parent ece456c
commit ab9af72
Show file tree

Hide file tree

Showing 3 changed files with 2 additions and 77 deletions.
diff --git a/.../src/main/java/org/apache/beam/runners/samza/translation/RedistributeByKeyTranslator.java b/.../src/main/java/org/apache/beam/runners/samza/translation/RedistributeByKeyTranslator.java
diff --git a/...rs/samza/src/main/java/org/apache/beam/runners/samza/translation/ReshuffleTranslator.java b/...rs/samza/src/main/java/org/apache/beam/runners/samza/translation/ReshuffleTranslator.java
@@ -42,16 +42,6 @@
 public class ReshuffleTranslator<K, InT, OutT>
     implements TransformTranslator<PTransform<PCollection<KV<K, InT>>, PCollection<KV<K, OutT>>>> {
 
-  private final String prefix;
-
-  ReshuffleTranslator(String prefix) {
-    this.prefix = prefix;
-  }
-
-  ReshuffleTranslator() {
-    this("rshfl-");
-  }
-
   @Override
   public void translate(
       PTransform<PCollection<KV<K, InT>>, PCollection<KV<K, OutT>>> transform,
@@ -70,7 +60,7 @@ public void translate(
             inputStream,
             inputCoder.getKeyCoder(),
             elementCoder,
-            prefix + ctx.getTransformId(),
+            "rshfl-" + ctx.getTransformId(),
             ctx.getPipelineOptions().getMaxSourceParallelism() > 1);
 
     ctx.registerMessageStream(output, outputStream);
@@ -93,7 +83,7 @@ public void translatePortable(
             inputStream,
             ((KvCoder<K, InT>) windowedInputCoder.getValueCoder()).getKeyCoder(),
             windowedInputCoder,
-            prefix + ctx.getTransformId(),
+            "rshfl-" + ctx.getTransformId(),
             ctx.getPipelineOptions().getMaxSourceParallelism() > 1);
 
     ctx.registerMessageStream(outputId, outputStream);

diff --git a/...amza/src/main/java/org/apache/beam/runners/samza/translation/SamzaPipelineTranslator.java b/...amza/src/main/java/org/apache/beam/runners/samza/translation/SamzaPipelineTranslator.java
@@ -179,7 +179,6 @@ public Map<String, TransformTranslator<?>> getTransformTranslators() {
       return ImmutableMap.<String, TransformTranslator<?>>builder()
           .put(PTransformTranslation.READ_TRANSFORM_URN, new ReadTranslator<>())
           .put(PTransformTranslation.RESHUFFLE_URN, new ReshuffleTranslator<>())
-          .put(PTransformTranslation.REDISTRIBUTE_BY_KEY_URN, new RedistributeByKeyTranslator<>())
           .put(PTransformTranslation.PAR_DO_TRANSFORM_URN, new ParDoBoundMultiTranslator<>())
           .put(PTransformTranslation.GROUP_BY_KEY_TRANSFORM_URN, new GroupByKeyTranslator<>())
           .put(PTransformTranslation.COMBINE_PER_KEY_TRANSFORM_URN, new GroupByKeyTranslator<>())