[pull] master from apache:master

.github/trigger_files/beam_CloudML_Benchmarks_Dataflow.json

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

.test-infra/tools/stale_cleaner.py

@@ -365,6 +365,7 @@ def clean_pubsub_topics():
     prefixes = [
         "psit_topic_input",
         "psit_topic_output",
+        "psit_topic_ordering",
         "wc_topic_input",
         "wc_topic_output",
         "leader_board_it_input_topic",
@@ -421,6 +422,12 @@ def clean_pubsub_subscriptions():
     # Restrict subscription cleanup to the NYC taxi prefix only.
     prefixes = [
         "taxirides-realtime_beam_",
+        "pubsub_io_performance",
+        "psit_sub_input",
+        "psit_sub_output",
+        "psit_sub_ordering",
+        "wc_subscription_input",
+        "wc_subscription_output",
     ]
 
     # Create a PubSubSubscriptionCleaner instance

infra/enforcement/README.md

@@ -138,7 +138,7 @@ The enforcement tools are consolidated into a single daily workflow (`.github/wo
 
 This unified workflow executes both security domains sequentially:
 - **IAM Policy Enforcement:** Validates user bindings against the defined policies.
-- **Unmanaged Keys Audit:** Detects rogue service account keys generated outside the official rotation system.
+- **Unmanaged Keys Audit:** Detects rogue service account keys generated outside the official rotation system and reports them to the `[IAC_DRIFT_SA_KEY]` issue.
 
 **Note**:
 - **Manual trigger**: The workflow can also be triggered manually via `workflow_dispatch`.
@@ -175,7 +175,7 @@ python account_keys.py --action generate
 - **check**: Validates service account keys and their permissions against defined policies and reports any differences (default behavior)
 - **announce**: Creates or updates a GitHub issue and sends an email notification when service account keys policies differ from the defined ones.
   - For general configuration errors, it updates the main compliance issue.
-  - **For unmanaged/rogue keys (Security Alerts)**, it consolidates alerts into a dedicated `[SECURITY]` issue acting as a live dashboard. It updates the issue by placing the newest audit report at the top and moving the previous reports into a collapsed `<details>` history section. If the keys are revoked and the infrastructure becomes healthy, the system automatically resolves and closes the issue.
+  - **For unmanaged/rogue keys**, it consolidates alerts into a dedicated `[IAC_DRIFT_SA_KEY]` issue acting as a live dashboard. It updates the issue by placing the newest audit report at the top and moving the previous reports into a collapsed `<details>` history section. If the keys are revoked and the infrastructure becomes healthy, the system automatically resolves and closes the issue.
 - **print**: Prints announcement details for testing purposes without creating actual GitHub issues or sending emails
 - **generate**: Updates the compliance file to match the current GCP service account keys and Secret Manager permissions
 

infra/enforcement/account_keys.py

@@ -28,6 +28,7 @@
 
 SECRET_MANAGER_LABEL = "beam-infra-secret-manager"
 IAC_DRIFT_SA_KEY = "IAC_DRIFT_SA_KEY"
+ACCOUNT_KEYS_POLICY = "ACCOUNT_KEYS_POLICY"
 
 class AuthorizedUser(TypedDict):
     email: str
@@ -385,8 +386,8 @@ def create_announcement(self, recipient: str) -> None:
 
         if general_issues:
             self.logger.info(f"Found {len(general_issues)} general compliance issues. Triggering announcement...")
-            title = f"[{IAC_DRIFT_SA_KEY}] Action Required: Unauthorized Service Accounts Detected"
-            body = f"Unauthorized Service Accounts Report\n\n"
+            title = f"[{ACCOUNT_KEYS_POLICY}] Action Required: Service Account Policy Drift"
+            body = f"Service Account Policy Drift Report\n\n"
             body += f"Account keys for project {self.project_id} are not compliant with the defined policies on {self.service_account_keys_file}\n\n"
             for issue in general_issues:
                 body += f"- {issue}\n"
@@ -423,8 +424,8 @@ def print_announcement(self, recipient: str) -> None:
 
         if general_issues:
             self.logger.info("Printing general compliance announcement...")
-            title = f"[IAC_DRIFT_SA_KEY] Action Required: Unauthorized Service Accounts Detected"
-            body = f"Unauthorized Service Accounts Report\n\n"
+            title = f"[{ACCOUNT_KEYS_POLICY}] Action Required: Service Account Policy Drift"
+            body = f"Service Account Policy Drift Report\n\n"
             body += f"Account keys for project {self.project_id} are not compliant with the defined policies on {self.service_account_keys_file}\n\n"
             for issue in general_issues:
                 body += f"- {issue}\n"

infra/enforcement/sending.py

@@ -228,7 +228,7 @@ def report_unmanaged_keys(self, project_id: str, compilance_issues: List[str]) -
             self.logger.info("No compliance issues to report to Github.")
             return
 
-        issue_title = "[SECURITY] Action Required: Unmanaged Service Account Keys Detected"
+        issue_title = "[IAC_DRIFT_SA_KEY] Action Required: Unmanaged Service Account Keys Detected"
         #markdown body
         timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
         new_report = f"### Unmanaged Keys Audit Report ({timestamp})\n"
@@ -238,6 +238,12 @@ def report_unmanaged_keys(self, project_id: str, compilance_issues: List[str]) -
             new_report += f"- {issue_text}\n"
 
         new_report += "\n*Please investigate and revoke these keys if they are not part of the official rotation system.*"
+        remediation = "\n\n### Remediation\n"
+        remediation += "1. Delete all reported keys as soon as possible.\n"
+        remediation += "2. Replace the deleted keys using the official Beam key rotation system. It creates the service account key and registers its key ID and private key in the corresponding managed Secret Manager secret (`<service-account-id>-key`). Do not create replacement keys manually in IAM.\n"
+        remediation += "3. Run the audit again to confirm that the reported keys have been removed and the replacement keys are managed by the rotation system.\n"
+        remediation += "\nFor more information, consult `infra/keys/README.md`."
+        new_report += remediation
         open_issues = self._get_open_issues(issue_title)
 
         if open_issues:
@@ -272,7 +278,7 @@ def resolve_unmanaged_keys(self) -> None:
         Finds any open security issues regarding rogue keys and automatically closes them
         if the infrastructure is now healthy.
         """
-        issue_title = "[SECURITY] Action Required: Unmanaged Service Account Keys Detected"
+        issue_title = "[IAC_DRIFT_SA_KEY] Action Required: Unmanaged Service Account Keys Detected"
         open_issues = self._get_open_issues(issue_title)
         if open_issues:
             target_issue = open_issues[0]

infra/enforcement/test_sending.py

@@ -51,7 +51,7 @@ def test_get_open_issues_flaky_retry(self, mock_request):
             "items": [
                 {
                     "number": 1234,
-                    "title": "[SECURITY] Action Required: Unmanaged Service Account Keys Detected",
+                    "title": "[IAC_DRIFT_SA_KEY] Action Required: Unmanaged Service Account Keys Detected",
                     "body": "Test body",
                     "state": "open",
                     "html_url": "https://github.com/apache/beam/issues/1234",
@@ -64,7 +64,7 @@ def test_get_open_issues_flaky_retry(self, mock_request):
         mock_request.side_effect = [mock_response_fail, mock_response_success]
 
         # Call get_open_issues
-        issues = self.client._get_open_issues("[SECURITY] Action Required: Unmanaged Service Account Keys Detected")
+        issues = self.client._get_open_issues("[IAC_DRIFT_SA_KEY] Action Required: Unmanaged Service Account Keys Detected")
 
         # Verify that two requests were made (one retry)
         self.assertEqual(mock_request.call_count, 2)
@@ -79,7 +79,7 @@ def test_get_open_issues_query_format(self, mock_request):
         mock_response.json.return_value = {"items": []}
         mock_request.return_value = mock_response
 
-        title = "[SECURITY] Action Required: Unmanaged Service Account Keys Detected"
+        title = "[IAC_DRIFT_SA_KEY] Action Required: Unmanaged Service Account Keys Detected"
         self.client._get_open_issues(title)
 
         # Verify that the query parameter was passed correctly to requests

sdks/java/io/snowflake/build.gradle

@@ -30,6 +30,8 @@ dependencies {
   implementation project(path: ":sdks:java:extensions:google-cloud-platform-core")
   permitUnusedDeclared project(path: ":sdks:java:extensions:google-cloud-platform-core")
   implementation library.java.slf4j_api
+  provided library.java.everit_json_schema
+  permitUnusedDeclared library.java.everit_json_schema
   implementation group: 'net.snowflake', name: 'snowflake-jdbc', version: '4.0.2'
   implementation group: 'com.opencsv', name: 'opencsv', version: '5.12.0'
   implementation 'net.snowflake:snowflake-ingest-sdk:4.4.2'

sdks/java/io/snowflake/src/main/java/org/apache/beam/sdk/io/snowflake/SnowflakeIO.java

@@ -34,7 +34,6 @@
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
-import javax.annotation.Nullable;
 import javax.sql.DataSource;
 import net.snowflake.client.api.datasource.SnowflakeDataSource;
 import net.snowflake.client.api.datasource.SnowflakeDataSourceFactory;
@@ -87,6 +86,7 @@
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Joiner;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Splitter;
+import org.checkerframework.checker.nullness.qual.Nullable;
 import org.joda.time.Duration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -231,6 +231,7 @@ public interface CsvMapper<T> extends Serializable {
    */
   @FunctionalInterface
   public interface UserDataMapper<T> extends Serializable {
+    @Nullable
     Object[] mapRow(T element);
   }