You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by sc...@apache.org on 2024/08/20 07:44:03 UTC
(beam) branch master updated: Adds an ORDERED_LIST_STATE capability to the Java SDK. (#32067)
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 26cd5df5995 Adds an ORDERED_LIST_STATE capability to the Java SDK. (#32067)
26cd5df5995 is described below
commit 26cd5df5995bad8742e3495428f77da26c520ab8
Author: Andrew Crites <cr...@google.com>
AuthorDate: Tue Aug 20 00:43:57 2024 -0700
Adds an ORDERED_LIST_STATE capability to the Java SDK. (#32067)
---
CHANGES.md | 1 +
.../proto/org/apache/beam/model/pipeline/v1/beam_runner_api.proto | 4 ++++
.../main/java/org/apache/beam/sdk/util/construction/Environments.java | 1 +
.../java/org/apache/beam/sdk/util/construction/EnvironmentsTest.java | 3 +++
4 files changed, 9 insertions(+)
diff --git a/CHANGES.md b/CHANGES.md
index 1c0957b1fbf..78f243a2055 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -73,6 +73,7 @@
* Go SDK Minimum Go Version updated to 1.21 ([#32092](https://github.com/apache/beam/pull/32092)).
* [BigQueryIO] Added support for withFormatRecordOnFailureFunction() for STORAGE_WRITE_API and STORAGE_API_AT_LEAST_ONCE methods (Java) ([#31354](https://github.com/apache/beam/issues/31354)).
* Updated Go protobuf package to new version (Go) ([#21515](https://github.com/apache/beam/issues/21515)).
+* Adds OrderedListState support for Java SDK via FnApi.
## Breaking Changes
diff --git a/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/beam_runner_api.proto b/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/beam_runner_api.proto
index 422c2e1a5f7..10434e514ea 100644
--- a/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/beam_runner_api.proto
+++ b/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/beam_runner_api.proto
@@ -1681,6 +1681,10 @@ message StandardProtocols {
// the ProcessBundleProgressResponse messages.
SDK_CONSUMING_RECEIVED_DATA = 9
[(beam_urn) = "beam:protocol:sdk_consuming_received_data:v1"];
+
+ // Indicates whether the SDK supports ordered list state.
+ ORDERED_LIST_STATE = 10
+ [(beam_urn) = "beam:protocol:ordered_list_state:v1"];
}
}
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/Environments.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/Environments.java
index 9bdb45fe32e..52120b396e1 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/Environments.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/Environments.java
@@ -488,6 +488,7 @@ public class Environments {
capabilities.add(BeamUrns.getUrn(Primitives.TO_STRING));
capabilities.add(BeamUrns.getUrn(StandardProtocols.Enum.DATA_SAMPLING));
capabilities.add(BeamUrns.getUrn(StandardProtocols.Enum.SDK_CONSUMING_RECEIVED_DATA));
+ capabilities.add(BeamUrns.getUrn(StandardProtocols.Enum.ORDERED_LIST_STATE));
return capabilities.build();
}
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/EnvironmentsTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/EnvironmentsTest.java
index 93ec4f7e8f7..3c6862bd3b3 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/EnvironmentsTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/EnvironmentsTest.java
@@ -216,6 +216,9 @@ public class EnvironmentsTest implements Serializable {
assertThat(
Environments.getJavaCapabilities(),
hasItem(BeamUrns.getUrn(RunnerApi.StandardProtocols.Enum.DATA_SAMPLING)));
+ assertThat(
+ Environments.getJavaCapabilities(),
+ hasItem(BeamUrns.getUrn(RunnerApi.StandardProtocols.Enum.ORDERED_LIST_STATE)));
// Check that SDF truncation is supported
assertThat(
Environments.getJavaCapabilities(),