Log Fused Stage Name Instead of computation id in more places

.github/trigger_files/IO_Iceberg_Integration_Tests.json

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

.github/trigger_files/IO_Iceberg_Integration_Tests_Dataflow.json

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

.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.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/trigger_files/beam_PostCommit_Python_Dependency.json

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

.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": 2
+  "modification": 3
 }

.github/trigger_files/beam_PostCommit_Python_Xlang_IO_Direct.json

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

.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": 5
+  "modification": 7
 }

.github/workflows/beam_PostCommit_Java_Delta_IO_Dataflow.yml

@@ -64,8 +64,6 @@ jobs:
         job_phrase: ["Run PostCommit Java Delta IO Dataflow"]
     steps:
       - uses: actions/checkout@v7
-        with:
-          persist-credentials: false
       - name: Setup repository
         uses: ./.github/actions/setup-action
         with:

.github/workflows/beam_PreCommit_GHA.yml

@@ -80,6 +80,22 @@ jobs:
         comment_phrase: ${{ matrix.job_phrase }}
         github_token: ${{ secrets.GITHUB_TOKEN }}
         github_job: ${{ matrix.job_name }} (${{ matrix.job_phrase }})
+    - name: Check for gsutil references
+      run: |
+        echo "Checking codebase for gsutil..."
+        # Search for 'gsutil', excluding this workflow file itself to avoid false positives.
+        if git grep -n "gsutil" -- ':!.github/workflows/beam_PreCommit_GHA.yml'; then
+          echo "ERROR: Found references to gsutil in the codebase. Please use 'gcloud storage' instead."
+          exit 1
+        elif [ "$(date +%Y%m)" -ge 202704 ]; then
+          echo "ERROR: Current date is April 2027 or later."
+          echo "Please verify gsutil deprecation date is still March 2027 (Reference: https://docs.cloud.google.com/storage/docs/gsutil)."
+          echo "If so, then delete this workflow step."
+          exit 1
+        else
+          echo "SUCCESS: No references to gsutil found."
+        fi
+      shell: bash
     - name: Setup environment
       uses: ./.github/actions/setup-environment-action
       with:
@@ -90,7 +106,7 @@ jobs:
     - name: Validate GHA Allowlist
       uses: apache/infrastructure-actions/allowlist-check@main # zizmor: ignore[unpinned-uses]
     - name: Run zizmor
-      uses: zizmorcore/zizmor-action@6fc4b006235f201fdab3722e17240ab420d580e5 # v0.6.1
+      uses: zizmorcore/zizmor-action@3dc1ecc9bcb9e94e9b2c709687979e1298497054 # v0.6.2
       with:
         advanced-security: true
     - name: run GHA PreCommit script

.github/workflows/beam_Publish_Beam_SDK_Snapshots.yml

@@ -90,8 +90,21 @@ jobs:
         # This is needed to run pipelines that use the default environment at HEAD, for example, when a
         # pipeline uses an expansion service built from HEAD.
         run: |
-          BEAM_VERSION_LINE=$(cat gradle.properties | grep "sdk_version")
-          echo "BEAM_VERSION=${BEAM_VERSION_LINE#*sdk_version=}" >> $GITHUB_ENV
+          BEAM_VERSION_LINE=$(grep -m1 '^sdk_version=' gradle.properties || true)
+          if [ -z "$BEAM_VERSION_LINE" ]; then
+            echo "Could not find sdk_version in gradle.properties"
+            exit 1
+          fi
+
+          BEAM_VERSION="${BEAM_VERSION_LINE#sdk_version=}"
+
+          # Prevent environment file injection via CR/LF.
+          if [[ "$BEAM_VERSION" =~ [$'\r\n'] ]]; then
+            echo "Invalid sdk_version: contains newline characters"
+            exit 1
+          fi
+
+          printf 'BEAM_VERSION=%s\n' "$BEAM_VERSION" >> "$GITHUB_ENV"
       - name: Set latest tag only on master branch
         if: github.ref == 'refs/heads/master'
         run: echo "LATEST_TAG=,latest" >> $GITHUB_ENV

.github/workflows/build_wheels.yml

@@ -201,7 +201,7 @@ jobs:
     if: needs.check_env_variables.outputs.gcp-variables-set == 'true' && github.event_name != 'pull_request'
     steps:
       - name: Remove existing files on GCS bucket
