[pull] master from apache:master

.github/trigger_files/beam_PostCommit_Python_Xlang_Messaging_Direct.json

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

.github/workflows/README.md

@@ -534,6 +534,7 @@ PostCommit Jobs run in a schedule against master branch and generally do not get
 | [ Publish Beam SDK Snapshots ](https://github.com/apache/beam/actions/workflows/beam_Publish_Beam_SDK_Snapshots.yml) | N/A | [![.github/workflows/beam_Publish_Beam_SDK_Snapshots.yml](https://github.com/apache/beam/actions/workflows/beam_Publish_Beam_SDK_Snapshots.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_Publish_Beam_SDK_Snapshots.yml?query=event%3Aschedule) |
 | [ Publish BeamMetrics ](https://github.com/apache/beam/actions/workflows/beam_Publish_BeamMetrics.yml) | N/A | [![.github/workflows/beam_Publish_BeamMetrics.yml](https://github.com/apache/beam/actions/workflows/beam_Publish_BeamMetrics.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_Publish_BeamMetrics.yml?query=event%3Aschedule)
 | [ Publish Docker Snapshots ](https://github.com/apache/beam/actions/workflows/beam_Publish_Docker_Snapshots.yml) | N/A | [![.github/workflows/beam_Publish_Docker_Snapshots.yml](https://github.com/apache/beam/actions/workflows/beam_Publish_Docker_Snapshots.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_Publish_Docker_Snapshots.yml?query=event%3Aschedule) |
+| [ Publish Python VLLM Image ](https://github.com/apache/beam/actions/workflows/beam_Publish_Python_VLLM_Image.yml) | N/A | [![.github/workflows/beam_Publish_Python_VLLM_Image.yml](https://github.com/apache/beam/actions/workflows/beam_Publish_Python_VLLM_Image.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_Publish_Python_VLLM_Image.yml?query=event%3Aschedule) |
 | [ Publish Website  ](https://github.com/apache/beam/actions/workflows/beam_Publish_Website.yml) | N/A | [![.github/workflows/beam_Publish_Website.yml](https://github.com/apache/beam/actions/workflows/beam_Publish_Website.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_Publish_Website.yml?query=event%3Aschedule) |
 | [ Release Nightly Snapshot ](https://github.com/apache/beam/actions/workflows/beam_Release_NightlySnapshot.yml) | N/A | [![.github/workflows/beam_Release_NightlySnapshot.yml](https://github.com/apache/beam/actions/workflows/beam_Release_NightlySnapshot.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_Release_NightlySnapshot.yml?query=event%3Aschedule) |
 | [ Release Nightly Snapshot Python ](https://github.com/apache/beam/actions/workflows/beam_Release_Python_NightlySnapshot.yml) | N/A | [![.github/workflows/beam_Release_Python_NightlySnapshot.yml](https://github.com/apache/beam/actions/workflows/beam_Release_Python_NightlySnapshot.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_Release_Python_NightlySnapshot.yml?query=event%3Aschedule) |

.github/workflows/beam_PostCommit_Python_Xlang_Messaging_Direct.yml

@@ -63,7 +63,7 @@ jobs:
         job_name: ["beam_PostCommit_Python_Xlang_Messaging_Direct"]
         job_phrase: ["Run Python_Xlang_Messaging_Direct PostCommit"]
     steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@v7
       - name: Setup repository
         uses: ./.github/actions/setup-action
         with:

.github/workflows/beam_Publish_Python_VLLM_Image.yml

@@ -0,0 +1,75 @@
+# 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.
+
+name: Publish Python VLLM Image
+
+on:
+  schedule:
+    - cron: '0 0 * * 0' # Run weekly (Sunday at 00:00)
+  workflow_dispatch:
+
+# Setting explicit permissions for the action to avoid the default permissions
+permissions:
+  actions: write
+  pull-requests: read
+  checks: read
+  contents: read
+  deployments: read
+  id-token: none
+  issues: read
+  discussions: read
+  packages: read
+  pages: read
+  repository-projects: read
+  security-events: read
+  statuses: read
+
+# This allows a subsequently queued workflow run to interrupt previous runs
+concurrency:
+  group: '${{ github.workflow }} @ ${{ github.event.issue.number || github.sha || github.head_ref || github.ref }}-${{ github.event.schedule || github.event.comment.id || github.event.sender.login }}'
+  cancel-in-progress: true
+
+jobs:
+  build_and_push_image:
+    if: |
+      github.event_name == 'workflow_dispatch' ||
+      (github.event_name == 'schedule' && github.repository == 'apache/beam')
+    runs-on: [self-hosted, ubuntu-24.04, main]
+    timeout-minutes: 60
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v7
+      - name: Authenticate on GCP
+        uses: google-github-actions/auth@7c6bc770dae815cd3e89ee6cdf493a5fab2cc093
+        with:
+          service_account: ${{ secrets.GCP_SA_EMAIL }}
+          credentials_json: ${{ secrets.GCP_SA_KEY }}
+      - name: Set up Cloud SDK
+        uses: google-github-actions/setup-gcloud@aa5489c8933f4cc7a4f7d45035b3b1440c9c10db
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@d7f5e7f509e45cec5c76c4d5afdd7de93d0b3df5
+      - name: GCloud Docker credential helper
+        run: |
+          gcloud auth configure-docker us.gcr.io
+      - name: Build and Push Multi-Arch Image
+        run: |
+          docker buildx build \
+            --platform linux/amd64,linux/arm64 \
+            -t us.gcr.io/apache-beam-testing/python-postcommit-it/vllm:latest \
+            -f sdks/python/apache_beam/ml/inference/test_resources/vllm.dockerfile.old \
+            --push \
+            .

.github/workflows/playground_frontend_test.yml

@@ -48,7 +48,7 @@ jobs:
       - uses: actions/checkout@v7
 
       - name: 'Cache Flutter Dependencies'
-        uses: actions/cache@v4
+        uses: actions/cache@v6
         with:
           path: /opt/hostedtoolcache/flutter
           key: ${{ runner.OS }}-flutter-install-cache-${{ env.FLUTTER_VERSION }}

.github/workflows/tour_of_beam_frontend_test.yml

@@ -50,7 +50,7 @@ jobs:
       - uses: actions/checkout@v7
 
       - name: 'Cache Flutter Dependencies'
-        uses: actions/cache@v4
+        uses: actions/cache@v6
         with:
           path: /opt/hostedtoolcache/flutter
           key: ${{ runner.OS }}-flutter-install-cache-${{ env.FLUTTER_VERSION }}

runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowStreamingPipelineOptions.java

@@ -196,6 +196,13 @@ public interface DataflowStreamingPipelineOptions extends PipelineOptions {
 
   void setStuckCommitDurationMillis(int value);
 
+  @Description(
+      "Retry commits on stream errors until this much time has elapsed since the commit was scheduled. If zero, retry forever.")
+  @Default.InstanceFactory(CommitWorkStreamRetryTimeoutMillisFactory.class)
+  long getCommitWorkStreamRetryTimeoutMillis();
+
+  void setCommitWorkStreamRetryTimeoutMillis(long value);
+
   @Description(
       "Period for sending 'global get config' requests to the service. The duration is "
           + "specified as seconds in 'PTx.yS' format, e.g. 'PT5.125S'."
@@ -343,4 +350,14 @@ public Boolean create(PipelineOptions options) {
       return ExperimentalOptions.hasExperiment(options, "enable_windmill_service_direct_path");
     }
   }
+
+  class CommitWorkStreamRetryTimeoutMillisFactory implements DefaultValueFactory<Long> {
+    @Override
+    public Long create(PipelineOptions options) {
+      if (ExperimentalOptions.hasExperiment(options, "disable_commit_retry_timeout")) {
+        return 0L;
+      }
+      return Duration.standardMinutes(30).getMillis();
+    }
+  }
 }

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

@@ -756,6 +756,8 @@ public static StreamingDataflowWorker fromOptions(DataflowWorkerHarnessOptions o
                   new WorkHeartbeatResponseProcessor(computationStateCache::get))
               .setHealthCheckIntervalMillis(
                   options.getWindmillServiceStreamingRpcHealthCheckPeriodMs())
+              .setCommitWorkStreamRetryTimeout(
+                  java.time.Duration.ofMillis(options.getCommitWorkStreamRetryTimeoutMillis()))
               .build();
       return ConfigFetcherComputationStateCacheAndWindmillClient.builder()
           .setWindmillDispatcherClient(dispatcherClient)

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java

@@ -20,7 +20,6 @@
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
 
-import com.google.auto.value.AutoValue;
 import java.io.PrintWriter;
 import java.time.Duration;
 import java.util.HashMap;
@@ -77,6 +76,7 @@ private static class StreamAndRequest {
   private final JobHeader jobHeader;
   private final int streamingRpcBatchLimit;
   private volatile boolean logMissingResponse = true;
+  private final Duration maxRetryDuration;
 
   private GrpcCommitWorkStream(
       String backendWorkerToken,
@@ -90,6 +90,7 @@ private GrpcCommitWorkStream(
       AtomicLong idGenerator,
       int streamingRpcBatchLimit,
       Duration halfClosePhysicalStreamAfter,
+      Duration maxRetryDuration,
       ScheduledExecutorService executor) {
     super(
         LOG,
@@ -104,6 +105,7 @@ private GrpcCommitWorkStream(
     this.idGenerator = idGenerator;
     this.jobHeader = jobHeader;
     this.streamingRpcBatchLimit = streamingRpcBatchLimit;
+    this.maxRetryDuration = maxRetryDuration;
   }
 
   static GrpcCommitWorkStream create(
@@ -118,6 +120,7 @@ static GrpcCommitWorkStream create(
       AtomicLong idGenerator,
       int streamingRpcBatchLimit,
       Duration halfClosePhysicalStreamAfter,
+      Duration maxRetryDuration,
       ScheduledExecutorService executor) {
     return new GrpcCommitWorkStream(
         backendWorkerToken,
@@ -130,6 +133,7 @@ static GrpcCommitWorkStream create(
         idGenerator,
         streamingRpcBatchLimit,
         halfClosePhysicalStreamAfter,
+        maxRetryDuration,
         executor);
   }
 
@@ -224,14 +228,48 @@ public void onResponse(StreamingCommitResponse response) {
       failureHandler.throwIfNonEmpty();
     }
 
-    @Override
     @SuppressWarnings("ReferenceEquality")
+    private boolean belongsToThisHandler(StreamAndRequest streamAndRequest) {
+      return streamAndRequest.handler == this;
+    }
+
+    @Override
     public boolean hasPendingRequests() {
-      return pending.entrySet().stream().anyMatch(e -> e.getValue().handler == this);
+      return pending.entrySet().stream().anyMatch(e -> belongsToThisHandler(e.getValue()));
     }
 
     @Override
+    @SuppressWarnings("ReferenceEquality")
     public void onDone(Status status) {
+      if (maxRetryDuration.compareTo(Duration.ZERO) > 0) {
+        // Remove the requests that have exceeded the retry time so they are not retried.
+        long nowNanos = System.nanoTime();
+        long maxRetryDurationNanos = maxRetryDuration.toNanos();
+        Iterator<Map.Entry<Long, StreamAndRequest>> iterator = pending.entrySet().iterator();
+        int keptRequests = 0, removedRequests = 0;
+        while (iterator.hasNext()) {
+          StreamAndRequest streamAndRequest = checkNotNull(iterator.next().getValue());
+          PendingRequest pendingRequest = streamAndRequest.request;
+          if (!belongsToThisHandler(streamAndRequest)
+              || nowNanos - pendingRequest.getStartTimeNanos() < maxRetryDurationNanos) {
+            ++keptRequests;
+            continue;
+          }
+          ++removedRequests;
+          iterator.remove();
+          try {
+            pendingRequest.completeWithStatus(CommitStatus.ABORTED);
+          } catch (RuntimeException e) {
+            LOG.warn("Exception while aborting commit due to retry timeout.", e);
+          }
+        }
+        if (removedRequests > 0) {
+          LOG.info(
+              "Aborting {} commits which have exceeded retry deadline, kept {}. Work will be retried as needed by service.",
+              removedRequests,
+              keptRequests);
+        }
+      }
       if (status.isOk() && hasPendingRequests()) {
         LOG.warn("Unexpected requests without responses on drained physical stream, retrying.");
       }
@@ -270,7 +308,7 @@ private void flushInternal(Map<Long, PendingRequest> requests)
 
     if (requests.size() == 1) {
       Map.Entry<Long, PendingRequest> elem = requests.entrySet().iterator().next();
-      if (elem.getValue().request().getSerializedSize()
+      if (elem.getValue().getRequest().getSerializedSize()
           > AbstractWindmillStream.RPC_STREAM_CHUNK_SIZE) {
         issueMultiChunkRequest(elem.getKey(), elem.getValue());
       } else {
@@ -286,7 +324,7 @@ private void issueSingleRequest(long id, PendingRequest pendingRequest)
     StreamingCommitWorkRequest.Builder requestBuilder = StreamingCommitWorkRequest.newBuilder();
     requestBuilder
         .addCommitChunkBuilder()
-        .setComputationId(pendingRequest.computationId())
+        .setComputationId(pendingRequest.getComputationId())
         .setRequestId(id)
         .setShardingKey(pendingRequest.shardingKey())
         .setSerializedWorkItemCommit(pendingRequest.serializedCommit());
@@ -311,9 +349,9 @@ private void issueBatchedRequest(Map<Long, PendingRequest> requests)
     for (Map.Entry<Long, PendingRequest> entry : requests.entrySet()) {
       PendingRequest request = entry.getValue();
       StreamingCommitRequestChunk.Builder chunkBuilder = requestBuilder.addCommitChunkBuilder();
-      if (lastComputation == null || !lastComputation.equals(request.computationId())) {
-        chunkBuilder.setComputationId(request.computationId());
-        lastComputation = request.computationId();
+      if (lastComputation == null || !lastComputation.equals(request.getComputationId())) {
+        chunkBuilder.setComputationId(request.getComputationId());
+        lastComputation = request.getComputationId();
       }
       chunkBuilder
           .setRequestId(entry.getKey())
@@ -338,7 +376,7 @@ private void issueBatchedRequest(Map<Long, PendingRequest> requests)
 
   private void issueMultiChunkRequest(long id, PendingRequest pendingRequest)
       throws WindmillStreamShutdownException {
-    checkNotNull(pendingRequest.computationId(), "Cannot commit WorkItem w/o a computationId.");
+    checkNotNull(pendingRequest.getComputationId(), "Cannot commit WorkItem w/o a computationId.");
     ByteString serializedCommit = pendingRequest.serializedCommit();
     synchronized (this) {
       if (isShutdown) {
@@ -359,7 +397,7 @@ private void issueMultiChunkRequest(long id, PendingRequest pendingRequest)
             StreamingCommitRequestChunk.newBuilder()
                 .setRequestId(id)
                 .setSerializedWorkItemCommit(chunk)
-                .setComputationId(pendingRequest.computationId())
+                .setComputationId(pendingRequest.getComputationId())
                 .setShardingKey(pendingRequest.shardingKey());
         int remaining = serializedCommit.size() - end;
         if (remaining > 0) {
@@ -376,34 +414,46 @@ private void issueMultiChunkRequest(long id, PendingRequest pendingRequest)
     }
   }
 
-  @AutoValue
-  abstract static class PendingRequest {
+  private static class PendingRequest {
+    private final String computationId;
+    private final WorkItemCommitRequest request;
+    private final Consumer<CommitStatus> onDone;
+    private final long startTimeNanos; // System.nanoTime() of when request began.
 
-    private static PendingRequest create(
+    private PendingRequest(
         String computationId, WorkItemCommitRequest request, Consumer<CommitStatus> onDone) {
-      return new AutoValue_GrpcCommitWorkStream_PendingRequest(computationId, request, onDone);
+      this.computationId = computationId;
+      this.request = request;
+      this.onDone = onDone;
+      this.startTimeNanos = System.nanoTime();
     }
 
-    abstract String computationId();
+    String getComputationId() {
+      return computationId;
+    }
 
-    abstract WorkItemCommitRequest request();
+    WorkItemCommitRequest getRequest() {
+      return request;
+    }
 
-    abstract Consumer<CommitStatus> onDone();
+    long getStartTimeNanos() {
+      return startTimeNanos;
+    }
 
     private long getBytes() {
-      return (long) request().getSerializedSize() + computationId().length();
+      return (long) request.getSerializedSize() + computationId.length();
     }
 
     private ByteString serializedCommit() {
-      return request().toByteString();
+      return request.toByteString();
     }
 
     private void completeWithStatus(CommitStatus commitStatus) {
-      onDone().accept(commitStatus);
+      onDone.accept(commitStatus);
     }
 
     private long shardingKey() {
-      return request().getShardingKey();
+      return request.getShardingKey();
     }
 
     private void abort() {
@@ -462,7 +512,7 @@ public boolean commitWorkItem(
         return false;
       }
 
-      PendingRequest request = PendingRequest.create(computation, commitRequest, onDone);
+      PendingRequest request = new PendingRequest(computation, commitRequest, onDone);
       add(idGenerator.incrementAndGet(), request);
       return true;
     }