From b4523d86ef354a976b089f6198e5372f32a5b7f7 Mon Sep 17 00:00:00 2001 From: Danny Mccormick Date: Tue, 7 Oct 2025 07:33:06 -0400 Subject: [PATCH 01/24] x-lang gbek tests --- ...stCommit_XVR_PythonUsingJava_Dataflow.json | 4 + .../transforms/validate_runner_xlang_test.py | 85 +++++++++++++++++++ 2 files changed, 89 insertions(+) create mode 100644 .github/trigger_files/beam_PostCommit_XVR_PythonUsingJava_Dataflow.json diff --git a/.github/trigger_files/beam_PostCommit_XVR_PythonUsingJava_Dataflow.json b/.github/trigger_files/beam_PostCommit_XVR_PythonUsingJava_Dataflow.json new file mode 100644 index 000000000000..6a55e29ae15d --- /dev/null +++ b/.github/trigger_files/beam_PostCommit_XVR_PythonUsingJava_Dataflow.json @@ -0,0 +1,4 @@ +{ + "comment": "Modify this file in a trivial way to cause this test suite to run.", + "modification": 1 +} \ No newline at end of file diff --git a/sdks/python/apache_beam/transforms/validate_runner_xlang_test.py b/sdks/python/apache_beam/transforms/validate_runner_xlang_test.py index 8e8e79648250..764167e8bc0f 100644 --- a/sdks/python/apache_beam/transforms/validate_runner_xlang_test.py +++ b/sdks/python/apache_beam/transforms/validate_runner_xlang_test.py @@ -52,17 +52,27 @@ import logging import os +import random +import string import typing import unittest import pytest import apache_beam as beam +from apache_beam.options.pipeline_options import SetupOptions from apache_beam.testing.test_pipeline import TestPipeline from apache_beam.testing.util import assert_that from apache_beam.testing.util import equal_to +from apache_beam.transforms.util import GcpSecret +from apache_beam.transforms.util import Secret from apache_beam.transforms.external import ImplicitSchemaPayloadBuilder +try: + from google.cloud import secretmanager +except ImportError: + secretmanager = None # type: ignore[assignment] + TEST_PREFIX_URN = "beam:transforms:xlang:test:prefix" TEST_MULTI_URN = "beam:transforms:xlang:test:multi" TEST_GBK_URN = "beam:transforms:xlang:test:gbk" @@ -140,6 +150,24 @@ def run_group_by_key(self, pipeline): | beam.Map(lambda x: "{}:{}".format(x[0], ','.join(sorted(x[1]))))) assert_that(res, equal_to(['0:1,2', '1:3'])) + def run_group_by_key_no_assert(self, pipeline): + """ + Target transform - GroupByKey, with no assertion for checking errors + (https://beam.apache.org/documentation/programming-guide/#groupbykey) + Test scenario - Grouping a collection of KV to a collection of + KV> by key + Boundary conditions checked - + - PCollection> to external transforms + - PCollection>> from external transforms + """ + with pipeline as p: + res = ( + p + | beam.Create([(0, "1"), (0, "2"), + (1, "3")], reshuffle=False).with_output_types( + typing.Tuple[int, str]) + | beam.ExternalTransform(TEST_GBK_URN, None, self.expansion_service)) + def run_cogroup_by_key(self, pipeline): """ Target transform - CoGroupByKey @@ -243,6 +271,42 @@ def run_partition(self, pipeline): class ValidateRunnerXlangTest(unittest.TestCase): _multiprocess_can_split_ = True + def setUp(self): + if secretmanager is not None: + self.project_id = 'apache-beam-testing' + secret_postfix = ''.join(random.choice(string.digits) for _ in range(6)) + self.secret_id = 'gbek_secret_tests_' + secret_postfix + self.client = secretmanager.SecretManagerServiceClient() + self.project_path = f'projects/{self.project_id}' + self.secret_path = f'{self.project_path}/secrets/{self.secret_id}' + try: + self.client.get_secret(request={'name': self.secret_path}) + except Exception: + self.client.create_secret( + request={ + 'parent': self.project_path, + 'secret_id': self.secret_id, + 'secret': { + 'replication': { + 'automatic': {} + } + } + }) + self.client.add_secret_version( + request={ + 'parent': self.secret_path, + 'payload': { + 'data': Secret.generate_secret_bytes() + } + }) + version_name = f'{self.secret_path}/versions/latest' + self.gcp_secret = GcpSecret(version_name) + self.secret_option = f'type:GcpSecret;version_name:{version_name}' + + def tearDown(self): + if secretmanager is not None: + self.client.delete_secret(request={'name': self.secret_path}) + def create_pipeline(self): test_pipeline = TestPipeline() test_pipeline.not_use_test_runner_api = True @@ -266,6 +330,27 @@ def test_group_by_key(self, test_pipeline=None): CrossLanguageTestPipelines().run_group_by_key( test_pipeline or self.create_pipeline()) + # This test and test_group_by_key_gbek_bad_secret validate that the gbek + # pipeline option is correctly passed through + @pytest.mark.uses_java_expansion_service + @pytest.mark.uses_python_expansion_service + @unittest.skipIf(secretmanager is None, 'GCP dependencies are not installed') + def test_group_by_key_gbek(self, test_pipeline=None): + test_pipeline = test_pipeline or self.create_pipeline() + good_secret = self.secret_option + test_pipeline.options.view_as(SetupOptions).gbek = good_secret + CrossLanguageTestPipelines().run_group_by_key(test_pipeline) + + @pytest.mark.uses_java_expansion_service + @pytest.mark.uses_python_expansion_service + @unittest.skipIf(secretmanager is None, 'GCP dependencies are not installed') + def test_group_by_key_gbek_bad_secret(self, test_pipeline=None): + test_pipeline = test_pipeline or self.create_pipeline() + nonexistent_secret = 'version_name:nonexistent_secret' + test_pipeline.options.view_as(SetupOptions).gbek = nonexistent_secret + with self.assertRaisesRegex(Exception, 'Secret string must contain a valid type parameter'): + CrossLanguageTestPipelines().run_group_by_key_no_assert(test_pipeline) + @pytest.mark.uses_java_expansion_service @pytest.mark.uses_python_expansion_service def test_cogroup_by_key(self, test_pipeline=None): From 343189f4d603a19a3f9fcd7ab14740ae25133867 Mon Sep 17 00:00:00 2001 From: Danny Mccormick Date: Tue, 7 Oct 2025 08:55:48 -0400 Subject: [PATCH 02/24] Add java test --- ...stCommit_XVR_JavaUsingPython_Dataflow.json | 4 + .../construction/ValidateRunnerXlangTest.java | 99 +++++++++++++++++++ 2 files changed, 103 insertions(+) create mode 100644 .github/trigger_files/beam_PostCommit_XVR_JavaUsingPython_Dataflow.json diff --git a/.github/trigger_files/beam_PostCommit_XVR_JavaUsingPython_Dataflow.json b/.github/trigger_files/beam_PostCommit_XVR_JavaUsingPython_Dataflow.json new file mode 100644 index 000000000000..6a55e29ae15d --- /dev/null +++ b/.github/trigger_files/beam_PostCommit_XVR_JavaUsingPython_Dataflow.json @@ -0,0 +1,4 @@ +{ + "comment": "Modify this file in a trivial way to cause this test suite to run.", + "modification": 1 +} \ No newline at end of file diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java index c41b2151d4cc..72e98360b8bb 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java @@ -19,6 +19,7 @@ import java.io.IOException; import java.io.Serializable; +import java.security.SecureRandom; import java.util.Arrays; import org.apache.beam.model.pipeline.v1.ExternalTransforms; import org.apache.beam.sdk.Pipeline; @@ -286,6 +287,104 @@ public void test() { } } + /** + * Motivation behind GroupByKeyWithGbekTest. + * + *

