Merge branch 'master' into skip-spannertests

Author: claudevdm

.github/trigger_files/beam_PostCommit_Python.json

@@ -1,5 +1,6 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run.",
-  "modification": 34
+  "pr": "36271",
+  "modification": 35
 }
 

.github/trigger_files/beam_PostCommit_Python_Xlang_IO_Dataflow.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 5
+  "modification": 0
 }

CHANGES.md

@@ -85,6 +85,7 @@
 ## Bugfixes
 
 * Fixed FirestoreV1 Beam connectors allow configuring inconsistent project/database IDs between RPC requests and routing headers #36895 (Java) ([#36895](https://github.com/apache/beam/issues/36895)).
+ Logical type and coder registry are saved for pipelines in the case of default pickler. This fixes a side effect of switching to cloudpickle as default pickler in Beam 2.65.0 (Python) ([#35738](https://github.com/apache/beam/issues/35738)).
 
 ## Known Issues
 

build.gradle.kts

@@ -322,6 +322,8 @@ tasks.register("javaPreCommit") {
   dependsOn(":sdks:java:io:xml:build")
   dependsOn(":sdks:java:javadoc:allJavadoc")
   dependsOn(":sdks:java:managed:build")
+  dependsOn("sdks:java:ml:inference:remote:build")
+  dependsOn("sdks:java:ml:inference:openai:build")
   dependsOn(":sdks:java:testing:expansion-service:build")
   dependsOn(":sdks:java:testing:jpms-tests:build")
   dependsOn(":sdks:java:testing:junit:build")

buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy

@@ -612,8 +612,8 @@ class BeamModulePlugin implements Plugin<Project> {
     def google_ads_version = "33.0.0"
     def google_clients_version = "2.0.0"
     def google_cloud_bigdataoss_version = "2.2.26"
-    // [bomupgrader] TODO(#35868): currently pinned, should be determined by: com.google.cloud:google-cloud-spanner, consistent with: google_cloud_platform_libraries_bom
-    def google_cloud_spanner_version = "6.95.1"
+    // [bomupgrader] TODO(#37008): currently pinned before gcp-bom moving beyond this version, should be determined by: com.google.cloud:google-cloud-spanner, consistent with: google_cloud_platform_libraries_bom
+    def google_cloud_spanner_version = "6.104.0"
     def google_code_gson_version = "2.10.1"
     def google_oauth_clients_version = "1.34.1"
     // [bomupgrader] determined by: io.grpc:grpc-netty, consistent with: google_cloud_platform_libraries_bom

model/fn-execution/src/main/proto/org/apache/beam/model/fn_execution/v1/beam_fn_api.proto

@@ -761,6 +761,15 @@ message Elements {
   // extensible and backward compatible
   message ElementMetadata {
     optional DrainMode.Enum drain = 1;
+    // (Optional) As part of https://www.w3.org/TR/trace-context/ we are forwarding a trace and participating in it.
+    // Traceparent header represents the incoming request in a tracing system in a common format.
+    // Example value: 00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01
+    optional string traceparent = 2;
+    // (Optional) tracestate extends traceparent with open telemetry data represented by a set of name/value pairs.
+    // Format specified https://www.w3.org/TR/trace-context/#list for interoperability and commonly used
+    // across IOs - Kafka, PubSub, http.
+    // Example value: congo=t61rcWkgMzE
+    optional string tracestate = 3;
   }
 
   // Represent the encoded user timer for a given instruction, transform and

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillSink.java

@@ -264,11 +264,15 @@ public long add(WindowedValue<T> data) throws IOException {
         }
       }
 
-      Windmill.KeyedMessageBundle.Builder keyedOutput = productionMap.get(key);
-      if (keyedOutput == null) {
-        keyedOutput = Windmill.KeyedMessageBundle.newBuilder().setKey(key);
-        productionMap.put(key, keyedOutput);
-      }
+      Windmill.KeyedMessageBundle.Builder keyedOutput =
+          productionMap.computeIfAbsent(
+              key,
+              (k) -> {
+                Windmill.KeyedMessageBundle.Builder builder =
+                    Windmill.KeyedMessageBundle.newBuilder();
+                builder.setKey(k);
+                return builder;
+              });
 
       try {
         messageBuilder

sdks/java/core/src/test/java/org/apache/beam/sdk/fn/splittabledofn/RestrictionTrackersTest.java

@@ -102,7 +102,7 @@ private static class RestrictionTrackerWithProgress extends RestrictionTracker<O
       implements HasProgress {
     private boolean blockTryClaim;
     private boolean blockTrySplit;
-    private boolean isBlocked;
+    private volatile boolean isBlocked;
     public static final Progress REPORT_PROGRESS = Progress.from(2.0, 3.0);
 
     public RestrictionTrackerWithProgress() {

sdks/java/extensions/ml/build.gradle

@@ -26,6 +26,7 @@ applyJavaNature(
 )
 
 description = 'Apache Beam :: SDKs :: Java :: Extensions :: ML'
+ext.summary = """beam-sdks-java-extensions-ml provides Apache Beam Java SDK machine learning integration with Google Cloud AI Video Intelligence service. For machine learning run inference modules, see beam-sdks-java-ml-reference-* artifacts."""
 
 dependencies {
     implementation project(path: ":sdks:java:core", configuration: "shadow")

sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/SpannerChangeStreamErrorTest.java

@@ -70,6 +70,7 @@
 import org.joda.time.Duration;
 import org.junit.After;
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
@@ -115,6 +116,7 @@ public void tearDown() throws NoSuchFieldException, IllegalAccessException {
   }
 
   @Test
+  @Ignore("https://github.com/apache/beam/issues/37002 Re-enable skipped tests.")
   // Error code UNAVAILABLE is retried repeatedly until the RPC times out.
   public void testUnavailableExceptionRetries() throws InterruptedException {
     DirectOptions options = PipelineOptionsFactory.as(DirectOptions.class);
@@ -155,6 +157,7 @@ public void testUnavailableExceptionRetries() throws InterruptedException {
   }
 
   @Test
+  @Ignore("https://github.com/apache/beam/issues/37002 Re-enable skipped tests.")
   // Error code ABORTED is retried repeatedly until it times out.
   public void testAbortedExceptionRetries() throws InterruptedException {
     mockSpannerService.setExecuteStreamingSqlExecutionTime(
@@ -218,6 +221,7 @@ public void testUnknownExceptionDoesNotRetry() {
   }
 
   @Test
+  @Ignore("https://github.com/apache/beam/issues/37002 Re-enable skipped tests.")
   // Error code RESOURCE_EXHAUSTED is retried repeatedly.
   public void testResourceExhaustedRetry() {
     mockSpannerService.setExecuteStreamingSqlExecutionTime(
@@ -281,6 +285,7 @@ public void testResourceExhaustedRetryWithDefaultSettings() {
   }
 
   @Test
+  @Ignore("https://github.com/apache/beam/issues/37002 Re-enable skipped tests.")
   public void testInvalidRecordReceived() {
     final Timestamp startTimestamp = Timestamp.ofTimeSecondsAndNanos(0, 1000);
     final Timestamp endTimestamp =