-        run: gsutil rm -r ${GCP_PATH} || true
+        run: gcloud storage rm -r ${GCP_PATH} || true
 
   upload_source_to_gcs:
     name: Upload python source distribution to GCS bucket
@@ -217,7 +217,7 @@ jobs:
           name: source_zip
           path: source/
       - name: Copy sources to GCS bucket
-        run: gsutil cp -r -a public-read source/* ${GCP_PATH}
+        run: gcloud storage cp -r --predefined-acl=publicRead source/* ${GCP_PATH}
 
   build_wheels:
     name: Build python ${{matrix.py_version}} wheels on ${{matrix.os_python.arch}} for ${{ matrix.os_python.os }}
@@ -330,7 +330,7 @@ jobs:
         merge-multiple: true
         path: wheelhouse/
     - name: Copy wheels to GCS bucket
-      run: gsutil cp -r -a public-read wheelhouse/* ${GCP_PATH}
+      run: gcloud storage cp -r --predefined-acl=publicRead wheelhouse/* ${GCP_PATH}
     - name: Create github action information file on GCS bucket
       run: |
         cat > github_action_info <<EOF
@@ -348,9 +348,9 @@ jobs:
         GITHUB_BASE_REF=$GITHUB_BASE_REF
         EOF
         echo $(cat github_action_info)
-        gsutil cp -a public-read github_action_info ${GCP_PATH}
+        gcloud storage cp --predefined-acl=publicRead github_action_info ${GCP_PATH}
     - name: Upload GitHub event file to GCS bucket
-      run: gsutil cp -a public-read ${GITHUB_EVENT_PATH} ${GCP_PATH}
+      run: gcloud storage cp --predefined-acl=publicRead ${GITHUB_EVENT_PATH} ${GCP_PATH}
 
   list_files_on_gcs:
     name: List files on Google Cloud Storage Bucket
@@ -361,7 +361,7 @@ jobs:
     if: needs.check_env_variables.outputs.gcp-variables-set == 'true' && github.event_name != 'pull_request'
     steps:
     - name: List file on Google Cloud Storage Bucket
-      run: gsutil ls "${GCP_PATH}*"
+      run: gcloud storage ls "${GCP_PATH}*"
 
   branch_repo_nightly:
     permissions:

.github/workflows/codeql.yml

@@ -153,7 +153,7 @@ jobs:
 
     # Initializes the CodeQL tools for scanning.
     - name: Initialize CodeQL
-      uses: github/codeql-action/init@v4
+      uses: github/codeql-action/init@v4.37.4
       with:
         languages: ${{ matrix.language }}
         build-mode: ${{ matrix.build-mode }}
@@ -192,6 +192,6 @@ jobs:
         fi
 
     - name: Perform CodeQL Analysis
-      uses: github/codeql-action/analyze@v4
+      uses: github/codeql-action/analyze@v4.37.4
       with:
         category: "/language:${{matrix.language}}"

.github/workflows/run_rc_validation_go_wordcount.yml

@@ -74,7 +74,7 @@ jobs:
       # --- Prism Steps ---
       - name: Download Input File from GCS (Prism)
         working-directory: ./${{ steps.setup_go.outputs.work_dir }}
-        run: gsutil cp ${{ env.GCS_INPUT_PATH }} ./kinglear.txt
+        run: gcloud storage cp ${{ env.GCS_INPUT_PATH }} ./kinglear.txt
 
       - name: Run Go WordCount with PrismRunner
         working-directory: ./${{ steps.setup_go.outputs.work_dir }}
@@ -118,17 +118,17 @@ jobs:
             --environment_config=apache/beam_go_sdk:${{ env.CONTAINER_TAG }}
 
       - name: Check Dataflow Output in GCS
-        working-directory: ./${{ steps.setup_go.outputs.work_dir }} # Added working directory for consistency, though not strictly needed for gsutil
+        working-directory: ./${{ steps.setup_go.outputs.work_dir }} # Added working directory for consistency, though not strictly needed for gcloud storage
         env:
           # Re-define the output path pattern for checking
           GCS_OUTPUT_PATH_PATTERN: ${{ env.GCS_OUTPUT_PREFIX }}/${{ env.RC_TAG }}/dataflow/output*
         run: |
           echo "Checking for Dataflow output files in GCS at: $GCS_OUTPUT_PATH_PATTERN"
-          # Use gsutil stat. The -q flag suppresses errors for non-existent files,
-          # allowing us to check the exit code. Exit code 0 means found, 1 means not found.
-          if gsutil -q stat $GCS_OUTPUT_PATH_PATTERN; then
+          # Use gcloud storage ls. Redirect stdout/stderr to suppress output.
+          # Exit code 0 means found, non-zero means not found.
+          if gcloud storage ls $GCS_OUTPUT_PATH_PATTERN > /dev/null 2>&1; then
             echo "Output files found in GCS."
-            FILE_COUNT=$(gsutil ls $GCS_OUTPUT_PATH_PATTERN | wc -l)
+            FILE_COUNT=$(gcloud storage ls $GCS_OUTPUT_PATH_PATTERN | wc -l)
             if [ "$FILE_COUNT" -gt 0 ]; then echo "Found $FILE_COUNT output file(s)."; else echo "Error: Output path exists but contains no files."; exit 1; fi
           else
             echo "Error: Output files not found in GCS at $GCS_OUTPUT_PATH_PATTERN"

.github/workflows/run_rc_validation_python_mobile_gaming.yml

@@ -174,7 +174,7 @@ jobs:
       - name: Create GCS Bucket (if needed - reusing input bucket)
         run: |
           echo "Ensuring GCS Bucket exists: ${{ env.GCS_BUCKET }} in project ${{ env.GCP_PROJECT_ID }}"
-          gsutil mb -p ${{ env.GCP_PROJECT_ID }} ${{ env.GCS_BUCKET }} || echo "Bucket ${{ env.GCS_BUCKET }} likely already exists."
+          gcloud storage buckets create ${{ env.GCS_BUCKET }} --project=${{ env.GCP_PROJECT_ID }} || echo "Bucket ${{ env.GCS_BUCKET }} likely already exists."
         shell: bash
 
       - name: Create PubSub Topic
@@ -533,8 +533,8 @@ jobs:
         if: always()
         run: |
           echo "Deleting objects in GCS Bucket: ${{ env.GCS_BUCKET }}/temp/"
-          gsutil -m rm -r "${{ env.GCS_BUCKET }}/temp/leaderboard/**" || echo "Failed to delete objects in GCS leaderboard temp folder."
-          gsutil -m rm -r "${{ env.GCS_BUCKET }}/temp/gamestats/**" || echo "Failed to delete objects in GCS gamestats temp folder."
+          gcloud storage rm -r "${{ env.GCS_BUCKET }}/temp/leaderboard/**" || echo "Failed to delete objects in GCS leaderboard temp folder."
+          gcloud storage rm -r "${{ env.GCS_BUCKET }}/temp/gamestats/**" || echo "Failed to delete objects in GCS gamestats temp folder."
           echo "Removing local log and jobid files..."
           rm -f leaderboard_dataflow_submit.log gamestats_dataflow_submit.log injector_run.log
           rm -f leaderboard_dataflow_jobid.txt # Remove Leaderboard jobid file here

.github/workflows/run_rc_validation_python_yaml.yml

@@ -265,9 +265,9 @@ jobs:
           sleep 60
           # Check if any files matching the pattern exist within the unique output folder.
           echo "Checking for files matching pattern: ${OUTPUT_PATTERN}"
-          if gsutil ls "${OUTPUT_PATTERN}" > /dev/null 2>&1; then
+          if gcloud storage ls "${OUTPUT_PATTERN}" > /dev/null 2>&1; then
             echo "SUCCESS: Found output files matching pattern in GCS."
-            gsutil ls "${OUTPUT_PATTERN}" # List found files
+            gcloud storage ls "${OUTPUT_PATTERN}" # List found files
           else
             echo "ERROR: No output files found matching pattern '${OUTPUT_PATTERN}' in GCS bucket."
             exit 1
@@ -280,7 +280,7 @@ jobs:
         run: |
           echo "Deleting unique run folder in GCS: ${GCS_UNIQUE_FOLDER_PREFIX}"
           # Delete the entire unique folder for this run, including temp, staging, and output
-          gsutil -m rm -r "${GCS_UNIQUE_FOLDER_PREFIX}" || echo "Failed to delete unique run folder ${GCS_UNIQUE_FOLDER_PREFIX} in GCS. Manual cleanup might be required."
+          gcloud storage rm -r "${GCS_UNIQUE_FOLDER_PREFIX}" || echo "Failed to delete unique run folder ${GCS_UNIQUE_FOLDER_PREFIX} in GCS. Manual cleanup might be required."
 
           echo "Removing local log, yaml, and jobid files..."
           rm -f yaml_dataflow_submit.log ${{ env.YAML_PIPELINE_FILE }} yaml_dataflow_jobid.txt

.test-infra/dataproc/flink_cluster.sh

@@ -91,7 +91,7 @@ YARN_APPLICATION_MASTER=""
 
 function upload_init_actions() {
   echo "Uploading initialization actions to GCS bucket: $GCS_BUCKET"
-  gsutil cp -r $INIT_ACTIONS_FOLDER_NAME/* $GCS_BUCKET/$INIT_ACTIONS_FOLDER_NAME
+  gcloud storage cp -r $INIT_ACTIONS_FOLDER_NAME/* $GCS_BUCKET/$INIT_ACTIONS_FOLDER_NAME
 }
 
 function get_leader() {

.test-infra/metrics/build.gradle

@@ -54,6 +54,7 @@ composeUp {
   dependsOn "createEmptyConfig"
 }
 dockerCompose {
+  projectName = 'beammetrics'
   environment.put 'DOCKER_CONFIG', project.rootProject.buildDir
 }
 

.test-infra/metrics/influxdb/Dockerfile

@@ -16,14 +16,13 @@
 # limitations under the License.
 ################################################################################
 
-FROM python:3.10-slim
-
-RUN pip install --no-cache-dir gsutil
+FROM alpine:latest
 
 WORKDIR /
 
-RUN gsutil cp gs://apache-beam-testing-metrics/influxdb-backup.tar.gz . && \
-tar xzf influxdb-backup.tar.gz
+RUN apk add --no-cache curl tar && \
+    curl -sS https://storage.googleapis.com/apache-beam-testing-metrics/influxdb-backup.tar.gz -o influxdb-backup.tar.gz && \
+    tar xzf influxdb-backup.tar.gz
 
 
 FROM influxdb:1.8.0

.test-infra/metrics/kubernetes/beam-influxdb-autobackup.yaml

@@ -47,9 +47,12 @@ spec:
                 - mountPath: /backup
                   name: shared-data
             - name: copy-to-gsc-bucket
-              image: gcr.io/apache-beam-testing/gsutil
+              image: gcr.io/google.com/cloudsdktool/google-cloud-cli:stable-slim
+              env:
+                - name: CLOUDSDK_CORE_PROJECT
+                  value: apache-beam-testing
               command: ['sh', '-c', 'tar czf influxdb-backup.tar.gz /backup
-                        && gsutil cp influxdb-backup.tar.gz
+                        && gcloud storage cp influxdb-backup.tar.gz
                         gs://apache-beam-testing-metrics/']
               volumeMounts:
                 - mountPath: /backup

.test-infra/tools/stale_cleaner.py

@@ -336,10 +336,13 @@ def _active_resources(self) -> dict:
             for subscription in self.client.list_subscriptions(request={"project": self.project_path}):
                 subscription_name = subscription.name
                 # Apply prefix filtering if prefixes are defined
-                if not self.prefixes or any(subscription_name.startswith(f"{self.project_path}/subscriptions/{prefix}") for prefix in self.prefixes):
-                    # Check if the subscription has a topic associated with it
-                    if subscription.detached:
+                if subscription.detached:
                         d[subscription_name] = GoogleCloudResource(resource_name=subscription_name, clock=self.clock)
+                #Only attached subscriptions with the NYC taxi prefix are eligible.
+                elif any(
+                    subscription_name.startswith(f"{self.project_path}/subscriptions/{prefix}") for prefix in self.prefixes
+                ):
+                    d[subscription_name] = GoogleCloudResource(resource_name=subscription_name, clock=self.clock)
 
         return d
 
@@ -416,8 +419,10 @@ def clean_pubsub_subscriptions():
     project_id = DEFAULT_PROJECT_ID
     bucket_name = DEFAULT_BUCKET_NAME
 
-    # No prefixes are defined for subscriptions so we will delete all stale subscriptions
-    prefixes = []
+    # Restrict subscription cleanup to the NYC taxi prefix only.
+    prefixes = [
+        "taxirides-realtime_beam_",
+    ]
 
     # Create a PubSubSubscriptionCleaner instance
     cleaner = PubSubSubscriptionCleaner(project_id=project_id, bucket_name=bucket_name,