Target transform – GroupByKey + * (https://beam.apache.org/documentation/programming-guide/#groupbykey) Test scenario – Grouping + * a collection of KV to a collection of KV> by key Boundary conditions + * checked – –> PCollection> to external transforms –> PCollection>> + * from external transforms while using GroupByEncryptedKey overrides + */ + @RunWith(JUnit4.class) + public static class GroupByKeyWithGbekTest extends ValidateRunnerXlangTestBase { + private static final String PROJECT_ID = "apache-beam-testing"; + private static final String SECRET_ID = "gbek-test"; + private static String gcpSecretVersionName; + private static String secretId; + @Rule public ExpectedException thrown = ExpectedException.none(); + + @BeforeClass + public static void setUpClass() throws IOException { + secretId = String.format("%s-%d", SECRET_ID, new SecureRandom().nextInt(10000)); + SecretManagerServiceClient client; + try { + client = SecretManagerServiceClient.create(); + } catch (IOException e) { + gcpSecretVersionName = null; + return; + } + ProjectName projectName = ProjectName.of(PROJECT_ID); + SecretName secretName = SecretName.of(PROJECT_ID, secretId); + + try { + client.getSecret(secretName); + } catch (Exception e) { + com.google.cloud.secretmanager.v1.Secret secret = + com.google.cloud.secretmanager.v1.Secret.newBuilder() + .setReplication( + com.google.cloud.secretmanager.v1.Replication.newBuilder() + .setAutomatic( + com.google.cloud.secretmanager.v1.Replication.Automatic.newBuilder() + .build()) + .build()) + .build(); + client.createSecret(projectName, secretId, secret); + byte[] secretBytes = new byte[32]; + new SecureRandom().nextBytes(secretBytes); + client.addSecretVersion( + secretName, SecretPayload.newBuilder().setData(ByteString.copyFrom(secretBytes)).build()); + } + gcpSecretVersionName = secretName.toString() + "/versions/latest"; + expansionAddr = + String.format("localhost:%s", Integer.valueOf(System.getProperty("expansionPort"))); + } + + @AfterClass + public static void tearDownClass() throws IOException { + if (gcpSecretVersionName != null) { + SecretManagerServiceClient client = SecretManagerServiceClient.create(); + SecretName secretName = SecretName.of(PROJECT_ID, secretId); + client.deleteSecret(secretName); + } + } + + @Test + @Category({ + ValidatesRunner.class, + UsesJavaExpansionService.class, + UsesPythonExpansionService.class + }) + public void test() { + if (gcpSecretVersionName == null) { + // Skip test if we couldn't set up secret manager + return; + } + PipelineOptions options = testPipeline.testingPipelineOptions(); + options.setGbek(String.format("type:gcpsecret;version_name:%s", gcpSecretVersionName)); + testPipeline = Pipeline.create(options); + groupByKeyTest(testPipeline); + } + + @Test + @Category({ + ValidatesRunner.class, + UsesJavaExpansionService.class, + UsesPythonExpansionService.class + }) + public void testFailure() { + if (gcpSecretVersionName == null) { + // Skip test if we couldn't set up secret manager + return; + } + PipelineOptions options = testPipeline.testingPipelineOptions(); + options.setGbek(String.format("version_name:%s", gcpSecretVersionName)); + testPipeline = Pipeline.create(options); + thrown.expect(RuntimeException.class); + groupByKeyTest(testPipeline); + } + } + /** * Motivation behind coGroupByKeyTest. * From 8dc83ecc27d668dd4503653c5556b73f69ddd7ea Mon Sep 17 00:00:00 2001 From: Danny Mccormick Date: Tue, 7 Oct 2025 18:33:52 -0400 Subject: [PATCH 03/24] missing import --- .../beam/sdk/util/construction/ValidateRunnerXlangTest.java | 1 + 1 file changed, 1 insertion(+) diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java index 72e98360b8bb..5953f662b52f 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java @@ -45,6 +45,7 @@ import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables; import org.junit.Test; import org.junit.experimental.categories.Category; +import org.junit.rules.ExpectedException; import org.junit.runner.RunWith; import org.junit.runners.JUnit4; From 2c260ecfb1168b314c8ec66ce38e1b586f221000 Mon Sep 17 00:00:00 2001 From: Danny Mccormick Date: Tue, 7 Oct 2025 19:26:25 -0400 Subject: [PATCH 04/24] Move towards standardizing on base64 --- .../beam/sdk/options/PipelineOptions.java | 5 ++- .../sdk/transforms/GroupByEncryptedKey.java | 10 +++-- .../transforms/GroupByEncryptedKeyTest.java | 7 ++- .../beam/sdk/transforms/GroupByKeyIT.java | 5 ++- .../beam/sdk/transforms/GroupByKeyTest.java | 4 +- .../construction/ValidateRunnerXlangTest.java | 43 +++++++++++++------ .../apache_beam/options/pipeline_options.py | 5 ++- 7 files changed, 54 insertions(+), 25 deletions(-) diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java index 62022b219c2a..b977a375981b 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java @@ -420,7 +420,7 @@ public Long create(PipelineOptions options) { *

Beam will infer the secret type and value based on the secret itself. This guarantees that * any data at rest during the performing a GBK, so this can be used to guarantee that data is not * unencrypted. Runners with this behavior include the Dataflow, Flink, and Spark runners. The - * option should be structured like: + * secret should be a base64 encoded 32 byte value. The option should be structured like: * * 


    * --gbek=type:;:
@@ -439,7 +439,8 @@ public Long create(PipelineOptions options) {
           + " infer the secret type and value based on the secret itself. This guarantees that"
           + " any data at rest during the performing a GBK, so this can be used to guarantee"
           + " that data is not unencrypted. Runners with this behavior include the Dataflow,"
-          + " Flink, and Spark runners. The option should be structured like:"
+          + " Flink, and Spark runners. The secret should be a base64 encoded 32 byte value."
+          + " The option should be structured like:"
           + " --gbek=type:;:, for example "
           + " --gbek=type:GcpSecret;version_name:my_secret/versions/latest. All variables "
           + " should use snake case to allow consistency across languages.")
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByEncryptedKey.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByEncryptedKey.java
index 6ed0a31b3b95..a77fd63258e0 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByEncryptedKey.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByEncryptedKey.java
@@ -39,8 +39,8 @@
  * the output. This is useful when the keys contain sensitive data that should not be stored at rest
  * by the runner.
  *
- * 
The transform requires a {@link Secret} which returns a 32 byte secret which can be used to
- * generate a {@link SecretKeySpec} object using the HmacSHA256 algorithm.
+ * 
The transform requires a {@link Secret} which returns a base64 encoded 32 byte secret which
+ * can be used to generate a {@link SecretKeySpec} object using the HmacSHA256 algorithm.
  *
  * 
Note the following caveats: 1) Runners can implement arbitrary materialization steps, so this
  * does not guarantee that the whole pipeline will not have unencrypted data at rest by itself. 2)
@@ -153,7 +153,7 @@ private static class EncryptMessage extends DoFn, KV
     public void setup() {
       try {
         this.cipher = Cipher.getInstance("AES/GCM/NoPadding");
-        this.secretKeySpec = new SecretKeySpec(this.hmacKey.getSecretBytes(), "AES");
+        this.secretKeySpec =
+            new SecretKeySpec(
+                java.util.Base64.getDecoder().decode(this.hmacKey.getSecretBytes()), "AES");
       } catch (Exception ex) {
         throw new RuntimeException(
             "Failed to initialize cryptography libraries needed for GroupByEncryptedKey", ex);
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByEncryptedKeyTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByEncryptedKeyTest.java
index ba4c50e5a41e..1887bb2d7afb 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByEncryptedKeyTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByEncryptedKeyTest.java
@@ -58,7 +58,7 @@ public class GroupByEncryptedKeyTest implements Serializable {
 
   private static class FakeSecret implements Secret {
     private final byte[] secret =
-        "aKwI2PmqYFt2p5tNKCyBS5qYmHhHsGZc".getBytes(Charset.defaultCharset());
+        "YUt3STJQbXFZRnQycDV0TktDeUJTNXFZV0hoSHNHWmM=".getBytes(Charset.defaultCharset());
 
     @Override
     public byte[] getSecretBytes() {
@@ -123,7 +123,10 @@ public static void setup() throws IOException {
       byte[] secretBytes = new byte[32];
       new SecureRandom().nextBytes(secretBytes);
       client.addSecretVersion(
-          secretName, SecretPayload.newBuilder().setData(ByteString.copyFrom(secretBytes)).build());
+          secretName,
+          SecretPayload.newBuilder()
+              .setData(ByteString.copyFrom(java.util.Base64.getEncoder().encode(secretBytes)))
+              .build());
     }
     gcpSecret = new GcpSecret(secretName.toString() + "/versions/latest");
   }
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyIT.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyIT.java
index 60477a4c242f..2709856038a7 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyIT.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyIT.java
@@ -82,7 +82,10 @@ public static void setup() throws IOException {
       byte[] secretBytes = new byte[32];
       new SecureRandom().nextBytes(secretBytes);
       client.addSecretVersion(
-          secretName, SecretPayload.newBuilder().setData(ByteString.copyFrom(secretBytes)).build());
+          secretName,
+          SecretPayload.newBuilder()
+              .setData(ByteString.copyFrom(java.util.Base64.getEncoder().encode(secretBytes)))
+              .build());
     }
     gcpSecretVersionName = secretName.toString() + "/versions/latest";
   }
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java
index 326da99f1a81..a0c42ccc55a9 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java
@@ -153,7 +153,9 @@ public static void setup() throws IOException {
         new SecureRandom().nextBytes(secretBytes);
         client.addSecretVersion(
             secretName,
-            SecretPayload.newBuilder().setData(ByteString.copyFrom(secretBytes)).build());
+            SecretPayload.newBuilder()
+                .setData(ByteString.copyFrom(java.util.Base64.getEncoder().encode(secretBytes)))
+                .build());
       }
       gcpSecretVersionName = secretName.toString() + "/versions/latest";
     }
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
index 5953f662b52f..c6c0eca554d0 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
@@ -17,6 +17,11 @@
  */
 package org.apache.beam.sdk.util.construction;
 
+import com.google.cloud.secretmanager.v1.ProjectName;
+import com.google.cloud.secretmanager.v1.SecretManagerServiceClient;
+import com.google.cloud.secretmanager.v1.SecretName;
+import com.google.cloud.secretmanager.v1.SecretPayload;
+import com.google.protobuf.ByteString;
 import java.io.IOException;
 import java.io.Serializable;
 import java.security.SecureRandom;
@@ -24,11 +29,13 @@
 import org.apache.beam.model.pipeline.v1.ExternalTransforms;
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.coders.RowCoder;
+import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.schemas.Schema;
 import org.apache.beam.sdk.schemas.Schema.Field;
 import org.apache.beam.sdk.schemas.Schema.FieldType;
 import org.apache.beam.sdk.schemas.SchemaTranslation;
 import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
 import org.apache.beam.sdk.testing.UsesJavaExpansionService;
 import org.apache.beam.sdk.testing.UsesPythonExpansionService;
 import org.apache.beam.sdk.testing.ValidatesRunner;
@@ -43,6 +50,9 @@
 import org.apache.beam.sdk.values.Row;
 import org.apache.beam.sdk.values.TypeDescriptors;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Rule;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 import org.junit.rules.ExpectedException;
@@ -306,7 +316,7 @@ public static class GroupByKeyWithGbekTest extends ValidateRunnerXlangTestBase {
     @Rule public ExpectedException thrown = ExpectedException.none();
 
     @BeforeClass
-    public static void setUpClass() throws IOException {
+    public static void setUpClass() {
       secretId = String.format("%s-%d", SECRET_ID, new SecureRandom().nextInt(10000));
       SecretManagerServiceClient client;
       try {
@@ -334,19 +344,26 @@ public static void setUpClass() throws IOException {
         byte[] secretBytes = new byte[32];
         new SecureRandom().nextBytes(secretBytes);
         client.addSecretVersion(
-            secretName, SecretPayload.newBuilder().setData(ByteString.copyFrom(secretBytes)).build());
+            secretName,
+            SecretPayload.newBuilder()
+                .setData(ByteString.copyFrom(java.util.Base64.getEncoder().encode(secretBytes)))
+                .build());
       }
       gcpSecretVersionName = secretName.toString() + "/versions/latest";
       expansionAddr =
-        String.format("localhost:%s", Integer.valueOf(System.getProperty("expansionPort")));
+          String.format("localhost:%s", Integer.valueOf(System.getProperty("expansionPort")));
     }
 
     @AfterClass
-    public static void tearDownClass() throws IOException {
+    public static void tearDownClass() {
       if (gcpSecretVersionName != null) {
-        SecretManagerServiceClient client = SecretManagerServiceClient.create();
-        SecretName secretName = SecretName.of(PROJECT_ID, secretId);
-        client.deleteSecret(secretName);
+        try {
+          SecretManagerServiceClient client = SecretManagerServiceClient.create();
+          SecretName secretName = SecretName.of(PROJECT_ID, secretId);
+          client.deleteSecret(secretName);
+        } catch (IOException e) {
+          // Do nothing.
+        }
       }
     }
 
@@ -361,10 +378,10 @@ public void test() {
         // Skip test if we couldn't set up secret manager
         return;
       }
-      PipelineOptions options = testPipeline.testingPipelineOptions();
+      PipelineOptions options = TestPipeline.testingPipelineOptions();
       options.setGbek(String.format("type:gcpsecret;version_name:%s", gcpSecretVersionName));
-      testPipeline = Pipeline.create(options);
-      groupByKeyTest(testPipeline);
+      TestPipeline updatedTestPipeline = TestPipeline.fromOptions(options);
+      groupByKeyTest(updatedTestPipeline);
     }
 
     @Test
@@ -378,11 +395,11 @@ public void testFailure() {
         // Skip test if we couldn't set up secret manager
         return;
       }
-      PipelineOptions options = testPipeline.testingPipelineOptions();
+      PipelineOptions options = TestPipeline.testingPipelineOptions();
       options.setGbek(String.format("version_name:%s", gcpSecretVersionName));
-      testPipeline = Pipeline.create(options);
+      TestPipeline updatedTestPipeline = TestPipeline.fromOptions(options);
       thrown.expect(RuntimeException.class);
-      groupByKeyTest(testPipeline);
+      groupByKeyTest(updatedTestPipeline);
     }
   }
 
