This is an automated email from the ASF dual-hosted git repository.
scwhittle pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new fe27feb2ab5 [Dataflow Streaming] Enable state tag encoding v2 (#38705)
fe27feb2ab5 is described below
commit fe27feb2ab5fa2af9330747fb0fb86d3c40a071c
Author: Arun Pandian <[email protected]>
AuthorDate: Mon Jun 1 02:04:18 2026 -0700
[Dataflow Streaming] Enable state tag encoding v2 (#38705)
* Enable state tag encoding v2 by default for new Dataflow Streaming Engine
jobs
Also added CHANGES.md entry detailing the change, how to disable it, and
job update/downgrade limitations.
* Respect UpdateCompatibility for tag encoding v2
---
CHANGES.md | 2 +-
.../beam/runners/dataflow/DataflowRunner.java | 6 +++
.../beam/runners/dataflow/DataflowRunnerTest.java | 44 ++++++++++++++++++++++
3 files changed, 51 insertions(+), 1 deletion(-)
diff --git a/CHANGES.md b/CHANGES.md
index 3b4af42daf8..669cabd7c5a 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -69,7 +69,7 @@
## New Features / Improvements
-* X feature added (Java/Python)
([#X](https://github.com/apache/beam/issues/X)).
+* (Java) Enabled state tag encoding v2 by default for new Dataflow Streaming
Engine jobs. It can be disabled by passing
`--experiments=disable_streaming_engine_state_tag_encoding_v2` or
`--updateCompatibilityVersion=2.74.0` pipeline option. Note that the tag
encoding version cannot change during a job update. Jobs using tag encoding v2
(enabled by default for new jobs on 2.75.0+) cannot be downgraded to Beam
versions prior to 2.73.0, as only versions 2.73.0 and later support tag
encoding [...]
## Breaking Changes
diff --git
a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
index d04afb351e4..4864f2cf453 100644
---
a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
+++
b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
@@ -101,6 +101,7 @@ import org.apache.beam.sdk.options.ExperimentalOptions;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsValidator;
import org.apache.beam.sdk.options.SdkHarnessOptions;
+import org.apache.beam.sdk.options.StreamingOptions;
import org.apache.beam.sdk.options.ValueProvider.NestedValueProvider;
import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.runners.PTransformOverride;
@@ -1310,6 +1311,11 @@ public class DataflowRunner extends
PipelineRunner<DataflowPipelineJob> {
// Experiment marking that the harness supports tag encoding v2
// Backend will enable tag encoding v2 only if the harness supports it.
experiments.add("streaming_engine_state_tag_encoding_v2_supported");
+ // Experiment requesting tag encoding v2 on new jobs starting with
2.75.0. During job
+ // updates old job's tag encoding version is carried over by the
backend.
+ if (!StreamingOptions.updateCompatibilityVersionLessThan(options,
"2.75.0")) {
+ experiments.add("enable_streaming_engine_state_tag_encoding_v2");
+ }
options.setExperiments(ImmutableList.copyOf(experiments));
}
diff --git
a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java
b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java
index 122c4aeee34..073b30f928d 100644
---
a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java
+++
b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java
@@ -2923,4 +2923,48 @@ public class DataflowRunnerTest implements Serializable {
PAssert.that(output).containsInAnyOrder("value:UPDATE_BEFORE");
pipeline.run();
}
+
+ @Test
+ public void testStreamingStateTagEncodingV2PreCompatibility() throws
Exception {
+ DataflowPipelineOptions options = buildPipelineOptions();
+ options.as(StreamingOptions.class).setStreaming(true);
+ options.as(StreamingOptions.class).setUpdateCompatibilityVersion("2.74.0");
+ Pipeline p = Pipeline.create(options);
+
+ p.run();
+
+ List<String> experiments = options.getExperiments();
+ assertNotNull(experiments);
+
assertTrue(experiments.contains("streaming_engine_state_tag_encoding_v2_supported"));
+
assertFalse(experiments.contains("enable_streaming_engine_state_tag_encoding_v2"));
+ }
+
+ @Test
+ public void testStreamingStateTagEncodingV2PostCompatibility() throws
Exception {
+ DataflowPipelineOptions options = buildPipelineOptions();
+ options.as(StreamingOptions.class).setStreaming(true);
+ options.as(StreamingOptions.class).setUpdateCompatibilityVersion("2.75.0");
+ Pipeline p = Pipeline.create(options);
+
+ p.run();
+
+ List<String> experiments = options.getExperiments();
+ assertNotNull(experiments);
+
assertTrue(experiments.contains("streaming_engine_state_tag_encoding_v2_supported"));
+
assertTrue(experiments.contains("enable_streaming_engine_state_tag_encoding_v2"));
+ }
+
+ @Test
+ public void testStreamingStateTagEncodingV2NoCompatibility() throws
Exception {
+ DataflowPipelineOptions options = buildPipelineOptions();
+ options.as(StreamingOptions.class).setStreaming(true);
+ Pipeline p = Pipeline.create(options);
+
+ p.run();
+
+ List<String> experiments = options.getExperiments();
+ assertNotNull(experiments);
+
assertTrue(experiments.contains("streaming_engine_state_tag_encoding_v2_supported"));
+
assertTrue(experiments.contains("enable_streaming_engine_state_tag_encoding_v2"));
+ }
}
