Merge pull request #1006: [proxima-beam-core] bump beam to 2.70.0

Author: je-ik

beam/core/src/main/java/cz/o2/proxima/beam/util/state/MethodCallUtils.java

@@ -86,9 +86,11 @@
 import org.apache.beam.sdk.util.ByteBuddyUtils;
 import org.apache.beam.sdk.util.CoderUtils;
 import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.OutputBuilder;
 import org.apache.beam.sdk.values.Row;
 import org.apache.beam.sdk.values.TimestampedValue;
 import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.WindowedValues;
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.joda.time.Duration;
 import org.joda.time.Instant;
@@ -339,6 +341,12 @@ private static <T> OutputReceiver<T> singleOutput(
       TupleTag<T> mainTag) {
 
     return new OutputReceiver<T>() {
+      @Override
+      public OutputBuilder<T> builder(T value) {
+        return WindowedValues.<T>builder()
+            .setReceiver(output -> multiOutput.get(mainTag).builder(value).output());
+      }
+
       @Override
       public void output(T output) {
         if (elem == null) {
@@ -734,6 +742,11 @@ public TimestampedOutputReceiver(OutputReceiver<T> parentReceiver, Instant times
       this.elementTimestamp = timestamp;
     }
 
+    @Override
+    public OutputBuilder<T> builder(T value) {
+      return parentReceiver.builder(value);
+    }
+
     @Override
     public void output(T output) {
       outputWithTimestamp(output, elementTimestamp);

beam/core/src/test/java/cz/o2/proxima/beam/core/direct/io/BatchLogReadTest.java

@@ -64,9 +64,11 @@
 import org.apache.beam.sdk.transforms.windowing.PaneInfo;
 import org.apache.beam.sdk.transforms.windowing.Repeatedly;
 import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.values.OutputBuilder;
 import org.apache.beam.sdk.values.PBegin;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.sdk.values.WindowedValues;
 import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -239,6 +241,12 @@ public void testInitialSplitting() {
     readFn.splitRestriction(
         list,
         new OutputReceiver<>() {
+          @Override
+          public OutputBuilder<PartitionList> builder(PartitionList value) {
+            return WindowedValues.<PartitionList>builder()
+                .setReceiver(output -> this.output(output.getValue()));
+          }
+
           @Override
           public void output(PartitionList part) {
             output.add(part);

buildSrc/src/main/groovy/cz.o2.proxima.java-conventions.gradle

@@ -24,7 +24,7 @@ plugins {
 java.sourceCompatibility = JavaVersion.VERSION_11
 
 def auto_service_version = "1.1.1"
-def beam_version = "2.68.0"
+def beam_version = "2.70.0"
 def errorprone_version = "2.18.0"
 def flink_version = "1.18.0"
 def grpc_version = "1.71.0"