diff --git a/sdks/python/apache_beam/options/pipeline_options.py b/sdks/python/apache_beam/options/pipeline_options.py
index 2d3b8b49d8d7..44810785d9b9 100644
--- a/sdks/python/apache_beam/options/pipeline_options.py
+++ b/sdks/python/apache_beam/options/pipeline_options.py
@@ -1726,8 +1726,9 @@ def _add_argparse_args(cls, parser):
             'secret itself. This guarantees that any data at rest during the '
             'GBK will be encrypted. Many runners only store data at rest when '
             'performing a GBK, so this can be used to guarantee that data is '
-            'not unencrypted. Runners with this behavior include the '
-            'Dataflow, Flink, and Spark runners. The option should be '
+            'not unencrypted. The secret should be a base64 encoded 32 byte '
+            'value. Runners with this behavior include the Dataflow, Flink, '
+            'and Spark runners. The option should be '
             'structured like: '
             '--gbek=type:;:, for example '
             '--gbek=type:GcpSecret;version_name:my_secret/versions/latest'))

From 54814f6d601df046d10d5f00bf1727c3a9718a3e Mon Sep 17 00:00:00 2001
From: Danny Mccormick 
Date: Tue, 7 Oct 2025 19:57:14 -0400
Subject: [PATCH 05/24] url encoded

---
 .../java/org/apache/beam/sdk/options/PipelineOptions.java   | 6 +++---
 .../org/apache/beam/sdk/transforms/GroupByEncryptedKey.java | 4 ++--
 .../apache/beam/sdk/transforms/GroupByEncryptedKeyTest.java | 4 ++--
 .../java/org/apache/beam/sdk/transforms/GroupByKeyIT.java   | 2 +-
 .../java/org/apache/beam/sdk/transforms/GroupByKeyTest.java | 2 +-
 sdks/python/apache_beam/options/pipeline_options.py         | 6 +++---
 6 files changed, 12 insertions(+), 12 deletions(-)

diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java
index b977a375981b..ff001b31a44a 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java
@@ -420,7 +420,7 @@ public Long create(PipelineOptions options) {
    * 
Beam will infer the secret type and value based on the secret itself. This guarantees that
    * any data at rest during the performing a GBK, so this can be used to guarantee that data is not
    * unencrypted. Runners with this behavior include the Dataflow, Flink, and Spark runners. The
-   * secret should be a base64 encoded 32 byte value. The option should be structured like:
+   * secret should be a url safe base64 encoded 32 byte value. The option should be structured like:
    *
    * 

    * --gbek=type:;:
@@ -439,8 +439,8 @@ public Long create(PipelineOptions options) {
           + " infer the secret type and value based on the secret itself. This guarantees that"
           + " any data at rest during the performing a GBK, so this can be used to guarantee"
           + " that data is not unencrypted. Runners with this behavior include the Dataflow,"
-          + " Flink, and Spark runners. The secret should be a base64 encoded 32 byte value."
-          + " The option should be structured like:"
+          + " Flink, and Spark runners. The secret should be a url safe base64 encoded 32 byte"
+          + " value. The option should be structured like:"
           + " --gbek=type:;:, for example "
           + " --gbek=type:GcpSecret;version_name:my_secret/versions/latest. All variables "
           + " should use snake case to allow consistency across languages.")
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByEncryptedKey.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByEncryptedKey.java
index a77fd63258e0..1f4b7535d89e 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByEncryptedKey.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByEncryptedKey.java
@@ -153,7 +153,7 @@ private static class EncryptMessage extends DoFn, KV;:, for example '
             '--gbek=type:GcpSecret;version_name:my_secret/versions/latest'))

From dfe121bcf3c562bdc1ac1d7ed3ec579374d52040 Mon Sep 17 00:00:00 2001
From: Danny Mccormick 
Date: Tue, 7 Oct 2025 20:01:54 -0400
Subject: [PATCH 06/24] More doc

---
 .../java/org/apache/beam/sdk/options/PipelineOptions.java | 8 ++++++--
 sdks/python/apache_beam/options/pipeline_options.py       | 5 ++++-
 2 files changed, 10 insertions(+), 3 deletions(-)

diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java
index ff001b31a44a..989e3a1e3193 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java
@@ -432,7 +432,9 @@ public Long create(PipelineOptions options) {
    * --gbek=type:GcpSecret;version_name:my_secret/versions/latest"
    * 

    *
-   * All variables should use snake case to allow consistency across languages.
+   * All variables should use snake case to allow consistency across languages. For an example of
+   * generating a properly formatted secret, see
+   * https://github.com/apache/beam/blob/c8df4da229da49d533491857e1bb4ab5dbf4fd37/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyIT.java#L82
    */
   @Description(
       "When set, will replace all GroupByKey transforms in the pipeline the option. Beam will"
@@ -440,7 +442,9 @@ public Long create(PipelineOptions options) {
           + " any data at rest during the performing a GBK, so this can be used to guarantee"
           + " that data is not unencrypted. Runners with this behavior include the Dataflow,"
           + " Flink, and Spark runners. The secret should be a url safe base64 encoded 32 byte"
-          + " value. The option should be structured like:"
+          + " value. For an example of generating a properly formatted secret, see"
+          + " https://github.com/apache/beam/blob/c8df4da229da49d533491857e1bb4ab5dbf4fd37/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyIT.java#L82"
+          + " When passing in the gbek option, it should be structured like:"
           + " --gbek=type:;:, for example "
           + " --gbek=type:GcpSecret;version_name:my_secret/versions/latest. All variables "
           + " should use snake case to allow consistency across languages.")
diff --git a/sdks/python/apache_beam/options/pipeline_options.py b/sdks/python/apache_beam/options/pipeline_options.py
index 7d7566cf20bd..d4903cfb0092 100644
--- a/sdks/python/apache_beam/options/pipeline_options.py
+++ b/sdks/python/apache_beam/options/pipeline_options.py
@@ -1727,7 +1727,10 @@ def _add_argparse_args(cls, parser):
             'GBK will be encrypted. Many runners only store data at rest when '
             'performing a GBK, so this can be used to guarantee that data is '
             'not unencrypted. The secret should be a url safe base64 encoded '
-            '32 byte value. Runners with this behavior include the Dataflow, '
+            '32 byte value. To generate a secret in this format, you can use '
+            'Secret.generate_secret_bytes(). For an example of this, see '
+            'https://github.com/apache/beam/blob/c8df4da229da49d533491857e1bb4ab5dbf4fd37/sdks/python/apache_beam/transforms/util_test.py#L356. '
+            'Runners with this behavior include the Dataflow, '
             'Flink, and Spark runners. The option should be '
             'structured like: '
             '--gbek=type:;:, for example '

From 903891ec0e4cd41a38e72aef2d0469397c679300 Mon Sep 17 00:00:00 2001
From: Danny Mccormick 
Date: Tue, 7 Oct 2025 20:16:29 -0400
Subject: [PATCH 07/24] yapf

---
 .../apache_beam/transforms/validate_runner_xlang_test.py       | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/sdks/python/apache_beam/transforms/validate_runner_xlang_test.py b/sdks/python/apache_beam/transforms/validate_runner_xlang_test.py
index 764167e8bc0f..b811efe416db 100644
--- a/sdks/python/apache_beam/transforms/validate_runner_xlang_test.py
+++ b/sdks/python/apache_beam/transforms/validate_runner_xlang_test.py
@@ -348,7 +348,8 @@ def test_group_by_key_gbek_bad_secret(self, test_pipeline=None):
     test_pipeline = test_pipeline or self.create_pipeline()
     nonexistent_secret = 'version_name:nonexistent_secret'
     test_pipeline.options.view_as(SetupOptions).gbek = nonexistent_secret
-    with self.assertRaisesRegex(Exception, 'Secret string must contain a valid type parameter'):
+    with self.assertRaisesRegex(
+        Exception, 'Secret string must contain a valid type parameter'):
       CrossLanguageTestPipelines().run_group_by_key_no_assert(test_pipeline)
 
   @pytest.mark.uses_java_expansion_service

From 5258999adf78263e1fc9f2d19b0c0238ffb4851b Mon Sep 17 00:00:00 2001
From: Danny Mccormick 
Date: Tue, 7 Oct 2025 23:22:16 -0400
Subject: [PATCH 08/24] test cleanup

---
 .../sdk/util/construction/ValidateRunnerXlangTest.java | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
index c6c0eca554d0..f26c64193ced 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
@@ -380,8 +380,8 @@ public void test() {
       }
       PipelineOptions options = TestPipeline.testingPipelineOptions();
       options.setGbek(String.format("type:gcpsecret;version_name:%s", gcpSecretVersionName));
-      TestPipeline updatedTestPipeline = TestPipeline.fromOptions(options);
-      groupByKeyTest(updatedTestPipeline);
+      testPipeline = TestPipeline.fromOptions(options);
+      groupByKeyTest(testPipeline);
     }
 
     @Test
@@ -397,9 +397,9 @@ public void testFailure() {
       }
       PipelineOptions options = TestPipeline.testingPipelineOptions();
       options.setGbek(String.format("version_name:%s", gcpSecretVersionName));
-      TestPipeline updatedTestPipeline = TestPipeline.fromOptions(options);
-      thrown.expect(RuntimeException.class);
-      groupByKeyTest(updatedTestPipeline);
+      testPipeline = TestPipeline.fromOptions(options);
+      thrown.expect(Exception.class);
+      groupByKeyTest(testPipeline);
     }
   }
 

From f8377c793a44317e737838700ce97a9242a93f24 Mon Sep 17 00:00:00 2001
From: Danny Mccormick 
Date: Wed, 8 Oct 2025 10:32:13 -0400
Subject: [PATCH 09/24] progress, kick presubmits

---
 .../construction/ValidateRunnerXlangTest.java | 61 ++++++++++++++-----
 1 file changed, 47 insertions(+), 14 deletions(-)

diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
index f26c64193ced..265139fecb1f 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
@@ -17,6 +17,9 @@
  */
 package org.apache.beam.sdk.util.construction;
 
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+
 import com.google.cloud.secretmanager.v1.ProjectName;
 import com.google.cloud.secretmanager.v1.SecretManagerServiceClient;
 import com.google.cloud.secretmanager.v1.SecretName;
@@ -26,16 +29,16 @@
 import java.io.Serializable;
 import java.security.SecureRandom;
 import java.util.Arrays;
+import java.util.List;
 import org.apache.beam.model.pipeline.v1.ExternalTransforms;
 import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.PipelineResult;
 import org.apache.beam.sdk.coders.RowCoder;
-import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.schemas.Schema;
 import org.apache.beam.sdk.schemas.Schema.Field;
 import org.apache.beam.sdk.schemas.Schema.FieldType;
 import org.apache.beam.sdk.schemas.SchemaTranslation;
 import org.apache.beam.sdk.testing.PAssert;
-import org.apache.beam.sdk.testing.TestPipeline;
 import org.apache.beam.sdk.testing.UsesJavaExpansionService;
 import org.apache.beam.sdk.testing.UsesPythonExpansionService;
 import org.apache.beam.sdk.testing.ValidatesRunner;
@@ -50,6 +53,8 @@
 import org.apache.beam.sdk.values.Row;
 import org.apache.beam.sdk.values.TypeDescriptors;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists;
