From c0bf13b881bdfd035494a38baf1861d58b4bb179 Mon Sep 17 00:00:00 2001 From: Tobias Kaymak Date: Mon, 12 Oct 2026 11:33:58 +0000 Subject: [PATCH 1/7] [Spark 4] Group the streaming stateful ParDo by encoded key bytes to support null keys A null Beam key became a null Spark grouping key and transformWithState failed with INTERNAL_ERROR_TWS. Grouping by the coder encoded key bytes is null safe and matches Beam key equality. Part of #36841. --- ...ostCommit_Java_ValidatesRunner_Spark4.json | 2 +- ...idatesRunner_SparkStructuredStreaming.json | 2 +- .../StatefulParDoStreamingTranslator.java | 10 +++--- .../state/BeamStatefulProcessor.java | 13 ++++--- .../streaming/StatefulParDoStreamingTest.java | 36 +++++++++++++++++++ 5 files changed, 52 insertions(+), 11 deletions(-) diff --git a/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark4.json b/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark4.json index c537844dc84a..e0266d62f2e0 100644 --- a/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark4.json +++ b/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Spark4.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run", - "modification": 3 + "modification": 4 } diff --git a/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming.json b/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming.json index bd93fd4e9881..c0d5a77072b2 100644 --- a/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming.json +++ b/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming.json @@ -12,5 +12,5 @@ "https://github.com/apache/beam/pull/40103": "noting that PR #40103 should run this test", "https://github.com/apache/beam/issues/40427": "Spark ValidatesRunner tests run serially", "https://github.com/apache/beam/pull/40281": "noting that PR #40281 should run this test", - "https://github.com/apache/beam/issues/36841": "Spark 4 streaming watermark lag" + "https://github.com/apache/beam/issues/36841": "Spark 4 streaming null keys" } diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/StatefulParDoStreamingTranslator.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/StatefulParDoStreamingTranslator.java index 1709ec570038..199570ae64a1 100644 --- a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/StatefulParDoStreamingTranslator.java +++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/StatefulParDoStreamingTranslator.java @@ -17,6 +17,7 @@ */ package org.apache.beam.runners.spark.structuredstreaming.translation.streaming; +import org.apache.beam.runners.spark.coders.CoderHelpers; import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsAccumulator; import org.apache.beam.runners.spark.structuredstreaming.translation.TransformTranslator; import org.apache.beam.runners.spark.structuredstreaming.translation.batch.DoFnRunnerFactory; @@ -35,7 +36,7 @@ import org.apache.beam.sdk.values.WindowingStrategy; import org.apache.spark.api.java.function.MapFunction; import org.apache.spark.sql.Dataset; -import org.apache.spark.sql.Encoder; +import org.apache.spark.sql.Encoders; import org.apache.spark.sql.streaming.OutputMode; import org.apache.spark.sql.streaming.TimeMode; @@ -57,7 +58,7 @@ protected void translate(ParDo.MultiOutput, OutputT> transform, Context DoFn, OutputT> doFn = transform.getFn(); KvCoder inputCoder = (KvCoder) input.getCoder(); - Encoder keyEnc = cxt.keyEncoderOf(inputCoder); + Coder keyCoder = inputCoder.getKeyCoder(); TupleTag mainOutputTag = transform.getMainOutputTag(); PCollection output = cxt.getOutput(mainOutputTag); Coder outputCoder = output.getCoder(); @@ -71,10 +72,11 @@ protected void translate(ParDo.MultiOutput, OutputT> transform, Context new BeamStatefulProcessor<>( doFn, runnerFactory, inputCoder, windowing, cxt.getOptionsSupplier(), metrics); - MapFunction>, K> keyFn = v -> v.getValue().getKey(); + MapFunction>, byte[]> keyFn = + v -> CoderHelpers.toByteArray(v.getValue().getKey(), keyCoder); Dataset> result = cxt.getDataset(input) - .groupByKey(keyFn, keyEnc) + .groupByKey(keyFn, Encoders.BINARY()) .transformWithState( processor, TimeMode.EventTime(), diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/state/BeamStatefulProcessor.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/state/BeamStatefulProcessor.java index e359ed6bd58b..2649f883af3f 100644 --- a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/state/BeamStatefulProcessor.java +++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/state/BeamStatefulProcessor.java @@ -41,6 +41,7 @@ import org.apache.beam.runners.spark.structuredstreaming.translation.batch.StatefulTaskRunner; import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop; import org.apache.beam.sdk.coders.Coder; +import org.apache.beam.sdk.coders.KvCoder; import org.apache.beam.sdk.coders.ListCoder; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.state.TimeDomain; @@ -68,7 +69,7 @@ /** Runs a stateful {@link DoFn} within Spark 4 Structured Streaming {@code transformWithState}. */ public class BeamStatefulProcessor - extends StatefulProcessor>, WindowedValue> { + extends StatefulProcessor>, WindowedValue> { private static final String BEAM_STATE = "beamState"; private static final String BEAM_TIMERS = "beamTimers"; @@ -76,7 +77,7 @@ public class BeamStatefulProcessor private final DoFn, OutputT> doFn; private final DoFnRunnerFactory, OutputT> runnerFactory; - private final Coder> inputCoder; + private final KvCoder inputCoder; private final WindowingStrategy windowingStrategy; private final Supplier optionsSupplier; private final MetricsAccumulator metrics; @@ -93,7 +94,7 @@ public class BeamStatefulProcessor public BeamStatefulProcessor( DoFn, OutputT> doFn, DoFnRunnerFactory, OutputT> runnerFactory, - Coder> inputCoder, + KvCoder inputCoder, WindowingStrategy windowingStrategy, Supplier optionsSupplier, MetricsAccumulator metrics) { @@ -138,7 +139,8 @@ private SparkTimerInternals timerInternals(TimerValues timerValues) { @Override public Iterator> handleInputRows( - K key, Iterator>> rows, TimerValues timerValues) { + byte[] keyBytes, Iterator>> rows, TimerValues timerValues) { + K key = CoderHelpers.fromByteArray(keyBytes, inputCoder.getKeyCoder()); List> outputs = new ArrayList<>(); SparkTimerInternals timerInternals = timerInternals(timerValues); @@ -159,7 +161,8 @@ public Iterator> handleInputRows( @Override public Iterator> handleExpiredTimer( - K key, TimerValues timerValues, ExpiredTimerInfo expiredTimerInfo) { + byte[] keyBytes, TimerValues timerValues, ExpiredTimerInfo expiredTimerInfo) { + K key = CoderHelpers.fromByteArray(keyBytes, inputCoder.getKeyCoder()); List> outputs = new ArrayList<>(); SparkTimerInternals timerInternals = timerInternals(timerValues); timerInternals.advanceWatermark(); diff --git a/runners/spark/4/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/StatefulParDoStreamingTest.java b/runners/spark/4/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/StatefulParDoStreamingTest.java index 2ff0bfc2bf05..2d3c127d72bc 100644 --- a/runners/spark/4/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/StatefulParDoStreamingTest.java +++ b/runners/spark/4/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/StatefulParDoStreamingTest.java @@ -33,6 +33,7 @@ import org.apache.beam.sdk.PipelineResult; import org.apache.beam.sdk.coders.KvCoder; import org.apache.beam.sdk.coders.StringUtf8Coder; +import org.apache.beam.sdk.coders.VoidCoder; import org.apache.beam.sdk.io.Read; import org.apache.beam.sdk.state.StateSpec; import org.apache.beam.sdk.state.StateSpecs; @@ -43,6 +44,7 @@ import org.apache.beam.sdk.state.ValueState; import org.apache.beam.sdk.transforms.DoFn; import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.transforms.WithKeys; import org.apache.beam.sdk.transforms.windowing.FixedWindows; import org.apache.beam.sdk.transforms.windowing.Window; import org.apache.beam.sdk.values.KV; @@ -69,12 +71,14 @@ public class StatefulParDoStreamingTest implements Serializable { private static final String TAG_BATCHES = "stateful-batches"; private static final String TAG_RESTART = "stateful-restart"; private static final String TAG_CHAINED = "stateful-chained"; + private static final String TAG_NULL_KEY = "stateful-null-key"; @After public void tearDown() { TestUnboundedSource.forget(TAG_BATCHES); TestUnboundedSource.forget(TAG_RESTART); TestUnboundedSource.forget(TAG_CHAINED); + TestUnboundedSource.forget(TAG_NULL_KEY); } private static Pipeline pipeline( @@ -167,6 +171,25 @@ public void testChainedStatefulParDoTimerOutputSurvives() throws Exception { assertTrue(collected.contains("timer-output")); } + @Test + public void testNullKey() throws Exception { + String collectorId = StreamingTestUtils.newCollectorId(TAG_NULL_KEY); + SparkStructuredStreamingPipelineOptions options = + StreamingTestUtils.streamingOptions(tempFolder); + options.setMaxRecordsPerBatch(1L); + Pipeline pipeline = Pipeline.create(options); + pipeline + .apply("Read", Read.from(new TestUnboundedSource(TAG_NULL_KEY, 1, 5))) + .apply("WithKeys", WithKeys.of((Void) null)) + .setCoder(KvCoder.of(VoidCoder.of(), StringUtf8Coder.of())) + .apply("StatefulParDo", ParDo.of(new NullKeyCountFn())) + .apply("Collect", ParDo.of(new StreamingTestUtils.CollectDoFn<>(collectorId))); + + assertEquals(PipelineResult.State.DONE, StreamingTestUtils.run(pipeline).getState()); + Set collected = StreamingTestUtils.collected(collectorId); + assertTrue(collected.contains(5)); + } + private static final class KeySelectorFn extends DoFn> { @ProcessElement public void process(@Element String element, OutputReceiver> out) { @@ -301,4 +324,17 @@ public void process( out.output(element.getValue()); } } + + private static final class NullKeyCountFn extends DoFn, Integer> { + @StateId("count") + private final StateSpec> countSpec = StateSpecs.value(); + + @ProcessElement + public void process( + @StateId("count") ValueState countState, OutputReceiver out) { + int count = (countState.read() == null ? 0 : countState.read()) + 1; + countState.write(count); + out.output(count); + } + } } From a2be98bcf3563cfb3319bd10581442e00d80f5d0 Mon Sep 17 00:00:00 2001 From: Tobias Kaymak Date: Mon, 12 Oct 2026 14:42:57 +0000 Subject: [PATCH 2/7] Retrigger CI, flaky StatefulStreamingParDoEvaluatorTest on Spark 3.4 From b8a876f7d4476abff1138e9d3091adb7cc47b7f5 Mon Sep 17 00:00:00 2001 From: Tobias Kaymak Date: Mon, 12 Oct 2026 18:56:23 +0000 Subject: [PATCH 3/7] Retrigger CI, flaky legacy Spark streaming tests in Spark Versions From b8b45ae42450af0ccd4132b4698a0974df9ed488 Mon Sep 17 00:00:00 2001 From: Tobias Kaymak Date: Tue, 13 Oct 2026 06:32:14 +0000 Subject: [PATCH 4/7] Retrigger CI, KafkaToPubsubE2ETest container startup flake in PreCommit Java From 5608646ba0678f1ec85d0662e94321f2ee66761e Mon Sep 17 00:00:00 2001 From: Tobias Kaymak Date: Fri, 16 Oct 2026 11:31:50 +0000 Subject: [PATCH 5/7] Use the structured streaming CoderHelpers in the stateful ParDo --- .../translation/streaming/StatefulParDoStreamingTranslator.java | 2 +- .../translation/streaming/state/BeamStatefulProcessor.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/StatefulParDoStreamingTranslator.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/StatefulParDoStreamingTranslator.java index 199570ae64a1..dd2186deafc5 100644 --- a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/StatefulParDoStreamingTranslator.java +++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/StatefulParDoStreamingTranslator.java @@ -17,11 +17,11 @@ */ package org.apache.beam.runners.spark.structuredstreaming.translation.streaming; -import org.apache.beam.runners.spark.coders.CoderHelpers; import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsAccumulator; import org.apache.beam.runners.spark.structuredstreaming.translation.TransformTranslator; import org.apache.beam.runners.spark.structuredstreaming.translation.batch.DoFnRunnerFactory; import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.SparkSideInputReader; +import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.CoderHelpers; import org.apache.beam.runners.spark.structuredstreaming.translation.streaming.state.BeamStatefulProcessor; import org.apache.beam.sdk.annotations.Internal; import org.apache.beam.sdk.coders.Coder; diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/state/BeamStatefulProcessor.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/state/BeamStatefulProcessor.java index 2649f883af3f..08c57fa15b1e 100644 --- a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/state/BeamStatefulProcessor.java +++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/state/BeamStatefulProcessor.java @@ -31,7 +31,6 @@ import org.apache.beam.runners.core.StatefulDoFnRunner; import org.apache.beam.runners.core.TimerInternals.TimerData; import org.apache.beam.runners.core.TimerInternals.TimerDataCoderV2; -import org.apache.beam.runners.spark.coders.CoderHelpers; import org.apache.beam.runners.spark.stateful.SparkStateInternals; import org.apache.beam.runners.spark.stateful.SparkStateInternals.StateCells; import org.apache.beam.runners.spark.stateful.SparkTimerInternals; @@ -39,6 +38,7 @@ import org.apache.beam.runners.spark.structuredstreaming.translation.batch.DoFnRunnerFactory; import org.apache.beam.runners.spark.structuredstreaming.translation.batch.DoFnRunnerFactory.DoFnRunnerWithTeardown; import org.apache.beam.runners.spark.structuredstreaming.translation.batch.StatefulTaskRunner; +import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.CoderHelpers; import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.KvCoder; From eb36e59da497cff45590c09404096237e22a8afc Mon Sep 17 00:00:00 2001 From: Tobias Kaymak Date: Mon, 19 Oct 2026 09:51:04 +0000 Subject: [PATCH 6/7] Merge the structured streaming CoderHelpers into the common one --- .../io/streaming/BeamReaderCache.java | 2 +- .../io/streaming/UnboundedSourceDataset.java | 2 +- .../batch/GroupByKeyTranslatorBatch.java | 2 +- .../translation/helpers/EncoderHelpers.java | 1 + .../streaming/ReadUnboundedTranslator.java | 2 +- .../StatefulParDoStreamingTranslator.java | 2 +- .../state/BeamStatefulProcessor.java | 2 +- .../batch/GroupByKeyTranslatorBatch.java | 2 +- .../batch/functions/SideInputValues.java | 2 +- .../translation/helpers/CoderHelpers.java | 59 ------------------- .../translation/helpers/EncoderHelpers.java | 1 + 11 files changed, 10 insertions(+), 67 deletions(-) delete mode 100644 runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/CoderHelpers.java diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamReaderCache.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamReaderCache.java index 1eb31e766e78..e961ff2f3807 100644 --- a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamReaderCache.java +++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamReaderCache.java @@ -27,7 +27,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.LongSupplier; -import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.CoderHelpers; +import org.apache.beam.runners.spark.coders.CoderHelpers; import org.apache.beam.sdk.io.UnboundedSource; import org.apache.beam.sdk.io.UnboundedSource.CheckpointMark; import org.apache.beam.sdk.io.UnboundedSource.UnboundedReader; diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/UnboundedSourceDataset.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/UnboundedSourceDataset.java index 0368f95ba6db..33106824c107 100644 --- a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/UnboundedSourceDataset.java +++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/UnboundedSourceDataset.java @@ -29,9 +29,9 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import org.apache.beam.runners.core.construction.SerializablePipelineOptions; +import org.apache.beam.runners.spark.coders.CoderHelpers; import org.apache.beam.runners.spark.structuredstreaming.SparkStructuredStreamingPipelineOptions; import org.apache.beam.runners.spark.structuredstreaming.io.streaming.BeamReaderCache.CachedReader; -import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.CoderHelpers; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.io.UnboundedSource; import org.apache.beam.sdk.io.UnboundedSource.CheckpointMark; diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTranslatorBatch.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTranslatorBatch.java index 7caf06cb38fd..2b930c579d00 100644 --- a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTranslatorBatch.java +++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTranslatorBatch.java @@ -18,13 +18,13 @@ package org.apache.beam.runners.spark.structuredstreaming.translation.batch; import static org.apache.beam.repackaged.core.org.apache.commons.lang3.ArrayUtils.EMPTY_BYTE_ARRAY; +import static org.apache.beam.runners.spark.coders.CoderHelpers.toByteArray; import static org.apache.beam.runners.spark.structuredstreaming.translation.batch.GroupByKeyHelpers.eligibleForGlobalGroupBy; import static org.apache.beam.runners.spark.structuredstreaming.translation.batch.GroupByKeyHelpers.eligibleForGroupByWindow; import static org.apache.beam.runners.spark.structuredstreaming.translation.batch.GroupByKeyHelpers.explodeWindowedKey; import static org.apache.beam.runners.spark.structuredstreaming.translation.batch.GroupByKeyHelpers.valueKey; import static org.apache.beam.runners.spark.structuredstreaming.translation.batch.GroupByKeyHelpers.valueValue; import static org.apache.beam.runners.spark.structuredstreaming.translation.batch.GroupByKeyHelpers.windowedKV; -import static org.apache.beam.runners.spark.structuredstreaming.translation.helpers.CoderHelpers.toByteArray; import static org.apache.beam.runners.spark.structuredstreaming.translation.helpers.EncoderHelpers.collectionEncoder; import static org.apache.beam.runners.spark.structuredstreaming.translation.helpers.EncoderHelpers.encoderOf; import static org.apache.beam.runners.spark.structuredstreaming.translation.helpers.EncoderHelpers.kvEncoder; diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java index 173b4653a19b..4cd92da0a15f 100644 --- a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java +++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java @@ -36,6 +36,7 @@ import java.util.Set; import java.util.TreeMap; import java.util.concurrent.ConcurrentHashMap; +import org.apache.beam.runners.spark.coders.CoderHelpers; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.transforms.windowing.GlobalWindow; diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/ReadUnboundedTranslator.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/ReadUnboundedTranslator.java index fb1691108b71..eb1421b45540 100644 --- a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/ReadUnboundedTranslator.java +++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/ReadUnboundedTranslator.java @@ -19,10 +19,10 @@ import static org.apache.spark.sql.functions.col; +import org.apache.beam.runners.spark.coders.CoderHelpers; import org.apache.beam.runners.spark.structuredstreaming.SparkStructuredStreamingPipelineOptions; import org.apache.beam.runners.spark.structuredstreaming.io.streaming.UnboundedSourceDataset; import org.apache.beam.runners.spark.structuredstreaming.translation.TransformTranslator; -import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.CoderHelpers; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.io.UnboundedSource; import org.apache.beam.sdk.transforms.windowing.GlobalWindow; diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/StatefulParDoStreamingTranslator.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/StatefulParDoStreamingTranslator.java index dd2186deafc5..199570ae64a1 100644 --- a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/StatefulParDoStreamingTranslator.java +++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/StatefulParDoStreamingTranslator.java @@ -17,11 +17,11 @@ */ package org.apache.beam.runners.spark.structuredstreaming.translation.streaming; +import org.apache.beam.runners.spark.coders.CoderHelpers; import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsAccumulator; import org.apache.beam.runners.spark.structuredstreaming.translation.TransformTranslator; import org.apache.beam.runners.spark.structuredstreaming.translation.batch.DoFnRunnerFactory; import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.SparkSideInputReader; -import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.CoderHelpers; import org.apache.beam.runners.spark.structuredstreaming.translation.streaming.state.BeamStatefulProcessor; import org.apache.beam.sdk.annotations.Internal; import org.apache.beam.sdk.coders.Coder; diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/state/BeamStatefulProcessor.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/state/BeamStatefulProcessor.java index 08c57fa15b1e..2649f883af3f 100644 --- a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/state/BeamStatefulProcessor.java +++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/state/BeamStatefulProcessor.java @@ -31,6 +31,7 @@ import org.apache.beam.runners.core.StatefulDoFnRunner; import org.apache.beam.runners.core.TimerInternals.TimerData; import org.apache.beam.runners.core.TimerInternals.TimerDataCoderV2; +import org.apache.beam.runners.spark.coders.CoderHelpers; import org.apache.beam.runners.spark.stateful.SparkStateInternals; import org.apache.beam.runners.spark.stateful.SparkStateInternals.StateCells; import org.apache.beam.runners.spark.stateful.SparkTimerInternals; @@ -38,7 +39,6 @@ import org.apache.beam.runners.spark.structuredstreaming.translation.batch.DoFnRunnerFactory; import org.apache.beam.runners.spark.structuredstreaming.translation.batch.DoFnRunnerFactory.DoFnRunnerWithTeardown; import org.apache.beam.runners.spark.structuredstreaming.translation.batch.StatefulTaskRunner; -import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.CoderHelpers; import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.KvCoder; diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTranslatorBatch.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTranslatorBatch.java index c55781ff84a8..c67ea3ad3e5b 100644 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTranslatorBatch.java +++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTranslatorBatch.java @@ -18,13 +18,13 @@ package org.apache.beam.runners.spark.structuredstreaming.translation.batch; import static org.apache.beam.repackaged.core.org.apache.commons.lang3.ArrayUtils.EMPTY_BYTE_ARRAY; +import static org.apache.beam.runners.spark.coders.CoderHelpers.toByteArray; import static org.apache.beam.runners.spark.structuredstreaming.translation.batch.GroupByKeyHelpers.eligibleForGlobalGroupBy; import static org.apache.beam.runners.spark.structuredstreaming.translation.batch.GroupByKeyHelpers.eligibleForGroupByWindow; import static org.apache.beam.runners.spark.structuredstreaming.translation.batch.GroupByKeyHelpers.explodeWindowedKey; import static org.apache.beam.runners.spark.structuredstreaming.translation.batch.GroupByKeyHelpers.valueKey; import static org.apache.beam.runners.spark.structuredstreaming.translation.batch.GroupByKeyHelpers.valueValue; import static org.apache.beam.runners.spark.structuredstreaming.translation.batch.GroupByKeyHelpers.windowedKV; -import static org.apache.beam.runners.spark.structuredstreaming.translation.helpers.CoderHelpers.toByteArray; import static org.apache.beam.runners.spark.structuredstreaming.translation.helpers.EncoderHelpers.collectionEncoder; import static org.apache.beam.runners.spark.structuredstreaming.translation.helpers.EncoderHelpers.encoderOf; import static org.apache.beam.runners.spark.structuredstreaming.translation.helpers.EncoderHelpers.kvEncoder; diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/functions/SideInputValues.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/functions/SideInputValues.java index 23c8d49c3091..0017f95b2a05 100644 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/functions/SideInputValues.java +++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/functions/SideInputValues.java @@ -34,8 +34,8 @@ import java.util.Map; import java.util.function.Function; import javax.annotation.Nullable; +import org.apache.beam.runners.spark.coders.CoderHelpers; import org.apache.beam.runners.spark.structuredstreaming.translation.EvaluationContext; -import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.CoderHelpers; import org.apache.beam.sdk.annotations.Internal; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/CoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/CoderHelpers.java deleted file mode 100644 index f8c63bc34f14..000000000000 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/CoderHelpers.java +++ /dev/null @@ -1,59 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.beam.runners.spark.structuredstreaming.translation.helpers; - -import java.io.IOException; -import org.apache.beam.sdk.coders.Coder; -import org.apache.beam.sdk.util.CoderUtils; - -/** Serialization utility class. */ -public final class CoderHelpers { - private CoderHelpers() {} - - /** - * Utility method for serializing an object using the specified coder. - * - * @param value Value to serialize. - * @param coder Coder to serialize with. - * @param type of value that is serialized - * @return Byte array representing serialized object. - */ - public static byte[] toByteArray(T value, Coder coder) { - try { - return CoderUtils.encodeToByteArray(coder, value); - } catch (IOException e) { - throw new IllegalStateException("Error encoding value: " + value, e); - } - } - - /** - * Utility method for deserializing a byte array using the specified coder. - * - * @param serialized bytearray to be deserialized. - * @param coder Coder to deserialize with. - * @param Type of object to be returned. - * @return Deserialized object. - */ - public static T fromByteArray(byte[] serialized, Coder coder) { - try { - return CoderUtils.decodeFromByteArray(coder, serialized); - } catch (IOException e) { - throw new IllegalStateException("Error decoding bytes for coder: " + coder, e); - } - } -} diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java index 29f01c84c02e..08ce47c67bb8 100644 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java +++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java @@ -36,6 +36,7 @@ import java.util.Set; import java.util.TreeMap; import java.util.concurrent.ConcurrentHashMap; +import org.apache.beam.runners.spark.coders.CoderHelpers; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.transforms.windowing.GlobalWindow; From 5980f5633596c2458f13fd81cedd3df2388df41e Mon Sep 17 00:00:00 2001 From: Tobias Kaymak Date: Mon, 19 Oct 2026 10:19:57 +0000 Subject: [PATCH 7/7] Decode with the common CoderHelpers in BeamMicroBatchSourceTest --- .../io/streaming/BeamMicroBatchSourceTest.java | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/runners/spark/4/src/test/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamMicroBatchSourceTest.java b/runners/spark/4/src/test/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamMicroBatchSourceTest.java index 3b2e36bf5ef7..559deb808036 100644 --- a/runners/spark/4/src/test/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamMicroBatchSourceTest.java +++ b/runners/spark/4/src/test/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamMicroBatchSourceTest.java @@ -46,6 +46,7 @@ import javax.annotation.Nullable; import org.apache.beam.runners.core.construction.SerializablePipelineOptions; import org.apache.beam.runners.spark.StreamingTest; +import org.apache.beam.runners.spark.coders.CoderHelpers; import org.apache.beam.runners.spark.structuredstreaming.SparkSessionRule; import org.apache.beam.runners.spark.structuredstreaming.SparkStructuredStreamingPipelineOptions; import org.apache.beam.runners.spark.structuredstreaming.io.streaming.UnboundedSourceDataset.BeamInputPartition; @@ -60,7 +61,6 @@ import org.apache.beam.sdk.io.UnboundedSource; import org.apache.beam.sdk.options.PipelineOptionsFactory; import org.apache.beam.sdk.transforms.windowing.GlobalWindow; -import org.apache.beam.sdk.util.CoderUtils; import org.apache.beam.sdk.values.WindowedValue; import org.apache.beam.sdk.values.WindowedValues; import org.apache.hadoop.conf.Configuration; @@ -209,7 +209,7 @@ public void testReadsElementsFromUnboundedSource() throws Exception { List values = new ArrayList<>(); for (List batch : batches(tag)) { for (Row row : batch) { - WindowedValue value = CoderUtils.decodeFromByteArray(CODER, row.getAs(COL_PAYLOAD)); + WindowedValue value = CoderHelpers.fromByteArray(row.getAs(COL_PAYLOAD), CODER); values.add(value.getValue()); assertEquals( value.getTimestamp().getMillis(), row.getAs(COL_EVENT_TS).getTime()); @@ -593,11 +593,7 @@ private static List values(List> batches) { } private static String decode(byte[] payload) { - try { - return CoderUtils.decodeFromByteArray(CODER, payload).getValue(); - } catch (IOException e) { - throw new IllegalStateException(e); - } + return CoderHelpers.fromByteArray(payload, CODER).getValue(); } private static List nonEmptySizes(List> batches) { @@ -748,7 +744,7 @@ private static int committedPosition(File checkpointDir, int shard) throws IOExc new BeamSourceCheckpoint(sourceDir(checkpointDir).getAbsolutePath(), new Configuration()); byte[] coded = checkpoint.readMark(shard, epoch); assertNotNull("no mark at committed epoch " + epoch + " for shard " + shard, coded); - return CoderUtils.decodeFromByteArray(TestUnboundedSource.MARK_CODER, coded).next; + return CoderHelpers.fromByteArray(coded, TestUnboundedSource.MARK_CODER).next; } // ---------------------------------------------------------------------------------------------