+import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Rule;
@@ -313,7 +318,6 @@ public static class GroupByKeyWithGbekTest extends ValidateRunnerXlangTestBase {
     private static final String SECRET_ID = "gbek-test";
     private static String gcpSecretVersionName;
     private static String secretId;
-    @Rule public ExpectedException thrown = ExpectedException.none();
 
     @BeforeClass
     public static void setUpClass() {
@@ -367,6 +371,13 @@ public static void tearDownClass() {
       }
     }
 
+    @After
+    @Override
+    public void tearDown() {
+      // Override tearDown since we're doing our own assertion instead of relying on base class
+      // assertions
+    }
+
     @Test
     @Category({
       ValidatesRunner.class,
@@ -378,11 +389,28 @@ public void test() {
         // Skip test if we couldn't set up secret manager
         return;
       }
-      PipelineOptions options = TestPipeline.testingPipelineOptions();
-      options.setGbek(String.format("type:gcpsecret;version_name:%s", gcpSecretVersionName));
-      testPipeline = TestPipeline.fromOptions(options);
       groupByKeyTest(testPipeline);
+      List additionalArgs =
+          Lists.newArrayList(
+              String.format("--gbek=type:gcpsecret;version_name:%s", gcpSecretVersionName));
+      PipelineResult pipelineResult = testPipeline.runWithAdditionalOptionArgs(additionalArgs);
+      pipelineResult.waitUntilFinish();
+      assertThat(pipelineResult.getState(), equalTo(PipelineResult.State.DONE));
     }
+  }
+
+  /**
+   * Motivation behind GroupByKeyWithGbekFailureCaseTest.
+   *
+   * 
Target transform – GroupByKey
+   * (https://beam.apache.org/documentation/programming-guide/#groupbykey) Test scenario – Grouping
+   * a collection of KV to a collection of KV> by key Boundary conditions
+   * checked – –> PCollection> to external transforms –> PCollection>>
+   * from external transforms while using GroupByEncryptedKey overrides
+   */
+  @RunWith(JUnit4.class)
+  public static class GroupByKeyWithGbekFailureCaseTest extends ValidateRunnerXlangTestBase {
+    @Rule public ExpectedException thrown = ExpectedException.none();
 
     @Test
     @Category({
@@ -390,16 +418,21 @@ public void test() {
       UsesJavaExpansionService.class,
       UsesPythonExpansionService.class
     })
-    public void testFailure() {
-      if (gcpSecretVersionName == null) {
-        // Skip test if we couldn't set up secret manager
-        return;
-      }
-      PipelineOptions options = TestPipeline.testingPipelineOptions();
-      options.setGbek(String.format("version_name:%s", gcpSecretVersionName));
-      testPipeline = TestPipeline.fromOptions(options);
+    public void test() {
       thrown.expect(Exception.class);
       groupByKeyTest(testPipeline);
+      List additionalArgs =
+          Lists.newArrayList(String.format("--gbek=version_name:some_version"));
+      PipelineResult pipelineResult = testPipeline.runWithAdditionalOptionArgs(additionalArgs);
+      pipelineResult.waitUntilFinish();
+      assertThat(pipelineResult.getState(), equalTo(PipelineResult.State.DONE));
+    }
+
+    @After
+    @Override
+    public void tearDown() {
+      // Override tearDown since we're doing our own assertion instead of relying on base class
+      // assertions
     }
   }
 

From ee97593a0ac8572b5f1362b9ab9fb5043a343611 Mon Sep 17 00:00:00 2001
From: Danny Mccormick 
Date: Wed, 8 Oct 2025 11:32:08 -0400
Subject: [PATCH 10/24] use options

---
 .../construction/ValidateRunnerXlangTest.java   | 17 ++++++++---------
 1 file changed, 8 insertions(+), 9 deletions(-)

diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
index 265139fecb1f..d1fc455b717f 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
@@ -29,16 +29,17 @@
 import java.io.Serializable;
 import java.security.SecureRandom;
 import java.util.Arrays;
-import java.util.List;
 import org.apache.beam.model.pipeline.v1.ExternalTransforms;
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.PipelineResult;
 import org.apache.beam.sdk.coders.RowCoder;
+import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.schemas.Schema;
 import org.apache.beam.sdk.schemas.Schema.Field;
 import org.apache.beam.sdk.schemas.Schema.FieldType;
 import org.apache.beam.sdk.schemas.SchemaTranslation;
 import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
 import org.apache.beam.sdk.testing.UsesJavaExpansionService;
 import org.apache.beam.sdk.testing.UsesPythonExpansionService;
 import org.apache.beam.sdk.testing.ValidatesRunner;
@@ -53,7 +54,6 @@
 import org.apache.beam.sdk.values.Row;
 import org.apache.beam.sdk.values.TypeDescriptors;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
-import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists;
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -390,10 +390,9 @@ public void test() {
         return;
       }
       groupByKeyTest(testPipeline);
-      List additionalArgs =
-          Lists.newArrayList(
-              String.format("--gbek=type:gcpsecret;version_name:%s", gcpSecretVersionName));
-      PipelineResult pipelineResult = testPipeline.runWithAdditionalOptionArgs(additionalArgs);
+      PipelineOptions options = TestPipeline.testingPipelineOptions();
+      options.setGbek(String.format("type:gcpsecret;version_name:%s", gcpSecretVersionName));
+      PipelineResult pipelineResult = testPipeline.run(options);
       pipelineResult.waitUntilFinish();
       assertThat(pipelineResult.getState(), equalTo(PipelineResult.State.DONE));
     }
@@ -421,9 +420,9 @@ public static class GroupByKeyWithGbekFailureCaseTest extends ValidateRunnerXlan
     public void test() {
       thrown.expect(Exception.class);
       groupByKeyTest(testPipeline);
-      List additionalArgs =
-          Lists.newArrayList(String.format("--gbek=version_name:some_version"));
-      PipelineResult pipelineResult = testPipeline.runWithAdditionalOptionArgs(additionalArgs);
+      PipelineOptions options = TestPipeline.testingPipelineOptions();
+      options.setGbek(String.format("version_name:some_version"));
+      PipelineResult pipelineResult = testPipeline.run(options);
       pipelineResult.waitUntilFinish();
       assertThat(pipelineResult.getState(), equalTo(PipelineResult.State.DONE));
     }

From c88f0b8e305135072fdca61533db47d12a0d6ca7 Mon Sep 17 00:00:00 2001
From: Danny Mccormick 
Date: Thu, 9 Oct 2025 13:00:46 -0400
Subject: [PATCH 11/24] Additional pieces

---
 .../construction/ValidateRunnerXlangTest.java | 41 ++++++-------------
 .../runners/portability/expansion_service.py  | 12 +-----
 sdks/python/apache_beam/transforms/util.py    |  4 +-
 3 files changed, 17 insertions(+), 40 deletions(-)

diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
index d1fc455b717f..425679e59a26 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
@@ -314,6 +314,7 @@ public void test() {
    */
   @RunWith(JUnit4.class)
   public static class GroupByKeyWithGbekTest extends ValidateRunnerXlangTestBase {
+    @Rule public ExpectedException thrown = ExpectedException.none();
     private static final String PROJECT_ID = "apache-beam-testing";
     private static final String SECRET_ID = "gbek-test";
     private static String gcpSecretVersionName;
@@ -327,7 +328,8 @@ public static void setUpClass() {
         client = SecretManagerServiceClient.create();
       } catch (IOException e) {
         gcpSecretVersionName = null;
-        return;
+        // return;
+        throw new RuntimeException("test");
       }
       ProjectName projectName = ProjectName.of(PROJECT_ID);
       SecretName secretName = SecretName.of(PROJECT_ID, secretId);
@@ -350,7 +352,7 @@ public static void setUpClass() {
         client.addSecretVersion(
             secretName,
             SecretPayload.newBuilder()
-                .setData(ByteString.copyFrom(java.util.Base64.getEncoder().encode(secretBytes)))
+                .setData(ByteString.copyFrom(java.util.Base64.getUrlEncoder().encode(secretBytes)))
                 .build());
       }
       gcpSecretVersionName = secretName.toString() + "/versions/latest";
@@ -389,27 +391,14 @@ public void test() {
         // Skip test if we couldn't set up secret manager
         return;
       }
-      groupByKeyTest(testPipeline);
       PipelineOptions options = TestPipeline.testingPipelineOptions();
       options.setGbek(String.format("type:gcpsecret;version_name:%s", gcpSecretVersionName));
-      PipelineResult pipelineResult = testPipeline.run(options);
+      Pipeline pipeline = Pipeline.create(options);
+      groupByKeyTest(pipeline);
+      PipelineResult pipelineResult = pipeline.run();
       pipelineResult.waitUntilFinish();
       assertThat(pipelineResult.getState(), equalTo(PipelineResult.State.DONE));
     }
-  }
-
-  /**
-   * Motivation behind GroupByKeyWithGbekFailureCaseTest.
-   *
-   * 
Target transform – GroupByKey
-   * (https://beam.apache.org/documentation/programming-guide/#groupbykey) Test scenario – Grouping
-   * a collection of KV to a collection of KV> by key Boundary conditions
-   * checked – –> PCollection> to external transforms –> PCollection>>
-   * from external transforms while using GroupByEncryptedKey overrides
-   */
-  @RunWith(JUnit4.class)
-  public static class GroupByKeyWithGbekFailureCaseTest extends ValidateRunnerXlangTestBase {
-    @Rule public ExpectedException thrown = ExpectedException.none();
 
     @Test
     @Category({
@@ -417,22 +406,16 @@ public static class GroupByKeyWithGbekFailureCaseTest extends ValidateRunnerXlan
       UsesJavaExpansionService.class,
       UsesPythonExpansionService.class
     })
-    public void test() {
+    public void testFailure() {
       thrown.expect(Exception.class);
-      groupByKeyTest(testPipeline);
       PipelineOptions options = TestPipeline.testingPipelineOptions();
-      options.setGbek(String.format("version_name:some_version"));
-      PipelineResult pipelineResult = testPipeline.run(options);
+      options.setGbek("version_name:fake_secret");
+      Pipeline pipeline = Pipeline.create(options);
+      groupByKeyTest(pipeline);
+      PipelineResult pipelineResult = pipeline.run();
       pipelineResult.waitUntilFinish();
       assertThat(pipelineResult.getState(), equalTo(PipelineResult.State.DONE));
     }
-
-    @After
-    @Override
-    public void tearDown() {
-      // Override tearDown since we're doing our own assertion instead of relying on base class
-      // assertions
-    }
   }
 
   /**
diff --git a/sdks/python/apache_beam/runners/portability/expansion_service.py b/sdks/python/apache_beam/runners/portability/expansion_service.py
index 12e3ffb69702..4464d2f89b07 100644
--- a/sdks/python/apache_beam/runners/portability/expansion_service.py
+++ b/sdks/python/apache_beam/runners/portability/expansion_service.py
@@ -56,16 +56,8 @@ def __init__(self, options=None, loopback_address=None):
   def Expand(self, request, context=None):
     try:
       options = copy.deepcopy(self._options)
-      request_options = pipeline_options.PipelineOptions.from_runner_api(
-          request.pipeline_options)
-      # TODO(https://github.com/apache/beam/issues/20090): Figure out the
-      # correct subset of options to apply to expansion.
-      if request_options.view_as(
-          pipeline_options.StreamingOptions).update_compatibility_version:
-        options.view_as(
-            pipeline_options.StreamingOptions
-        ).update_compatibility_version = request_options.view_as(
-            pipeline_options.StreamingOptions).update_compatibility_version
+      options = pipeline_options.PipelineOptions.from_runner_api(
+          request.pipeline_options, options)
       pipeline = beam_pipeline.Pipeline(options=options)
 
       def with_pipeline(component, pcoll_id=None):
diff --git a/sdks/python/apache_beam/transforms/util.py b/sdks/python/apache_beam/transforms/util.py
index 5af9d904895a..5c5d368cb540 100644
--- a/sdks/python/apache_beam/transforms/util.py
+++ b/sdks/python/apache_beam/transforms/util.py
@@ -506,7 +506,9 @@ def process(
         continue
       seen_keys.add(encoded_key)
       real_key = self.key_coder.decode(encoded_key)
-
+      abc = (
+          real_key,
+          list(self.filter_elements_by_key(encoded_key, encoded_elements)))
       yield (
           real_key,
           list(self.filter_elements_by_key(encoded_key, encoded_elements)))

From 58192127d4fe1b27b81fa604362156fa4bcf18bc Mon Sep 17 00:00:00 2001
From: Danny Mccormick 
Date: Thu, 9 Oct 2025 13:01:36 -0400
Subject: [PATCH 12/24] Add pipeline options piece

---
 .../apache_beam/options/pipeline_options.py    | 18 ++++++++++++++----
 1 file changed, 14 insertions(+), 4 deletions(-)

diff --git a/sdks/python/apache_beam/options/pipeline_options.py b/sdks/python/apache_beam/options/pipeline_options.py
index d4903cfb0092..184616177dee 100644
--- a/sdks/python/apache_beam/options/pipeline_options.py
+++ b/sdks/python/apache_beam/options/pipeline_options.py
@@ -287,6 +287,10 @@ def _smart_split(self, values):
 
 
 class PipelineOptions(HasDisplayData):
+  # Set of options which should not be overriden when pipeline options are
+  # being merged (see from_runner_api). This primarily comes up when expanding
+  # the Python expansion service
+  NON_OVERIDABLE_OPTIONS = ['runner', 'experiments']
   """This class and subclasses are used as containers for command line options.
 
   These classes are wrappers over the standard argparse Python module
@@ -592,15 +596,21 @@ def to_struct_value(o):
         })
 
   @classmethod
-  def from_runner_api(cls, proto_options):
+  def from_runner_api(cls, proto_options, original_options = None):
     def from_urn(key):
       assert key.startswith('beam:option:')
       assert key.endswith(':v1')
       return key[12:-3]
 
-    return cls(
-        **{from_urn(key): value
-           for (key, value) in proto_options.items()})
+    parsed = {from_urn(key): value
+            for (key, value) in proto_options.items()}
+    if original_options is None:
+      return cls(
+          **parsed)
+    for (key, value) in parsed.items():
+      if value is not None and value and key not in cls.NON_OVERIDABLE_OPTIONS:
+        original_options._all_options[key] = value
+    return original_options
 
   def display_data(self):
     return self.get_all_options(drop_default=True, retain_unknown_options=True)

From d07ac314f3292656a2e909af4feb474821cccc29 Mon Sep 17 00:00:00 2001
From: Danny Mccormick 
Date: Thu, 9 Oct 2025 13:28:27 -0400
Subject: [PATCH 13/24] Format

---
 .../src/main/resources/beam/checkstyle/suppressions.xml   | 1 +
 sdks/python/apache_beam/options/pipeline_options.py       | 8 +++-----
 2 files changed, 4 insertions(+), 5 deletions(-)

diff --git a/sdks/java/build-tools/src/main/resources/beam/checkstyle/suppressions.xml b/sdks/java/build-tools/src/main/resources/beam/checkstyle/suppressions.xml
index 52e8467b1624..53cd7b7ad4d0 100644
--- a/sdks/java/build-tools/src/main/resources/beam/checkstyle/suppressions.xml
+++ b/sdks/java/build-tools/src/main/resources/beam/checkstyle/suppressions.xml
@@ -60,6 +60,7 @@
   
   
   
+  
   
   
   
diff --git a/sdks/python/apache_beam/options/pipeline_options.py b/sdks/python/apache_beam/options/pipeline_options.py
index 8139f0313b1f..0d602db7801b 100644
--- a/sdks/python/apache_beam/options/pipeline_options.py
+++ b/sdks/python/apache_beam/options/pipeline_options.py
@@ -596,17 +596,15 @@ def to_struct_value(o):
         })
 
   @classmethod
-  def from_runner_api(cls, proto_options, original_options = None):
+  def from_runner_api(cls, proto_options, original_options=None):
     def from_urn(key):
       assert key.startswith('beam:option:')
       assert key.endswith(':v1')
       return key[12:-3]
 
-    parsed = {from_urn(key): value
-            for (key, value) in proto_options.items()}
+    parsed = {from_urn(key): value for (key, value) in proto_options.items()}
     if original_options is None:
-      return cls(
-          **parsed)
+      return cls(**parsed)
     for (key, value) in parsed.items():
       if value is not None and value and key not in cls.NON_OVERIDABLE_OPTIONS:
         original_options._all_options[key] = value

From c27b0c9bfd349783c175ca3653f69e4767422ea2 Mon Sep 17 00:00:00 2001
From: Danny Mccormick 
Date: Thu, 9 Oct 2025 14:49:32 -0400
Subject: [PATCH 14/24] Move gbek into own test class

---
 sdks/python/apache_beam/transforms/util.py    |   4 +-
 .../transforms/validate_runner_xlang_test.py  | 119 +++++++++---------
 2 files changed, 64 insertions(+), 59 deletions(-)

diff --git a/sdks/python/apache_beam/transforms/util.py b/sdks/python/apache_beam/transforms/util.py
index 5c5d368cb540..5af9d904895a 100644
--- a/sdks/python/apache_beam/transforms/util.py
+++ b/sdks/python/apache_beam/transforms/util.py
@@ -506,9 +506,7 @@ def process(
         continue
       seen_keys.add(encoded_key)
       real_key = self.key_coder.decode(encoded_key)
-      abc = (
-          real_key,
-          list(self.filter_elements_by_key(encoded_key, encoded_elements)))
+
       yield (
           real_key,
           list(self.filter_elements_by_key(encoded_key, encoded_elements)))
diff --git a/sdks/python/apache_beam/transforms/validate_runner_xlang_test.py b/sdks/python/apache_beam/transforms/validate_runner_xlang_test.py
index b811efe416db..40f7478fd5d4 100644
--- a/sdks/python/apache_beam/transforms/validate_runner_xlang_test.py
+++ b/sdks/python/apache_beam/transforms/validate_runner_xlang_test.py
@@ -161,7 +161,7 @@ def run_group_by_key_no_assert(self, pipeline):
      - PCollection>> from external transforms
     """
     with pipeline as p:
-      res = (
+      (
           p
           | beam.Create([(0, "1"), (0, "2"),
                          (1, "3")], reshuffle=False).with_output_types(
@@ -271,6 +271,66 @@ def run_partition(self, pipeline):
 class ValidateRunnerXlangTest(unittest.TestCase):
   _multiprocess_can_split_ = True
 
+  def create_pipeline(self):
+    test_pipeline = TestPipeline()
+    test_pipeline.not_use_test_runner_api = True
+    return test_pipeline
+
+  @pytest.mark.uses_java_expansion_service
+  @pytest.mark.uses_python_expansion_service
+  def test_prefix(self, test_pipeline=None):
+    CrossLanguageTestPipelines().run_prefix(
+        test_pipeline or self.create_pipeline())
+
+  @pytest.mark.uses_java_expansion_service
+  @pytest.mark.uses_python_expansion_service
+  def test_multi_input_output_with_sideinput(self, test_pipeline=None):
+    CrossLanguageTestPipelines().run_multi_input_output_with_sideinput(
+        test_pipeline or self.create_pipeline())
+
+  @pytest.mark.uses_java_expansion_service
+  @pytest.mark.uses_python_expansion_service
+  def test_group_by_key(self, test_pipeline=None):
+    CrossLanguageTestPipelines().run_group_by_key(
+        test_pipeline or self.create_pipeline())
+
+  @pytest.mark.uses_java_expansion_service
+  @pytest.mark.uses_python_expansion_service
+  def test_cogroup_by_key(self, test_pipeline=None):
+    CrossLanguageTestPipelines().run_cogroup_by_key(
+        test_pipeline or self.create_pipeline())
+
+  @pytest.mark.uses_java_expansion_service
+  @pytest.mark.uses_python_expansion_service
+  def test_combine_globally(self, test_pipeline=None):
+    CrossLanguageTestPipelines().run_combine_globally(
+        test_pipeline or self.create_pipeline())
+
+  @pytest.mark.uses_java_expansion_service
+  @pytest.mark.uses_python_expansion_service
+  def test_combine_per_key(self, test_pipeline=None):
+    CrossLanguageTestPipelines().run_combine_per_key(
+        test_pipeline or self.create_pipeline())
+
+  # TODO: enable after fixing BEAM-10507
+  # @pytest.mark.uses_java_expansion_service
+  # @pytest.mark.uses_python_expansion_service
+  def test_flatten(self, test_pipeline=None):
+    CrossLanguageTestPipelines().run_flatten(
+        test_pipeline or self.create_pipeline())
+
+  @pytest.mark.uses_java_expansion_service
+  @pytest.mark.uses_python_expansion_service
+  def test_partition(self, test_pipeline=None):
+    CrossLanguageTestPipelines().run_partition(
+        test_pipeline or self.create_pipeline())
+
+
+@unittest.skipUnless(
+    os.environ.get('EXPANSION_PORT'),
+    "EXPANSION_PORT environment var is not provided.")
+@unittest.skipIf(secretmanager is None, 'secretmanager not installed')
+class ValidateRunnerGBEKTest(unittest.TestCase):
   def setUp(self):
     if secretmanager is not None:
       self.project_id = 'apache-beam-testing'
@@ -312,77 +372,24 @@ def create_pipeline(self):
     test_pipeline.not_use_test_runner_api = True
     return test_pipeline
 
-  @pytest.mark.uses_java_expansion_service
-  @pytest.mark.uses_python_expansion_service
-  def test_prefix(self, test_pipeline=None):
-    CrossLanguageTestPipelines().run_prefix(
-        test_pipeline or self.create_pipeline())
-
-  @pytest.mark.uses_java_expansion_service
-  @pytest.mark.uses_python_expansion_service
-  def test_multi_input_output_with_sideinput(self, test_pipeline=None):
-    CrossLanguageTestPipelines().run_multi_input_output_with_sideinput(
-        test_pipeline or self.create_pipeline())
-
-  @pytest.mark.uses_java_expansion_service
-  @pytest.mark.uses_python_expansion_service
-  def test_group_by_key(self, test_pipeline=None):
-    CrossLanguageTestPipelines().run_group_by_key(
-        test_pipeline or self.create_pipeline())
-
   # This test and test_group_by_key_gbek_bad_secret validate that the gbek
   # pipeline option is correctly passed through
   @pytest.mark.uses_java_expansion_service
   @pytest.mark.uses_python_expansion_service
-  @unittest.skipIf(secretmanager is None, 'GCP dependencies are not installed')
   def test_group_by_key_gbek(self, test_pipeline=None):
     test_pipeline = test_pipeline or self.create_pipeline()
     good_secret = self.secret_option
     test_pipeline.options.view_as(SetupOptions).gbek = good_secret
     CrossLanguageTestPipelines().run_group_by_key(test_pipeline)
 
-  @pytest.mark.uses_java_expansion_service
-  @pytest.mark.uses_python_expansion_service
-  @unittest.skipIf(secretmanager is None, 'GCP dependencies are not installed')
-  def test_group_by_key_gbek_bad_secret(self, test_pipeline=None):
-    test_pipeline = test_pipeline or self.create_pipeline()
+    # Verify actually using secret manager
+    test_pipeline = self.create_pipeline()
     nonexistent_secret = 'version_name:nonexistent_secret'
     test_pipeline.options.view_as(SetupOptions).gbek = nonexistent_secret
     with self.assertRaisesRegex(
         Exception, 'Secret string must contain a valid type parameter'):
       CrossLanguageTestPipelines().run_group_by_key_no_assert(test_pipeline)
 
-  @pytest.mark.uses_java_expansion_service
-  @pytest.mark.uses_python_expansion_service
-  def test_cogroup_by_key(self, test_pipeline=None):
-    CrossLanguageTestPipelines().run_cogroup_by_key(
-        test_pipeline or self.create_pipeline())
-
-  @pytest.mark.uses_java_expansion_service
-  @pytest.mark.uses_python_expansion_service
-  def test_combine_globally(self, test_pipeline=None):
-    CrossLanguageTestPipelines().run_combine_globally(
-        test_pipeline or self.create_pipeline())
-
-  @pytest.mark.uses_java_expansion_service
-  @pytest.mark.uses_python_expansion_service
-  def test_combine_per_key(self, test_pipeline=None):
-    CrossLanguageTestPipelines().run_combine_per_key(
-        test_pipeline or self.create_pipeline())
-
-  # TODO: enable after fixing BEAM-10507
-  # @pytest.mark.uses_java_expansion_service
-  # @pytest.mark.uses_python_expansion_service
-  def test_flatten(self, test_pipeline=None):
-    CrossLanguageTestPipelines().run_flatten(
-        test_pipeline or self.create_pipeline())
-
-  @pytest.mark.uses_java_expansion_service
-  @pytest.mark.uses_python_expansion_service
-  def test_partition(self, test_pipeline=None):
-    CrossLanguageTestPipelines().run_partition(
-        test_pipeline or self.create_pipeline())
-
 
 if __name__ == '__main__':
   logging.getLogger().setLevel(logging.INFO)

From fbecc6287cc59f0e0716c1e8e1e83d2d2d237010 Mon Sep 17 00:00:00 2001
From: Danny Mccormick 
Date: Thu, 9 Oct 2025 15:01:26 -0400
Subject: [PATCH 15/24] Remove python -> java tests (see #36457)

---
 ...stCommit_XVR_PythonUsingJava_Dataflow.json |  4 -
 .../transforms/validate_runner_xlang_test.py  | 93 -------------------
 2 files changed, 97 deletions(-)
 delete mode 100644 .github/trigger_files/beam_PostCommit_XVR_PythonUsingJava_Dataflow.json

diff --git a/.github/trigger_files/beam_PostCommit_XVR_PythonUsingJava_Dataflow.json b/.github/trigger_files/beam_PostCommit_XVR_PythonUsingJava_Dataflow.json
deleted file mode 100644
index 6a55e29ae15d..000000000000
--- a/.github/trigger_files/beam_PostCommit_XVR_PythonUsingJava_Dataflow.json
+++ /dev/null
@@ -1,4 +0,0 @@
-{
-    "comment": "Modify this file in a trivial way to cause this test suite to run.",
-    "modification": 1
-}
\ No newline at end of file
diff --git a/sdks/python/apache_beam/transforms/validate_runner_xlang_test.py b/sdks/python/apache_beam/transforms/validate_runner_xlang_test.py
index 40f7478fd5d4..8e8e79648250 100644
--- a/sdks/python/apache_beam/transforms/validate_runner_xlang_test.py
+++ b/sdks/python/apache_beam/transforms/validate_runner_xlang_test.py
@@ -52,27 +52,17 @@
 
 import logging
 import os
-import random
-import string
 import typing
 import unittest
 
 import pytest
 
 import apache_beam as beam
-from apache_beam.options.pipeline_options import SetupOptions
 from apache_beam.testing.test_pipeline import TestPipeline
 from apache_beam.testing.util import assert_that
 from apache_beam.testing.util import equal_to
-from apache_beam.transforms.util import GcpSecret
-from apache_beam.transforms.util import Secret
 from apache_beam.transforms.external import ImplicitSchemaPayloadBuilder
 
-try:
-  from google.cloud import secretmanager
-except ImportError:
-  secretmanager = None  # type: ignore[assignment]
-
 TEST_PREFIX_URN = "beam:transforms:xlang:test:prefix"
 TEST_MULTI_URN = "beam:transforms:xlang:test:multi"
 TEST_GBK_URN = "beam:transforms:xlang:test:gbk"
@@ -150,24 +140,6 @@ def run_group_by_key(self, pipeline):
           | beam.Map(lambda x: "{}:{}".format(x[0], ','.join(sorted(x[1])))))
       assert_that(res, equal_to(['0:1,2', '1:3']))
 
-  def run_group_by_key_no_assert(self, pipeline):
-    """
-    Target transform - GroupByKey, with no assertion for checking errors
-    (https://beam.apache.org/documentation/programming-guide/#groupbykey)
-    Test scenario - Grouping a collection of KV to a collection of
-    KV> by key
-    Boundary conditions checked -
-     - PCollection> to external transforms
-     - PCollection>> from external transforms
-    """
-    with pipeline as p:
-      (
-          p
-          | beam.Create([(0, "1"), (0, "2"),
-                         (1, "3")], reshuffle=False).with_output_types(
-                             typing.Tuple[int, str])
-          | beam.ExternalTransform(TEST_GBK_URN, None, self.expansion_service))
-
   def run_cogroup_by_key(self, pipeline):
     """
     Target transform - CoGroupByKey
@@ -326,71 +298,6 @@ def test_partition(self, test_pipeline=None):
         test_pipeline or self.create_pipeline())
 
 
-@unittest.skipUnless(
-    os.environ.get('EXPANSION_PORT'),
-    "EXPANSION_PORT environment var is not provided.")
-@unittest.skipIf(secretmanager is None, 'secretmanager not installed')
-class ValidateRunnerGBEKTest(unittest.TestCase):
-  def setUp(self):
-    if secretmanager is not None:
-      self.project_id = 'apache-beam-testing'
-      secret_postfix = ''.join(random.choice(string.digits) for _ in range(6))
-      self.secret_id = 'gbek_secret_tests_' + secret_postfix
-      self.client = secretmanager.SecretManagerServiceClient()
-      self.project_path = f'projects/{self.project_id}'
-      self.secret_path = f'{self.project_path}/secrets/{self.secret_id}'
-      try:
-        self.client.get_secret(request={'name': self.secret_path})
-      except Exception:
-        self.client.create_secret(
-            request={
-                'parent': self.project_path,
-                'secret_id': self.secret_id,
-                'secret': {
-                    'replication': {
-                        'automatic': {}
-                    }
-                }
-            })
-        self.client.add_secret_version(
-            request={
-                'parent': self.secret_path,
-                'payload': {
-                    'data': Secret.generate_secret_bytes()
-                }
-            })
-      version_name = f'{self.secret_path}/versions/latest'
-      self.gcp_secret = GcpSecret(version_name)
-      self.secret_option = f'type:GcpSecret;version_name:{version_name}'
-
-  def tearDown(self):
-    if secretmanager is not None:
-      self.client.delete_secret(request={'name': self.secret_path})
-
-  def create_pipeline(self):
-    test_pipeline = TestPipeline()
-    test_pipeline.not_use_test_runner_api = True
-    return test_pipeline
-
-  # This test and test_group_by_key_gbek_bad_secret validate that the gbek
-  # pipeline option is correctly passed through
-  @pytest.mark.uses_java_expansion_service
-  @pytest.mark.uses_python_expansion_service
-  def test_group_by_key_gbek(self, test_pipeline=None):
-    test_pipeline = test_pipeline or self.create_pipeline()
-    good_secret = self.secret_option
-    test_pipeline.options.view_as(SetupOptions).gbek = good_secret
-    CrossLanguageTestPipelines().run_group_by_key(test_pipeline)
-
-    # Verify actually using secret manager
-    test_pipeline = self.create_pipeline()
-    nonexistent_secret = 'version_name:nonexistent_secret'
-    test_pipeline.options.view_as(SetupOptions).gbek = nonexistent_secret
-    with self.assertRaisesRegex(
-        Exception, 'Secret string must contain a valid type parameter'):
-      CrossLanguageTestPipelines().run_group_by_key_no_assert(test_pipeline)
-
-
 if __name__ == '__main__':
   logging.getLogger().setLevel(logging.INFO)
   unittest.main()

From 799e0bf9c5ed7265b2702ad75ce1d84a7e6315dd Mon Sep 17 00:00:00 2001
From: Danny Mccormick 
Date: Fri, 10 Oct 2025 11:07:47 -0400
Subject: [PATCH 16/24] Simplify to get faster repro

---
 .../org/apache/beam/gradle/BeamModulePlugin.groovy     |  2 +-
 .../sdk/util/construction/ValidateRunnerXlangTest.java | 10 +++++-----
 2 files changed, 6 insertions(+), 6 deletions(-)

diff --git a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
index c58b653b7eb9..6dff885b8565 100644
--- a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
+++ b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
@@ -2870,7 +2870,7 @@ class BeamModulePlugin implements Plugin {
           if (sdk == "Java") {
             useJUnit{ includeCategories 'org.apache.beam.sdk.testing.UsesJavaExpansionService' }
           } else if (sdk == "Python") {
-            useJUnit{ includeCategories 'org.apache.beam.sdk.testing.UsesPythonExpansionService' }
+            useJUnit{ include '**/ValidateRunnerXlangTest$GroupByKeyWithGbekTest.class' }
           } else {
             throw new GradleException("unsupported expansion service for Java validate runner tests.")
           }
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
index 425679e59a26..2eea7e207143 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
@@ -401,11 +401,11 @@ public void test() {
     }
 
     @Test
-    @Category({
-      ValidatesRunner.class,
-      UsesJavaExpansionService.class,
-      UsesPythonExpansionService.class
-    })
+    // @Category({
+    //   ValidatesRunner.class,
+    //   UsesJavaExpansionService.class,
+    //   UsesPythonExpansionService.class
+    // })
     public void testFailure() {
       thrown.expect(Exception.class);
       PipelineOptions options = TestPipeline.testingPipelineOptions();

From 89f9cc1f7339403404b6112d6b1761af382dada9 Mon Sep 17 00:00:00 2001
From: Danny McCormick 
Date: Sat, 11 Oct 2025 11:41:05 +0000
Subject: [PATCH 17/24] Get it working, need to figure out actual issue though

---
 .../sdk/util/construction/ValidateRunnerXlangTest.java    | 2 +-
 .../runners/portability/expansion_service_test.py         | 8 +++++++-
 2 files changed, 8 insertions(+), 2 deletions(-)

diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
index 2eea7e207143..b24fa6fe8867 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
@@ -329,7 +329,7 @@ public static void setUpClass() {
       } catch (IOException e) {
         gcpSecretVersionName = null;
         // return;
-        throw new RuntimeException("test");
+        throw new RuntimeException("An unexpected error occurred during operation.", e);
       }
       ProjectName projectName = ProjectName.of(PROJECT_ID);
       SecretName secretName = SecretName.of(PROJECT_ID, secretId);
diff --git a/sdks/python/apache_beam/runners/portability/expansion_service_test.py b/sdks/python/apache_beam/runners/portability/expansion_service_test.py
index 7aa2e5f16e5b..950030b1630c 100644
--- a/sdks/python/apache_beam/runners/portability/expansion_service_test.py
+++ b/sdks/python/apache_beam/runners/portability/expansion_service_test.py
@@ -137,7 +137,13 @@ def from_runner_api_parameter(
 @ptransform.PTransform.register_urn(TEST_GBK_URN, None)
 class GBKTransform(ptransform.PTransform):
   def expand(self, pcoll):
-    return pcoll | 'TestLabel' >> beam.GroupByKey()
+    def key_fn(x):
+      return x
+    return pcoll | 'TestLabel' >> beam.GroupByKey() | beam.Map(key_fn).with_output_types(
+               typing.Tuple[int, typing.Iterable[str]])
+    # Just the output types on GBK does not work yet, investigate why
+    # return pcoll | 'TestLabel' >> beam.GroupByKey().with_output_types(
+    #            typing.Tuple[int, typing.Iterable[str]])
 
   def to_runner_api_parameter(self, unused_context):
     return TEST_GBK_URN, None

From 9efbf7c25fa36d9900fd2a4a821e455b1b4ecd86 Mon Sep 17 00:00:00 2001
From: Danny McCormick 
Date: Mon, 13 Oct 2025 03:03:04 +0000
Subject: [PATCH 18/24] Fix type hinting

---
 .../runners/portability/expansion_service_test.py           | 6 +-----
 sdks/python/apache_beam/transforms/util.py                  | 5 ++++-
 2 files changed, 5 insertions(+), 6 deletions(-)

diff --git a/sdks/python/apache_beam/runners/portability/expansion_service_test.py b/sdks/python/apache_beam/runners/portability/expansion_service_test.py
index 950030b1630c..83df7630d161 100644
--- a/sdks/python/apache_beam/runners/portability/expansion_service_test.py
+++ b/sdks/python/apache_beam/runners/portability/expansion_service_test.py
@@ -139,11 +139,7 @@ class GBKTransform(ptransform.PTransform):
   def expand(self, pcoll):
     def key_fn(x):
       return x
-    return pcoll | 'TestLabel' >> beam.GroupByKey() | beam.Map(key_fn).with_output_types(
-               typing.Tuple[int, typing.Iterable[str]])
-    # Just the output types on GBK does not work yet, investigate why
-    # return pcoll | 'TestLabel' >> beam.GroupByKey().with_output_types(
-    #            typing.Tuple[int, typing.Iterable[str]])
+    return pcoll | 'TestLabel' >> beam.GroupByKey()
 
   def to_runner_api_parameter(self, unused_context):
     return TEST_GBK_URN, None
diff --git a/sdks/python/apache_beam/transforms/util.py b/sdks/python/apache_beam/transforms/util.py
index 5af9d904895a..b3b2e9699b14 100644
--- a/sdks/python/apache_beam/transforms/util.py
+++ b/sdks/python/apache_beam/transforms/util.py
@@ -565,12 +565,15 @@ def expand(self, pcoll):
 
     gbk = beam.GroupByKey()
     gbk._inside_gbek = True
+    output_type = Tuple[key_type, Iterable[value_type]]
 
     return (
         pcoll
         | beam.ParDo(_EncryptMessage(self._hmac_key, key_coder, value_coder))
         | gbk
-        | beam.ParDo(_DecryptMessage(self._hmac_key, key_coder, value_coder)))
+        | beam.ParDo(
+            _DecryptMessage(self._hmac_key, key_coder,
+                            value_coder)).with_output_types(output_type))
 
 
 class _BatchSizeEstimator(object):

From b44ed88ce1f5ddeb22c525a3300a9fabe8538cf5 Mon Sep 17 00:00:00 2001
From: Danny Mccormick 
Date: Mon, 13 Oct 2025 07:13:44 -0400
Subject: [PATCH 19/24] Clean up

---
 .../org/apache/beam/gradle/BeamModulePlugin.groovy  |  2 +-
 .../util/construction/ValidateRunnerXlangTest.java  | 13 ++++++-------
 .../runners/portability/expansion_service_test.py   |  2 --
 3 files changed, 7 insertions(+), 10 deletions(-)

diff --git a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
index 6dff885b8565..c58b653b7eb9 100644
--- a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
+++ b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
@@ -2870,7 +2870,7 @@ class BeamModulePlugin implements Plugin {
           if (sdk == "Java") {
             useJUnit{ includeCategories 'org.apache.beam.sdk.testing.UsesJavaExpansionService' }
           } else if (sdk == "Python") {
-            useJUnit{ include '**/ValidateRunnerXlangTest$GroupByKeyWithGbekTest.class' }
+            useJUnit{ includeCategories 'org.apache.beam.sdk.testing.UsesPythonExpansionService' }
           } else {
             throw new GradleException("unsupported expansion service for Java validate runner tests.")
           }
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
index b24fa6fe8867..ba69c369cfd7 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
@@ -328,8 +328,7 @@ public static void setUpClass() {
         client = SecretManagerServiceClient.create();
       } catch (IOException e) {
         gcpSecretVersionName = null;
-        // return;
-        throw new RuntimeException("An unexpected error occurred during operation.", e);
+        return;
       }
       ProjectName projectName = ProjectName.of(PROJECT_ID);
       SecretName secretName = SecretName.of(PROJECT_ID, secretId);
@@ -401,11 +400,11 @@ public void test() {
     }
 
     @Test
-    // @Category({
-    //   ValidatesRunner.class,
-    //   UsesJavaExpansionService.class,
-    //   UsesPythonExpansionService.class
-    // })
+    @Category({
+      ValidatesRunner.class,
+      UsesJavaExpansionService.class,
+      UsesPythonExpansionService.class
+    })
     public void testFailure() {
       thrown.expect(Exception.class);
       PipelineOptions options = TestPipeline.testingPipelineOptions();
diff --git a/sdks/python/apache_beam/runners/portability/expansion_service_test.py b/sdks/python/apache_beam/runners/portability/expansion_service_test.py
index 83df7630d161..7aa2e5f16e5b 100644
--- a/sdks/python/apache_beam/runners/portability/expansion_service_test.py
+++ b/sdks/python/apache_beam/runners/portability/expansion_service_test.py
@@ -137,8 +137,6 @@ def from_runner_api_parameter(
 @ptransform.PTransform.register_urn(TEST_GBK_URN, None)
 class GBKTransform(ptransform.PTransform):
   def expand(self, pcoll):
-    def key_fn(x):
-      return x
     return pcoll | 'TestLabel' >> beam.GroupByKey()
 
   def to_runner_api_parameter(self, unused_context):

From 928860545aa25faecbfd585088cdc2048a908879 Mon Sep 17 00:00:00 2001
From: Danny Mccormick 
Date: Mon, 13 Oct 2025 09:17:43 -0400
Subject: [PATCH 20/24] pipeline options tests

---
 sdks/python/apache_beam/options/pipeline_options.py | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/sdks/python/apache_beam/options/pipeline_options.py b/sdks/python/apache_beam/options/pipeline_options.py
index 0d602db7801b..00022c29cf95 100644
--- a/sdks/python/apache_beam/options/pipeline_options.py
+++ b/sdks/python/apache_beam/options/pipeline_options.py
@@ -64,6 +64,11 @@
 # that have a destination(dest) in parser.add_argument() different
 # from the flag name and whose default value is `None`.
 _FLAG_THAT_SETS_FALSE_VALUE = {'use_public_ips': 'no_use_public_ips'}
+# Set of options which should not be overriden when applying options from a
+# different language. This is relevant when using x-lang transforms where the
+# expansion service is started up with some pipeline options, and will
+# impact which options are passed in to expanded transforms' expand functions.
+_NON_OVERIDABLE_XLANG_OPTIONS = ['runner', 'experiments']
 
 
 def _static_value_provider_of(value_type):
@@ -290,7 +295,7 @@ class PipelineOptions(HasDisplayData):
   # Set of options which should not be overriden when pipeline options are
   # being merged (see from_runner_api). This primarily comes up when expanding
   # the Python expansion service
-  NON_OVERIDABLE_OPTIONS = ['runner', 'experiments']
+
   """This class and subclasses are used as containers for command line options.
 
   These classes are wrappers over the standard argparse Python module
@@ -606,7 +611,7 @@ def from_urn(key):
     if original_options is None:
       return cls(**parsed)
     for (key, value) in parsed.items():
-      if value is not None and value and key not in cls.NON_OVERIDABLE_OPTIONS:
+      if value is not None and value and key not in _NON_OVERIDABLE_XLANG_OPTIONS:
         original_options._all_options[key] = value
     return original_options
 

From 9aaa25e92abfc31bad679413dec6e5837a738a79 Mon Sep 17 00:00:00 2001
From: Danny Mccormick 
Date: Mon, 13 Oct 2025 10:09:25 -0400
Subject: [PATCH 21/24] simplify/lint

---
 sdks/python/apache_beam/options/pipeline_options.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sdks/python/apache_beam/options/pipeline_options.py b/sdks/python/apache_beam/options/pipeline_options.py
index 00022c29cf95..247daebd2c6d 100644
--- a/sdks/python/apache_beam/options/pipeline_options.py
+++ b/sdks/python/apache_beam/options/pipeline_options.py
@@ -611,7 +611,7 @@ def from_urn(key):
     if original_options is None:
       return cls(**parsed)
     for (key, value) in parsed.items():
-      if value is not None and value and key not in _NON_OVERIDABLE_XLANG_OPTIONS:
+      if value and key not in _NON_OVERIDABLE_XLANG_OPTIONS:
         original_options._all_options[key] = value
     return original_options
 

From 6c1245f3e9fcb6bcc26cbcca815719851b054efb Mon Sep 17 00:00:00 2001
From: Danny Mccormick 
Date: Mon, 13 Oct 2025 12:41:30 -0400
Subject: [PATCH 22/24] resolve gemini comments (minor)

---
 .../construction/ValidateRunnerXlangTest.java | 58 +++++++++----------
 1 file changed, 28 insertions(+), 30 deletions(-)

diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
index ba69c369cfd7..06288c07dbff 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java
@@ -323,38 +323,37 @@ public static class GroupByKeyWithGbekTest extends ValidateRunnerXlangTestBase {
     @BeforeClass
     public static void setUpClass() {
       secretId = String.format("%s-%d", SECRET_ID, new SecureRandom().nextInt(10000));
-      SecretManagerServiceClient client;
-      try {
-        client = SecretManagerServiceClient.create();
+      try (SecretManagerServiceClient client = SecretManagerServiceClient.create()) {
+        ProjectName projectName = ProjectName.of(PROJECT_ID);
+        SecretName secretName = SecretName.of(PROJECT_ID, secretId);
+
+        try {
+          client.getSecret(secretName);
+        } catch (Exception e) {
+          com.google.cloud.secretmanager.v1.Secret secret =
+              com.google.cloud.secretmanager.v1.Secret.newBuilder()
+                  .setReplication(
+                      com.google.cloud.secretmanager.v1.Replication.newBuilder()
+                          .setAutomatic(
+                              com.google.cloud.secretmanager.v1.Replication.Automatic.newBuilder()
+                                  .build())
+                          .build())
+                  .build();
+          client.createSecret(projectName, secretId, secret);
+          byte[] secretBytes = new byte[32];
+          new SecureRandom().nextBytes(secretBytes);
+          client.addSecretVersion(
+              secretName,
+              SecretPayload.newBuilder()
+                  .setData(
+                      ByteString.copyFrom(java.util.Base64.getUrlEncoder().encode(secretBytes)))
+                  .build());
+        }
+        gcpSecretVersionName = secretName.toString() + "/versions/latest";
       } catch (IOException e) {
         gcpSecretVersionName = null;
         return;
       }
-      ProjectName projectName = ProjectName.of(PROJECT_ID);
-      SecretName secretName = SecretName.of(PROJECT_ID, secretId);
-
-      try {
-        client.getSecret(secretName);
-      } catch (Exception e) {
-        com.google.cloud.secretmanager.v1.Secret secret =
-            com.google.cloud.secretmanager.v1.Secret.newBuilder()
-                .setReplication(
-                    com.google.cloud.secretmanager.v1.Replication.newBuilder()
-                        .setAutomatic(
-                            com.google.cloud.secretmanager.v1.Replication.Automatic.newBuilder()
-                                .build())
-                        .build())
-                .build();
-        client.createSecret(projectName, secretId, secret);
-        byte[] secretBytes = new byte[32];
-        new SecureRandom().nextBytes(secretBytes);
-        client.addSecretVersion(
-            secretName,
-            SecretPayload.newBuilder()
-                .setData(ByteString.copyFrom(java.util.Base64.getUrlEncoder().encode(secretBytes)))
-                .build());
-      }
-      gcpSecretVersionName = secretName.toString() + "/versions/latest";
       expansionAddr =
           String.format("localhost:%s", Integer.valueOf(System.getProperty("expansionPort")));
     }
@@ -362,8 +361,7 @@ public static void setUpClass() {
     @AfterClass
     public static void tearDownClass() {
       if (gcpSecretVersionName != null) {
-        try {
-          SecretManagerServiceClient client = SecretManagerServiceClient.create();
+        try (SecretManagerServiceClient client = SecretManagerServiceClient.create()) {
           SecretName secretName = SecretName.of(PROJECT_ID, secretId);
           client.deleteSecret(secretName);
         } catch (IOException e) {

From b4927637b4d27a43c2a5a2aaca9e1dd47443d559 Mon Sep 17 00:00:00 2001
From: Danny Mccormick 
Date: Tue, 14 Oct 2025 10:45:50 -0400
Subject: [PATCH 23/24] extra test

---
 .../options/pipeline_options_test.py          | 21 +++++++++++++++++++
 1 file changed, 21 insertions(+)

diff --git a/sdks/python/apache_beam/options/pipeline_options_test.py b/sdks/python/apache_beam/options/pipeline_options_test.py
index cd6cce204b78..7d9711ed902e 100644
--- a/sdks/python/apache_beam/options/pipeline_options_test.py
+++ b/sdks/python/apache_beam/options/pipeline_options_test.py
@@ -36,6 +36,7 @@
 from apache_beam.options.pipeline_options import ProfilingOptions
 from apache_beam.options.pipeline_options import TypeOptions
 from apache_beam.options.pipeline_options import WorkerOptions
+from apache_beam.options.pipeline_options import StandardOptions
 from apache_beam.options.pipeline_options import _BeamArgumentParser
 from apache_beam.options.pipeline_options_validator import PipelineOptionsValidator
 from apache_beam.options.value_provider import RuntimeValueProvider
@@ -308,6 +309,26 @@ def _add_argparse_args(cls, parser):
     self.assertEqual(result['test_arg_int'], 5)
     self.assertEqual(result['test_arg_none'], None)
 
+  def test_merging_options(self):
+    opts = PipelineOptions(flags=['--num_workers', '5'])
+    actual_opts = PipelineOptions.from_runner_api(opts.to_runner_api())
+    actual = actual_opts.view_as(WorkerOptions).num_workers
+    self.assertEqual(5, actual)
+
+  def test_merging_options_with_overriden_options(self):
+    opts = PipelineOptions(flags=['--num_workers', '5'])
+    base = PipelineOptions(flags=['--num_workers', '2'])
+    actual_opts = PipelineOptions.from_runner_api(opts.to_runner_api(), base)
+    actual = actual_opts.view_as(WorkerOptions).num_workers
+    self.assertEqual(5, actual)
+
+  def test_merging_options_with_overriden_runner(self):
+    opts = PipelineOptions(flags=['--runner', 'FnApiRunner'])
+    base = PipelineOptions(flags=['--runner', 'Direct'])
+    actual_opts = PipelineOptions.from_runner_api(opts.to_runner_api(), base)
+    actual = actual_opts.view_as(StandardOptions).runner
+    self.assertEqual('Direct', actual)
+
   def test_from_kwargs(self):
     class MyOptions(PipelineOptions):
       @classmethod

From 3a53427d32a901392754f3bd7a45f05d0d713468 Mon Sep 17 00:00:00 2001
From: Danny Mccormick 
Date: Tue, 14 Oct 2025 12:14:52 -0400
Subject: [PATCH 24/24] Lint: import ordering

---
 sdks/python/apache_beam/options/pipeline_options_test.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sdks/python/apache_beam/options/pipeline_options_test.py b/sdks/python/apache_beam/options/pipeline_options_test.py
index 7d9711ed902e..b9c2061744b8 100644
--- a/sdks/python/apache_beam/options/pipeline_options_test.py
+++ b/sdks/python/apache_beam/options/pipeline_options_test.py
@@ -34,9 +34,9 @@
 from apache_beam.options.pipeline_options import JobServerOptions
 from apache_beam.options.pipeline_options import PipelineOptions
 from apache_beam.options.pipeline_options import ProfilingOptions
+from apache_beam.options.pipeline_options import StandardOptions
 from apache_beam.options.pipeline_options import TypeOptions
 from apache_beam.options.pipeline_options import WorkerOptions
-from apache_beam.options.pipeline_options import StandardOptions
 from apache_beam.options.pipeline_options import _BeamArgumentParser
 from apache_beam.options.pipeline_options_validator import PipelineOptionsValidator
 from apache_beam.options.value_provider import RuntimeValueProvider