This is an automated email from the ASF dual-hosted git repository.
shunping pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 6053b781a2c [Java] Move GCP-related secret classes and tests from
sdks/java/core to google-cloud-platform-core (#40019)
6053b781a2c is described below
commit 6053b781a2cba1c274dbe7dd749bd389f7e1ab09
Author: Shunping Huang <[email protected]>
AuthorDate: Tue Sep 8 11:01:06 2026 -0400
[Java] Move GCP-related secret classes and tests from sdks/java/core to
google-cloud-platform-core (#40019)
---
.../trigger_files/beam_PostCommit_XVR_Direct.json | 2 +-
.../trigger_files/beam_PostCommit_XVR_Flink.json | 5 +-
...am_PostCommit_XVR_JavaUsingPython_Dataflow.json | 5 +-
.../trigger_files/beam_PostCommit_XVR_Spark3.json | 4 +-
.../org/apache/beam/gradle/BeamModulePlugin.groovy | 1 +
.../resources/beam/checkstyle/suppressions.xml | 8 +-
sdks/java/core/build.gradle | 13 --
.../sdk/transforms/GroupByEncryptedKeyTest.java | 153 +----------------
.../apache/beam/sdk/transforms/GroupByKeyTest.java | 112 -------------
.../java/org/apache/beam/sdk/util/SecretTest.java | 161 ------------------
.../apache/beam/sdk/util/TestSecretRegistrar.java} | 8 +-
.../util/construction/ValidateRunnerXlangTest.java | 67 +-------
.../google-cloud-platform-core/build.gradle | 6 +
.../gcp}/util/GcpHsmGeneratedSecret.java | 8 +-
.../gcp}/util/GcpHsmGeneratedSecretRegistrar.java | 3 +-
.../beam/sdk/extensions/gcp}/util/GcpSecret.java | 3 +-
.../extensions/gcp}/util/GcpSecretRegistrar.java | 3 +-
.../transforms/GcpGroupByEncryptedKeyTest.java} | 130 ++++++++------
.../gcp/transforms/GcpGroupByKeyIT.java} | 11 +-
.../gcp}/util/GcpSecretRegistrarTest.java | 3 +-
.../sdk/extensions/gcp/util/GcpSecretTest.java} | 186 ++++-----------------
sdks/java/io/components/build.gradle | 3 +
.../runners/portability/expansion_service_test.py | 6 +
23 files changed, 178 insertions(+), 723 deletions(-)
diff --git a/.github/trigger_files/beam_PostCommit_XVR_Direct.json
b/.github/trigger_files/beam_PostCommit_XVR_Direct.json
index 3b864120bf9..c04f8e34f87 100644
--- a/.github/trigger_files/beam_PostCommit_XVR_Direct.json
+++ b/.github/trigger_files/beam_PostCommit_XVR_Direct.json
@@ -1,4 +1,4 @@
{
- "modification": 1,
+ "modification": 2,
"https://github.com/apache/beam/pull/39990": "removing dead code from
FnApiDoFnRunner"
}
diff --git a/.github/trigger_files/beam_PostCommit_XVR_Flink.json
b/.github/trigger_files/beam_PostCommit_XVR_Flink.json
index a60830fe7d2..57f49e9baef 100644
--- a/.github/trigger_files/beam_PostCommit_XVR_Flink.json
+++ b/.github/trigger_files/beam_PostCommit_XVR_Flink.json
@@ -1,5 +1,4 @@
{
- "modification": 3,
- "trigger-2026-04-04": "portable_runner expand_sdf opt-in",
- "https://github.com/apache/beam/pull/39990": "removing dead code from
FnApiDoFnRunner"
+ "modification": 4,
+ "trigger-2026-04-04": "portable_runner expand_sdf opt-in"
}
diff --git
a/.github/trigger_files/beam_PostCommit_XVR_JavaUsingPython_Dataflow.json
b/.github/trigger_files/beam_PostCommit_XVR_JavaUsingPython_Dataflow.json
index 0ca37f8c8e2..7ab7bcd9a9c 100644
--- a/.github/trigger_files/beam_PostCommit_XVR_JavaUsingPython_Dataflow.json
+++ b/.github/trigger_files/beam_PostCommit_XVR_JavaUsingPython_Dataflow.json
@@ -1,5 +1,4 @@
{
- "comment": "Modify this file in a trivial way to cause this test suite to
run.",
- "modification": 1,
- "https://github.com/apache/beam/pull/39990": "removing dead code from
FnApiDoFnRunner"
+ "comment": "Modify this file in a trivial way to cause this test suite to
run.",
+ "modification": 2
}
diff --git a/.github/trigger_files/beam_PostCommit_XVR_Spark3.json
b/.github/trigger_files/beam_PostCommit_XVR_Spark3.json
index 74a5d652c7b..4fad3843d36 100644
--- a/.github/trigger_files/beam_PostCommit_XVR_Spark3.json
+++ b/.github/trigger_files/beam_PostCommit_XVR_Spark3.json
@@ -1,4 +1,4 @@
{
- "trigger-2026-07-08": "portable_runner expand_sdf opt-in 2",
- "https://github.com/apache/beam/pull/39990": "removing dead code from
FnApiDoFnRunner"
+ "modification": 1,
+ "trigger-2026-07-08": "portable_runner expand_sdf opt-in 2"
}
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 da54354f01b..be10dd4779c 100644
--- a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
+++ b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
@@ -803,6 +803,7 @@ class BeamModulePlugin implements Plugin<Project> {
grpc_google_cloud_pubsub_v1 :
"com.google.api.grpc:grpc-google-cloud-pubsub-v1", //
google_cloud_platform_libraries_bom sets version
grpc_google_common_protos :
"com.google.api.grpc:grpc-google-common-protos", //
google_cloud_platform_libraries_bom sets version
grpc_grpclb : "io.grpc:grpc-grpclb",
// google_cloud_platform_libraries_bom sets version
+ grpc_inprocess :
"io.grpc:grpc-inprocess", // google_cloud_platform_libraries_bom sets version
grpc_protobuf : "io.grpc:grpc-protobuf",
// google_cloud_platform_libraries_bom sets version
grpc_protobuf_lite :
"io.grpc:grpc-protobuf-lite", // google_cloud_platform_libraries_bom sets
version
grpc_netty : "io.grpc:grpc-netty", //
google_cloud_platform_libraries_bom sets version
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 76dffb30271..833fd30cf21 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
@@ -58,11 +58,9 @@
<!-- gRPC/protobuf exceptions -->
<!-- Non-vendored gRPC/protobuf imports are allowed for files that depend on
libraries that expose gRPC/protobuf in its public API -->
<suppress id="ForbidNonVendoredGrpcProtobuf"
files=".*sdk.*extensions.*protobuf.*" />
- <suppress id="ForbidNonVendoredGrpcProtobuf"
files=".*sdk.*core.*GcpHsmGeneratedSecret.*" />
- <suppress id="ForbidNonVendoredGrpcProtobuf"
files=".*sdk.*core.*GroupByEncryptedKeyTest.*" />
- <suppress id="ForbidNonVendoredGrpcProtobuf"
files=".*sdk.*core.*GroupByKeyTest.*" />
- <suppress id="ForbidNonVendoredGrpcProtobuf"
files=".*sdk.*core.*GroupByKeyIT.*" />
- <suppress id="ForbidNonVendoredGrpcProtobuf"
files=".*sdk.*core.*ValidateRunnerXlangTest.*" />
+ <suppress id="ForbidNonVendoredGrpcProtobuf"
files=".*sdk.*extensions.*gcp.*GcpHsmGeneratedSecret.*" />
+ <suppress id="ForbidNonVendoredGrpcProtobuf"
files=".*sdk.*extensions.*gcp.*GcpGroupByEncryptedKeyTest.*" />
+ <suppress id="ForbidNonVendoredGrpcProtobuf"
files=".*sdk.*extensions.*gcp.*GcpGroupByKeyIT.*" />
<suppress id="ForbidNonVendoredGrpcProtobuf" files=".*sdk.*extensions.*ml.*"
/>
<suppress id="ForbidNonVendoredGrpcProtobuf"
files=".*sdk.*io.*components.*ratelimiter.*" />
<suppress id="ForbidNonVendoredGrpcProtobuf" files=".*sdk.*io.*gcp.*" />
diff --git a/sdks/java/core/build.gradle b/sdks/java/core/build.gradle
index f532e9d1416..f95a8a53906 100644
--- a/sdks/java/core/build.gradle
+++ b/sdks/java/core/build.gradle
@@ -102,15 +102,6 @@ dependencies {
shadow library.java.slf4j_api
shadow library.java.snappy_java
shadow library.java.joda_time
- implementation
enforcedPlatform(library.java.google_cloud_platform_libraries_bom)
- implementation library.java.gax
- implementation library.java.google_cloud_kms
- implementation library.java.proto_google_cloud_kms_v1
- implementation library.java.google_cloud_tink
- implementation library.java.google_cloud_secret_manager
- implementation library.java.proto_google_cloud_secret_manager_v1
- implementation library.java.protobuf_java
- permitUnusedDeclared
enforcedPlatform(library.java.google_cloud_platform_libraries_bom)
provided library.java.json_org
implementation library.java.everit_json_schema
implementation library.java.guava
@@ -133,10 +124,6 @@ dependencies {
shadowTest library.java.log4j
shadowTest library.java.log4j2_api
shadowTest library.java.jamm
- shadowTest enforcedPlatform(library.java.google_cloud_platform_libraries_bom)
- shadowTest library.java.google_cloud_secret_manager
- shadowTest library.java.google_cloud_kms
- shadowTest library.java.google_cloud_tink
testRuntimeOnly library.java.slf4j_jdk14
}
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 a929d62d29d..7bcebcef5b7 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
@@ -17,18 +17,11 @@
*/
package org.apache.beam.sdk.transforms;
-import static org.junit.Assert.assertThrows;
-
-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.nio.charset.Charset;
-import java.security.SecureRandom;
+import java.nio.charset.StandardCharsets;
import java.util.Arrays;
+import java.util.Base64;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
@@ -39,14 +32,11 @@ import org.apache.beam.sdk.coders.VarIntCoder;
import org.apache.beam.sdk.testing.NeedsRunner;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
-import org.apache.beam.sdk.util.GcpHsmGeneratedSecret;
-import org.apache.beam.sdk.util.GcpSecret;
+import org.apache.beam.sdk.util.RawSecret;
import org.apache.beam.sdk.util.Secret;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.checkerframework.checker.nullness.qual.Nullable;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
@@ -100,138 +90,9 @@ public class GroupByEncryptedKeyTest implements
Serializable {
p.run();
}
- private static final String PROJECT_ID = "apache-beam-testing";
- private static final String SECRET_ID = "gbek-test";
- private static Secret gcpSecret;
- private static Secret gcpHsmGeneratedSecret;
- private static final String KEY_RING_ID = "gbek-test-key-ring";
- private static final String KEY_ID = "gbek-test-key";
-
- @BeforeClass
- public static void setup() throws IOException {
- SecretManagerServiceClient client = SecretManagerServiceClient.create();
- ProjectName projectName = ProjectName.of(PROJECT_ID);
- SecretName secretName = SecretName.of(PROJECT_ID, SECRET_ID);
-
- 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, SECRET_ID, 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());
- }
- gcpSecret = new GcpSecret(secretName.toString() + "/versions/latest");
-
- try {
- com.google.cloud.kms.v1.KeyManagementServiceClient kmsClient =
- com.google.cloud.kms.v1.KeyManagementServiceClient.create();
- String locationId = "global";
- com.google.cloud.kms.v1.KeyRingName keyRingName =
- com.google.cloud.kms.v1.KeyRingName.of(PROJECT_ID, locationId,
KEY_RING_ID);
- com.google.cloud.kms.v1.LocationName locationName =
- com.google.cloud.kms.v1.LocationName.of(PROJECT_ID, locationId);
- try {
- kmsClient.getKeyRing(keyRingName);
- } catch (Exception e) {
- kmsClient.createKeyRing(
- locationName, KEY_RING_ID,
com.google.cloud.kms.v1.KeyRing.newBuilder().build());
- }
-
- com.google.cloud.kms.v1.CryptoKeyName keyName =
- com.google.cloud.kms.v1.CryptoKeyName.of(PROJECT_ID, locationId,
KEY_RING_ID, KEY_ID);
- try {
- kmsClient.getCryptoKey(keyName);
- } catch (Exception e) {
- com.google.cloud.kms.v1.CryptoKey key =
- com.google.cloud.kms.v1.CryptoKey.newBuilder()
-
.setPurpose(com.google.cloud.kms.v1.CryptoKey.CryptoKeyPurpose.ENCRYPT_DECRYPT)
- .build();
- kmsClient.createCryptoKey(keyRingName, KEY_ID, key);
- }
- gcpHsmGeneratedSecret =
- new GcpHsmGeneratedSecret(
- PROJECT_ID,
- locationId,
- KEY_RING_ID,
- KEY_ID,
- String.format("gbek-test-job-%d", new
SecureRandom().nextInt(10000)));
- // Validate we have crypto permissions or skip these tests.
- gcpHsmGeneratedSecret.getSecretBytes();
- } catch (Exception e) {
- gcpHsmGeneratedSecret = null;
- }
- }
-
- @AfterClass
- public static void tearDown() throws IOException {
- SecretManagerServiceClient client = SecretManagerServiceClient.create();
- SecretName secretName = SecretName.of(PROJECT_ID, SECRET_ID);
- client.deleteSecret(secretName);
- }
-
- @Test
- @Category(NeedsRunner.class)
- public void testGroupByKeyGcpSecret() {
- List<KV<@Nullable String, Integer>> ungroupedPairs =
- Arrays.asList(
- KV.of(null, 3),
- KV.of("k1", 3),
- KV.of("k5", Integer.MAX_VALUE),
- KV.of("k5", Integer.MIN_VALUE),
- KV.of("k2", 66),
- KV.of("k1", 4),
- KV.of(null, 5),
- KV.of("k2", -33),
- KV.of("k3", 0));
-
- PCollection<KV<String, Integer>> input =
- p.apply(
- Create.of(ungroupedPairs)
- .withCoder(KvCoder.of(NullableCoder.of(StringUtf8Coder.of()),
VarIntCoder.of())));
-
- PCollection<KV<String, Iterable<Integer>>> output =
- input.apply(GroupByEncryptedKey.<String, Integer>create(gcpSecret));
-
- PAssert.that(output.apply("Sort", MapElements.via(new SortValues())))
- .containsInAnyOrder(
- KV.of("k1", Arrays.asList(3, 4)),
- KV.of(null, Arrays.asList(3, 5)),
- KV.of("k5", Arrays.asList(Integer.MIN_VALUE, Integer.MAX_VALUE)),
- KV.of("k2", Arrays.asList(-33, 66)),
- KV.of("k3", Arrays.asList(0)));
-
- p.run();
- }
-
- @Test
- @Category(NeedsRunner.class)
- public void testGroupByKeyGcpSecretThrows() {
- Secret gcpSecret = new GcpSecret("bad_path/versions/latest");
- p.apply(Create.of(KV.of("k1", 1)))
- .apply(GroupByEncryptedKey.<String, Integer>create(gcpSecret));
- assertThrows(RuntimeException.class, () -> p.run());
- }
-
@Test
@Category(NeedsRunner.class)
- public void testGroupByKeyGcpHsmGeneratedSecret() {
- if (gcpHsmGeneratedSecret == null) {
- return;
- }
+ public void testGroupByKeyRawSecret() {
List<KV<@Nullable String, Integer>> ungroupedPairs =
Arrays.asList(
KV.of(null, 3),
@@ -249,8 +110,12 @@ public class GroupByEncryptedKeyTest implements
Serializable {
Create.of(ungroupedPairs)
.withCoder(KvCoder.of(NullableCoder.of(StringUtf8Coder.of()),
VarIntCoder.of())));
+ // GroupByEncryptedKey expects Secret#getSecretBytes() to return
Base64-URL encoded
+ // bytes of a valid AES key (e.g. 32 bytes for AES-256).
+ byte[] secretBytes =
"test-encryption-key-secret-12345".getBytes(StandardCharsets.UTF_8);
+ Secret rawSecret = new
RawSecret(Base64.getUrlEncoder().encode(secretBytes));
PCollection<KV<String, Iterable<Integer>>> output =
- input.apply(GroupByEncryptedKey.<String,
Integer>create(gcpHsmGeneratedSecret));
+ input.apply(GroupByEncryptedKey.<String, Integer>create(rawSecret));
PAssert.that(output.apply("Sort", MapElements.via(new SortValues())))
.containsInAnyOrder(
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 3ff98d47939..5464838ad4d 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
@@ -26,18 +26,12 @@ import static org.hamcrest.Matchers.empty;
import static
org.hamcrest.collection.IsIterableContainingInAnyOrder.containsInAnyOrder;
import static org.junit.Assert.assertThrows;
-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.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.Serializable;
-import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -96,9 +90,7 @@ import org.checkerframework.checker.nullness.qual.Nullable;
import org.hamcrest.Matcher;
import org.joda.time.Duration;
import org.joda.time.Instant;
-import org.junit.AfterClass;
import org.junit.Assert;
-import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
@@ -624,110 +616,6 @@ public class GroupByKeyTest implements Serializable {
}
}
- /** Tests validating GroupByKey behaviors with the gbek flag set. */
- @RunWith(JUnit4.class)
- public static class GbekTests extends SharedTestBase {
- private static final String PROJECT_ID = "apache-beam-testing";
- private static final String SECRET_ID = "gbek-test";
- public static String gcpSecretVersionName;
- private static String secretId;
-
- @BeforeClass
- public static void setup() 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(java.util.Base64.getUrlEncoder().encode(secretBytes)))
- .build());
- }
- gcpSecretVersionName = secretName.toString() + "/versions/latest";
- }
-
- @AfterClass
- public static void tearDown() throws IOException {
- if (gcpSecretVersionName != null) {
- SecretManagerServiceClient client =
SecretManagerServiceClient.create();
- SecretName secretName = SecretName.of(PROJECT_ID, secretId);
- client.deleteSecret(secretName);
- }
- }
-
- @Test
- @Category(NeedsRunner.class)
- public void testGroupByKeyWithValidGcpSecretOption() {
- if (gcpSecretVersionName == null) {
- // Skip test if we couldn't set up secret manager
- return;
- }
- List<KV<String, Integer>> ungroupedPairs =
- Arrays.asList(
- KV.of("k1", 3),
- KV.of("k5", Integer.MAX_VALUE),
- KV.of("k5", Integer.MIN_VALUE),
- KV.of("k2", 66),
- KV.of("k1", 4),
- KV.of("k2", -33),
- KV.of("k3", 0));
-
- PCollection<KV<String, Integer>> input =
- p.apply(
- Create.of(ungroupedPairs)
- .withCoder(KvCoder.of(StringUtf8Coder.of(),
BigEndianIntegerCoder.of())));
-
- p.getOptions().setGbek(String.format("type:gcpsecret;version_name:%s",
gcpSecretVersionName));
- PCollection<KV<String, Iterable<Integer>>> output =
input.apply(GroupByKey.create());
-
- SerializableFunction<Iterable<KV<String, Iterable<Integer>>>, Void>
checker =
- containsKvs(
- kv("k1", 3, 4),
- kv("k5", Integer.MIN_VALUE, Integer.MAX_VALUE),
- kv("k2", 66, -33),
- kv("k3", 0));
- PAssert.that(output).satisfies(checker);
- PAssert.that(output).inWindow(GlobalWindow.INSTANCE).satisfies(checker);
-
- p.run();
- }
-
- @Test
- @Category(NeedsRunner.class)
- public void testGroupByKeyWithInvalidGcpSecretOption() {
- if (gcpSecretVersionName == null) {
- // Skip test if we couldn't set up secret manager
- return;
- }
-
p.getOptions().setGbek("type:gcpsecret;version_name:bad_path/versions/latest");
- p.apply(Create.of(KV.of("k1", 1))).apply(GroupByKey.create());
- assertThrows(RuntimeException.class, () -> p.run());
- }
- }
-
/** Tests validating GroupByKey behaviors with windowing. */
@RunWith(JUnit4.class)
public static class WindowTests extends SharedTestBase {
diff --git
a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/SecretTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/SecretTest.java
index 446688035a5..6810ea307b4 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/SecretTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/SecretTest.java
@@ -30,7 +30,6 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
-import org.apache.beam.sdk.util.common.ReflectHelpers;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
@@ -39,39 +38,6 @@ import org.junit.runners.JUnit4;
@RunWith(JUnit4.class)
public class SecretTest {
- @Test
- public void testParseSecretOptionWithValidGcpSecret() {
- String secretOption =
"type:gcpsecret;version_name:my_secret/versions/latest";
- Secret secret = Secret.parseSecretOption(secretOption);
- assertTrue(secret instanceof GcpSecret);
- assertEquals("my_secret/versions/latest", ((GcpSecret)
secret).getVersionName());
- assertEquals(new GcpSecret("my_secret/versions/latest"), secret);
-
- Secret secretFoo =
Secret.parseSecretOption("type:GcpSecret;version_name:foo");
- assertEquals(new GcpSecret("foo"), secretFoo);
-
- Secret secretMixedCase =
-
Secret.parseSecretOption("type:gcpsecreT;version_name:my_secret/versions/latest");
- assertEquals(new GcpSecret("my_secret/versions/latest"), secretMixedCase);
- }
-
- @Test
- public void testParseSecretOptionWithValidGcpHsmGeneratedSecret() {
- String secretOption =
-
"type:gcphsmgeneratedsecret;project_id:my-project;location_id:global;key_ring_id:my-key-ring;key_id:my-key;job_name:my-job";
- Secret secret = Secret.parseSecretOption(secretOption);
- assertTrue(secret instanceof GcpHsmGeneratedSecret);
- GcpHsmGeneratedSecret hsmSecret = (GcpHsmGeneratedSecret) secret;
- assertEquals("my-project", hsmSecret.getProjectId());
- assertEquals("global", hsmSecret.getLocationId());
- assertEquals("my-key-ring", hsmSecret.getKeyRingId());
- assertEquals("my-key", hsmSecret.getKeyId());
- assertEquals("HsmGeneratedSecret_my-job", hsmSecret.getSecretId());
- assertEquals(
- new GcpHsmGeneratedSecret("my-project", "global", "my-key-ring",
"my-key", "my-job"),
- secret);
- }
-
@Test
public void testParseSecretOptionWithMissingType() {
String secretOption = "version_name:my_secret/versions/latest";
@@ -86,24 +52,6 @@ public class SecretTest {
Exception exception =
assertThrows(IllegalArgumentException.class, () ->
Secret.parseSecretOption(secretOption));
assertTrue(exception.getMessage().contains("Invalid secret type
unsupported"));
- assertTrue(exception.getMessage().contains("GcpSecret"));
- assertTrue(exception.getMessage().contains("GoogleCloudSecretManager"));
- }
-
- @Test
- public void testParseSecretOptionWithInvalidGcpSecretParameter() {
- String secretOption = "type:gcpsecret;invalid_param:some_value";
- Exception exception =
- assertThrows(IllegalArgumentException.class, () ->
Secret.parseSecretOption(secretOption));
- assertTrue(exception.getMessage().contains("Invalid secret parameter
invalid_param"));
- }
-
- @Test
- public void testParseSecretOptionWithMissingSecretName() {
- String secretOption = "type:gcpsecreT";
- Exception exception =
- assertThrows(IllegalArgumentException.class, () ->
Secret.parseSecretOption(secretOption));
- assertTrue(exception.getMessage().contains("Secret name must be specified
in secret spec."));
}
@Test
@@ -128,21 +76,6 @@ public class SecretTest {
@Test
public void testSecretFactory() {
- String spec = "{\"name\": \"test-secret\", \"project\": \"proj\"}";
-
- Secret secretGcp = Secret.fromJson(spec, "GoogleCloudSecretManager");
- assertTrue(secretGcp instanceof GcpSecret);
- assertEquals(
- "projects/proj/secrets/test-secret/versions/latest",
- ((GcpSecret) secretGcp).getVersionName());
-
- String singleQuotedSpec = "{'name': 'test-secret', 'project': 'proj'}";
- Secret secretSingleQuoted = Secret.fromJson(singleQuotedSpec,
"GoogleCloudSecretManager");
- assertTrue(secretSingleQuoted instanceof GcpSecret);
- assertEquals(
- "projects/proj/secrets/test-secret/versions/latest",
- ((GcpSecret) secretSingleQuoted).getVersionName());
-
Secret secretRaw = Secret.fromJson("STATIC_SECRET_", null);
assertTrue(secretRaw instanceof RawSecret);
assertEquals("STATIC_SECRET_", secretRaw.getString());
@@ -152,23 +85,6 @@ public class SecretTest {
IllegalArgumentException.class, () -> Secret.fromJson("spec",
"unsupported_provider"));
assertTrue(
exception.getMessage().contains("Unsupported secret manager:
'unsupported_provider'"));
- assertTrue(exception.getMessage().contains("GoogleCloudSecretManager"));
- assertTrue(exception.getMessage().contains("GcpSecret"));
-
- // Case-insensitive secret manager lookup in fromJson
- Secret secretGcpLower = Secret.fromJson(spec, "googlecloudsecretmanager");
- assertTrue(secretGcpLower instanceof GcpSecret);
- Secret secretShortLower = Secret.fromJson(spec, "gcpsecret");
- assertTrue(secretShortLower instanceof GcpSecret);
- }
-
- @Test
- public void testSecretFactoryHsm() {
- String hsmSpec =
- "{\"project_id\": \"p\", \"location_id\": \"l\", \"key_ring_id\":
\"r\", \"key_id\": \"k\", \"job_name\": \"j\"}";
- Secret secretHsm = Secret.fromJson(hsmSpec,
"GoogleCloudHsmGeneratedSecretManager");
- assertTrue(secretHsm instanceof GcpHsmGeneratedSecret);
- assertEquals("p", ((GcpHsmGeneratedSecret) secretHsm).getProjectId());
}
@Test
@@ -188,60 +104,6 @@ public class SecretTest {
assertNotEquals(raw1, raw3);
assertFalse(raw1.equals("secret_value"));
assertFalse(raw1.equals(null));
-
- Map<String, String> spec1 = new HashMap<>();
- spec1.put("name", "sec");
- spec1.put("project", "proj");
- Map<String, String> spec2 = new HashMap<>();
- spec2.put("name", "sec");
- spec2.put("project", "proj");
- Map<String, String> spec3 = new HashMap<>();
- spec3.put("name", "other");
- spec3.put("project", "proj");
-
- GcpSecret gcp1 = GcpSecret.fromMap(spec1);
- GcpSecret gcp2 = GcpSecret.fromMap(spec2);
- GcpSecret gcp3 = GcpSecret.fromMap(spec3);
- assertEquals(gcp1, gcp2);
- assertEquals(gcp1.hashCode(), gcp2.hashCode());
- assertNotEquals(gcp1, gcp3);
- assertFalse(gcp1.equals(raw1));
- assertFalse(gcp1.equals(null));
-
- GcpHsmGeneratedSecret hsm1 = new GcpHsmGeneratedSecret("p", "l", "r", "k",
"j");
- GcpHsmGeneratedSecret hsm2 = new GcpHsmGeneratedSecret("p", "l", "r", "k",
"j");
- GcpHsmGeneratedSecret hsm3 = new GcpHsmGeneratedSecret("p", "l", "r", "k",
"other");
- assertEquals(hsm1, hsm2);
- assertEquals(hsm1.hashCode(), hsm2.hashCode());
- assertNotEquals(hsm1, hsm3);
- assertFalse(hsm1.equals(gcp1));
- assertFalse(hsm1.equals(null));
- }
-
- @Test
- public void testGcpSecretFromMapMissingSecretNameThrows() {
- Map<String, String> spec = Collections.singletonMap("project",
"my-project");
- Exception exception =
- assertThrows(IllegalArgumentException.class, () ->
GcpSecret.fromMap(spec));
- assertTrue(exception.getMessage().contains("Secret name must be
specified"));
- }
-
- @Test
- public void testGcpHsmGeneratedSecretFromMapMissingParamsThrows() {
- Map<String, String> spec = new HashMap<>();
- spec.put("project_id", "test-proj");
- spec.put("location_id", "global");
- Exception exception =
- assertThrows(NullPointerException.class, () ->
GcpHsmGeneratedSecret.fromMap(spec));
- assertTrue(
- exception
- .getMessage()
- .contains("key_ring_id must contain a valid value for keyRingId
parameter"));
- }
-
- @Test
- public void testResolveGcpProjectIdExplicit() {
- assertEquals("my-proj", GcpSecret.resolveGcpProjectId("my-proj",
"context"));
}
@Test
@@ -249,14 +111,6 @@ public class SecretTest {
RawSecret raw = new RawSecret("test_secret");
RawSecret deserializedRaw = SerializableUtils.clone(raw);
assertEquals(raw, deserializedRaw);
-
- GcpSecret gcp = new GcpSecret("projects/p/secrets/s/versions/1");
- GcpSecret deserializedGcp = SerializableUtils.clone(gcp);
- assertEquals(gcp, deserializedGcp);
-
- GcpHsmGeneratedSecret hsm = new GcpHsmGeneratedSecret("p", "l", "r", "k",
"j");
- GcpHsmGeneratedSecret deserializedHsm = SerializableUtils.clone(hsm);
- assertEquals(hsm, deserializedHsm);
}
@Test
@@ -319,19 +173,4 @@ public class SecretTest {
assertEquals(factory1, factories.get("duplicate_key"));
assertEquals(Collections.singleton("duplicate_key"), supportedTypes);
}
-
- @Test
- public void testLoadServicesOrderedDiscoversSecretRegistrars() {
- Iterable<SecretRegistrar> registrars =
- ReflectHelpers.loadServicesOrdered(SecretRegistrar.class);
- org.junit.Assert.assertNotNull(registrars);
- boolean foundGcp = false;
- for (SecretRegistrar registrar : registrars) {
- if (registrar instanceof GcpSecretRegistrar) {
- foundGcp = true;
- break;
- }
- }
- assertTrue("Expected GcpSecretRegistrar to be discovered", foundGcp);
- }
}
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpSecretRegistrar.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/TestSecretRegistrar.java
similarity index 80%
copy from
sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpSecretRegistrar.java
copy to
sdks/java/core/src/test/java/org/apache/beam/sdk/util/TestSecretRegistrar.java
index 61b31332e6d..3a3b6ae83e0 100644
---
a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpSecretRegistrar.java
+++
b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/TestSecretRegistrar.java
@@ -18,17 +18,17 @@
package org.apache.beam.sdk.util;
import com.google.auto.service.AutoService;
+import java.nio.charset.StandardCharsets;
import java.util.Map;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
-/** {@link AutoService} registrar for the {@link GcpSecret}. */
+/** Test {@link SecretRegistrar} for unit and cross-language tests. */
@AutoService(SecretRegistrar.class)
-public class GcpSecretRegistrar implements SecretRegistrar {
+public class TestSecretRegistrar implements SecretRegistrar {
@Override
public Map<String, SecretFactory> getSecretFactories() {
return ImmutableMap.of(
- "GoogleCloudSecretManager", GcpSecret::fromMap,
- "GcpSecret", GcpSecret::fromMap);
+ "TestSecret", spec -> new
RawSecret(spec.get("secret").getBytes(StandardCharsets.UTF_8)));
}
}
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 6db2105caf7..1b4fbc10df6 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
@@ -20,15 +20,11 @@ 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;
-import com.google.cloud.secretmanager.v1.SecretPayload;
-import com.google.protobuf.ByteString;
import java.io.IOException;
import java.io.Serializable;
-import java.security.SecureRandom;
+import java.nio.charset.StandardCharsets;
import java.util.Arrays;
+import java.util.Base64;
import org.apache.beam.model.pipeline.v1.ExternalTransforms;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
@@ -55,7 +51,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.junit.After;
-import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
@@ -89,7 +84,7 @@ import org.junit.runners.JUnit4;
* details.
*/
public class ValidateRunnerXlangTest {
- static class ValidateRunnerXlangTestBase extends BaseExternalTest implements
Serializable {
+ public static class ValidateRunnerXlangTestBase extends BaseExternalTest
implements Serializable {
// URNs for core cross-language transforms.
// See
https://docs.google.com/document/d/1xQp0ElIV84b8OCVz8CD2hvbiWdR8w4BvWxPTZJZA6NA
for
// further
@@ -316,61 +311,13 @@ public class ValidateRunnerXlangTest {
@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;
- private static String secretId;
@BeforeClass
public static void setUpClass() {
- secretId = String.format("%s-%d", SECRET_ID, new
SecureRandom().nextInt(10000));
- 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;
- }
expansionAddr =
String.format("localhost:%s",
Integer.valueOf(System.getProperty("expansionPort")));
}
- @AfterClass
- public static void tearDownClass() {
- if (gcpSecretVersionName != null) {
- try (SecretManagerServiceClient client =
SecretManagerServiceClient.create()) {
- SecretName secretName = SecretName.of(PROJECT_ID, secretId);
- client.deleteSecret(secretName);
- } catch (IOException e) {
- // Do nothing.
- }
- }
- }
-
@After
@Override
public void tearDown() {
@@ -385,12 +332,10 @@ public class ValidateRunnerXlangTest {
UsesPythonExpansionService.class
})
public void test() {
- if (gcpSecretVersionName == null) {
- // Skip test if we couldn't set up secret manager
- return;
- }
+ byte[] secretBytes =
"test-encryption-key-secret-12345".getBytes(StandardCharsets.UTF_8);
+ String base64Secret = Base64.getUrlEncoder().encodeToString(secretBytes);
PipelineOptions options = TestPipeline.testingPipelineOptions();
- options.setGbek(String.format("type:gcpsecret;version_name:%s",
gcpSecretVersionName));
+ options.setGbek(String.format("type:testsecret;secret:%s",
base64Secret));
Pipeline pipeline = Pipeline.create(options);
groupByKeyTest(pipeline);
PipelineResult pipelineResult = pipeline.run();
diff --git a/sdks/java/extensions/google-cloud-platform-core/build.gradle
b/sdks/java/extensions/google-cloud-platform-core/build.gradle
index 78cfe4739ec..3c6612070ad 100644
--- a/sdks/java/extensions/google-cloud-platform-core/build.gradle
+++ b/sdks/java/extensions/google-cloud-platform-core/build.gradle
@@ -59,6 +59,12 @@ dependencies {
implementation library.java.jackson_core
implementation library.java.jackson_databind
permitUnusedDeclared library.java.jackson_databind // BEAM-11761
+ implementation library.java.google_cloud_kms
+ implementation library.java.proto_google_cloud_kms_v1
+ implementation library.java.google_cloud_tink
+ implementation library.java.google_cloud_secret_manager
+ implementation library.java.proto_google_cloud_secret_manager_v1
+ implementation library.java.protobuf_java
testImplementation project(path: ":sdks:java:core", configuration:
"shadowTest")
testImplementation library.java.mockito_core
testRuntimeOnly library.java.slf4j_jdk14
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpHsmGeneratedSecret.java
b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpHsmGeneratedSecret.java
similarity index 96%
rename from
sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpHsmGeneratedSecret.java
rename to
sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpHsmGeneratedSecret.java
index 9a62bc49d3c..c1227cbff56 100644
---
a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpHsmGeneratedSecret.java
+++
b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpHsmGeneratedSecret.java
@@ -15,7 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.sdk.util;
+package org.apache.beam.sdk.extensions.gcp.util;
import com.google.api.gax.rpc.AlreadyExistsException;
import com.google.api.gax.rpc.NotFoundException;
@@ -43,15 +43,15 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
+import org.apache.beam.sdk.util.Secret;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * A {@link org.apache.beam.sdk.util.Secret} manager implementation that
generates a secret using
- * entropy from a GCP HSM key and stores it in Google Cloud Secret Manager. If
the secret already
- * exists, it will be retrieved.
+ * A {@link Secret} manager implementation that generates a secret using
entropy from a GCP HSM key
+ * and stores it in Google Cloud Secret Manager. If the secret already exists,
it will be retrieved.
*/
public class GcpHsmGeneratedSecret extends Secret {
private static final Logger LOG =
LoggerFactory.getLogger(GcpHsmGeneratedSecret.class);
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpHsmGeneratedSecretRegistrar.java
b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpHsmGeneratedSecretRegistrar.java
similarity index 93%
rename from
sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpHsmGeneratedSecretRegistrar.java
rename to
sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpHsmGeneratedSecretRegistrar.java
index 232fe7dfa83..ecdaf521ff0 100644
---
a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpHsmGeneratedSecretRegistrar.java
+++
b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpHsmGeneratedSecretRegistrar.java
@@ -15,10 +15,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.sdk.util;
+package org.apache.beam.sdk.extensions.gcp.util;
import com.google.auto.service.AutoService;
import java.util.Map;
+import org.apache.beam.sdk.util.SecretRegistrar;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
/** {@link AutoService} registrar for the {@link GcpHsmGeneratedSecret}. */
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpSecret.java
b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecret.java
similarity index 98%
rename from sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpSecret.java
rename to
sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecret.java
index 9aaa61bc3e0..08329830599 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpSecret.java
+++
b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecret.java
@@ -15,7 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.sdk.util;
+package org.apache.beam.sdk.extensions.gcp.util;
import com.google.cloud.secretmanager.v1.AccessSecretVersionResponse;
import com.google.cloud.secretmanager.v1.SecretManagerServiceClient;
@@ -29,6 +29,7 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
+import org.apache.beam.sdk.util.Secret;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Strings;
import org.checkerframework.checker.nullness.qual.Nullable;
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpSecretRegistrar.java
b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecretRegistrar.java
similarity index 93%
rename from
sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpSecretRegistrar.java
rename to
sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecretRegistrar.java
index 61b31332e6d..600ffa53391 100644
---
a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpSecretRegistrar.java
+++
b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecretRegistrar.java
@@ -15,10 +15,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.sdk.util;
+package org.apache.beam.sdk.extensions.gcp.util;
import com.google.auto.service.AutoService;
import java.util.Map;
+import org.apache.beam.sdk.util.SecretRegistrar;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
/** {@link AutoService} registrar for the {@link GcpSecret}. */
diff --git
a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByEncryptedKeyTest.java
b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/transforms/GcpGroupByEncryptedKeyTest.java
similarity index 80%
copy from
sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByEncryptedKeyTest.java
copy to
sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/transforms/GcpGroupByEncryptedKeyTest.java
index a929d62d29d..9232677f56c 100644
---
a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByEncryptedKeyTest.java
+++
b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/transforms/GcpGroupByEncryptedKeyTest.java
@@ -15,7 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.sdk.transforms;
+package org.apache.beam.sdk.extensions.gcp.transforms;
import static org.junit.Assert.assertThrows;
@@ -26,21 +26,26 @@ import com.google.cloud.secretmanager.v1.SecretPayload;
import com.google.protobuf.ByteString;
import java.io.IOException;
import java.io.Serializable;
-import java.nio.charset.Charset;
import java.security.SecureRandom;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
+import org.apache.beam.sdk.coders.BigEndianIntegerCoder;
import org.apache.beam.sdk.coders.KvCoder;
import org.apache.beam.sdk.coders.NullableCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.extensions.gcp.util.GcpHsmGeneratedSecret;
+import org.apache.beam.sdk.extensions.gcp.util.GcpSecret;
import org.apache.beam.sdk.testing.NeedsRunner;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
-import org.apache.beam.sdk.util.GcpHsmGeneratedSecret;
-import org.apache.beam.sdk.util.GcpSecret;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.GroupByEncryptedKey;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.MapElements;
+import org.apache.beam.sdk.transforms.SimpleFunction;
import org.apache.beam.sdk.util.Secret;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
@@ -53,63 +58,29 @@ import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
-/** Tests for {@link GroupByEncryptedKey}. */
+/** Tests for {@link GroupByEncryptedKey} and GBEK pipeline options using GCP
Secrets. */
@RunWith(JUnit4.class)
-public class GroupByEncryptedKeyTest implements Serializable {
+public class GcpGroupByEncryptedKeyTest implements Serializable {
@Rule public transient TestPipeline p = TestPipeline.create();
- private static class FakeSecret extends Secret {
- private final byte[] secret =
-
"YUt3STJQbXFZRnQycDV0TktDeUJTNXFZV0hoSHNHWmM".getBytes(Charset.defaultCharset());
-
- @Override
- public byte[] getSecretBytes() {
- return secret;
- }
- }
-
- @Test
- @Category(NeedsRunner.class)
- public void testGroupByKeyFakeSecret() {
- List<KV<String, Integer>> ungroupedPairs =
- Arrays.asList(
- KV.of("k1", 3),
- KV.of("k5", Integer.MAX_VALUE),
- KV.of("k5", Integer.MIN_VALUE),
- KV.of("k2", 66),
- KV.of("k1", 4),
- KV.of("k2", -33),
- KV.of("k3", 0));
-
- PCollection<KV<String, Integer>> input =
- p.apply(
- Create.of(ungroupedPairs)
- .withCoder(KvCoder.of(StringUtf8Coder.of(),
VarIntCoder.of())));
-
- PCollection<KV<String, Iterable<Integer>>> output =
- input.apply(GroupByEncryptedKey.<String, Integer>create(new
FakeSecret()));
-
- PAssert.that(output.apply("Sort", MapElements.via(new SortValues())))
- .containsInAnyOrder(
- KV.of("k1", Arrays.asList(3, 4)),
- KV.of("k5", Arrays.asList(Integer.MIN_VALUE, Integer.MAX_VALUE)),
- KV.of("k2", Arrays.asList(-33, 66)),
- KV.of("k3", Arrays.asList(0)));
-
- p.run();
- }
-
private static final String PROJECT_ID = "apache-beam-testing";
private static final String SECRET_ID = "gbek-test";
private static Secret gcpSecret;
+ private static String gcpSecretVersionName;
private static Secret gcpHsmGeneratedSecret;
private static final String KEY_RING_ID = "gbek-test-key-ring";
private static final String KEY_ID = "gbek-test-key";
@BeforeClass
public static void setup() throws IOException {
- SecretManagerServiceClient client = SecretManagerServiceClient.create();
+ SecretManagerServiceClient client;
+ try {
+ client = SecretManagerServiceClient.create();
+ } catch (IOException e) {
+ gcpSecretVersionName = null;
+ return;
+ }
ProjectName projectName = ProjectName.of(PROJECT_ID);
SecretName secretName = SecretName.of(PROJECT_ID, SECRET_ID);
@@ -134,7 +105,8 @@ public class GroupByEncryptedKeyTest implements
Serializable {
.setData(ByteString.copyFrom(java.util.Base64.getUrlEncoder().encode(secretBytes)))
.build());
}
- gcpSecret = new GcpSecret(secretName.toString() + "/versions/latest");
+ gcpSecretVersionName = secretName.toString() + "/versions/latest";
+ gcpSecret = new GcpSecret(gcpSecretVersionName);
try {
com.google.cloud.kms.v1.KeyManagementServiceClient kmsClient =
@@ -178,14 +150,19 @@ public class GroupByEncryptedKeyTest implements
Serializable {
@AfterClass
public static void tearDown() throws IOException {
- SecretManagerServiceClient client = SecretManagerServiceClient.create();
- SecretName secretName = SecretName.of(PROJECT_ID, SECRET_ID);
- client.deleteSecret(secretName);
+ if (gcpSecretVersionName != null) {
+ SecretManagerServiceClient client = SecretManagerServiceClient.create();
+ SecretName secretName = SecretName.of(PROJECT_ID, SECRET_ID);
+ client.deleteSecret(secretName);
+ }
}
@Test
@Category(NeedsRunner.class)
public void testGroupByKeyGcpSecret() {
+ if (gcpSecret == null) {
+ return;
+ }
List<KV<@Nullable String, Integer>> ungroupedPairs =
Arrays.asList(
KV.of(null, 3),
@@ -220,9 +197,9 @@ public class GroupByEncryptedKeyTest implements
Serializable {
@Test
@Category(NeedsRunner.class)
public void testGroupByKeyGcpSecretThrows() {
- Secret gcpSecret = new GcpSecret("bad_path/versions/latest");
+ Secret badGcpSecret = new GcpSecret("bad_path/versions/latest");
p.apply(Create.of(KV.of("k1", 1)))
- .apply(GroupByEncryptedKey.<String, Integer>create(gcpSecret));
+ .apply(GroupByEncryptedKey.<String, Integer>create(badGcpSecret));
assertThrows(RuntimeException.class, () -> p.run());
}
@@ -263,6 +240,51 @@ public class GroupByEncryptedKeyTest implements
Serializable {
p.run();
}
+ @Test
+ @Category(NeedsRunner.class)
+ public void testGroupByKeyWithValidGcpSecretOption() {
+ if (gcpSecretVersionName == null) {
+ return;
+ }
+ List<KV<String, Integer>> ungroupedPairs =
+ Arrays.asList(
+ KV.of("k1", 3),
+ KV.of("k5", Integer.MAX_VALUE),
+ KV.of("k5", Integer.MIN_VALUE),
+ KV.of("k2", 66),
+ KV.of("k1", 4),
+ KV.of("k2", -33),
+ KV.of("k3", 0));
+
+ PCollection<KV<String, Integer>> input =
+ p.apply(
+ Create.of(ungroupedPairs)
+ .withCoder(KvCoder.of(StringUtf8Coder.of(),
BigEndianIntegerCoder.of())));
+
+ p.getOptions().setGbek(String.format("type:gcpsecret;version_name:%s",
gcpSecretVersionName));
+ PCollection<KV<String, Iterable<Integer>>> output =
input.apply(GroupByKey.create());
+
+ PAssert.that(output.apply("Sort", MapElements.via(new SortValues())))
+ .containsInAnyOrder(
+ KV.of("k1", Arrays.asList(3, 4)),
+ KV.of("k5", Arrays.asList(Integer.MIN_VALUE, Integer.MAX_VALUE)),
+ KV.of("k2", Arrays.asList(66, -33)),
+ KV.of("k3", Arrays.asList(0)));
+
+ p.run();
+ }
+
+ @Test
+ @Category(NeedsRunner.class)
+ public void testGroupByKeyWithInvalidGcpSecretOption() {
+ if (gcpSecretVersionName == null) {
+ return;
+ }
+
p.getOptions().setGbek("type:gcpsecret;version_name:bad_path/versions/latest");
+ p.apply(Create.of(KV.of("k1", 1))).apply(GroupByKey.create());
+ assertThrows(RuntimeException.class, () -> p.run());
+ }
+
private static class SortValues
extends SimpleFunction<KV<String, Iterable<Integer>>, KV<String,
List<Integer>>> {
@Override
diff --git
a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyIT.java
b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/transforms/GcpGroupByKeyIT.java
similarity index 97%
rename from
sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyIT.java
rename to
sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/transforms/GcpGroupByKeyIT.java
index 431bdf448be..e8d3466cf09 100644
---
a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyIT.java
+++
b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/transforms/GcpGroupByKeyIT.java
@@ -15,7 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.sdk.transforms;
+package org.apache.beam.sdk.extensions.gcp.transforms;
import com.google.cloud.kms.v1.CryptoKey;
import com.google.cloud.kms.v1.CryptoKeyName;
@@ -34,10 +34,15 @@ import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.KvCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.extensions.gcp.util.GcpHsmGeneratedSecret;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
-import org.apache.beam.sdk.util.GcpHsmGeneratedSecret;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.Redistribute;
+import org.apache.beam.sdk.transforms.Sum;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.junit.AfterClass;
@@ -50,7 +55,7 @@ import org.junit.runners.JUnit4;
/** Integration test for GroupByKey transforms and some other transforms which
use GBK. */
@RunWith(JUnit4.class)
-public class GroupByKeyIT {
+public class GcpGroupByKeyIT {
@Rule public ExpectedException thrown = ExpectedException.none();
private static final String PROJECT_ID = "apache-beam-testing";
diff --git
a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/GcpSecretRegistrarTest.java
b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecretRegistrarTest.java
similarity index 96%
rename from
sdks/java/core/src/test/java/org/apache/beam/sdk/util/GcpSecretRegistrarTest.java
rename to
sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecretRegistrarTest.java
index e95483dd9bf..f8d3708f8d3 100644
---
a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/GcpSecretRegistrarTest.java
+++
b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecretRegistrarTest.java
@@ -15,7 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.sdk.util;
+package org.apache.beam.sdk.extensions.gcp.util;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.hasItems;
@@ -23,6 +23,7 @@ import static org.junit.Assert.fail;
import java.util.Map;
import java.util.ServiceLoader;
+import org.apache.beam.sdk.util.SecretRegistrar;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists;
import org.junit.Test;
import org.junit.runner.RunWith;
diff --git
a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/SecretTest.java
b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecretTest.java
similarity index 62%
copy from sdks/java/core/src/test/java/org/apache/beam/sdk/util/SecretTest.java
copy to
sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecretTest.java
index 446688035a5..c493c642dc2 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/SecretTest.java
+++
b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecretTest.java
@@ -15,29 +15,27 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.sdk.util;
+package org.apache.beam.sdk.extensions.gcp.util;
-import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
-import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
-import java.util.Set;
-import java.util.TreeSet;
-import org.apache.beam.sdk.util.common.ReflectHelpers;
+import org.apache.beam.sdk.util.RawSecret;
+import org.apache.beam.sdk.util.Secret;
+import org.apache.beam.sdk.util.SerializableUtils;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
-/** Tests for {@link org.apache.beam.sdk.util.Secret}. */
+/** Tests for {@link GcpSecret} and {@link GcpHsmGeneratedSecret}. */
@RunWith(JUnit4.class)
-public class SecretTest {
+public class GcpSecretTest {
@Test
public void testParseSecretOptionWithValidGcpSecret() {
@@ -72,24 +70,6 @@ public class SecretTest {
secret);
}
- @Test
- public void testParseSecretOptionWithMissingType() {
- String secretOption = "version_name:my_secret/versions/latest";
- Exception exception =
- assertThrows(IllegalArgumentException.class, () ->
Secret.parseSecretOption(secretOption));
- assertEquals("Secret string must contain a valid type parameter",
exception.getMessage());
- }
-
- @Test
- public void testParseSecretOptionWithUnsupportedType() {
- String secretOption =
"type:unsupported;version_name:my_secret/versions/latest";
- Exception exception =
- assertThrows(IllegalArgumentException.class, () ->
Secret.parseSecretOption(secretOption));
- assertTrue(exception.getMessage().contains("Invalid secret type
unsupported"));
- assertTrue(exception.getMessage().contains("GcpSecret"));
- assertTrue(exception.getMessage().contains("GoogleCloudSecretManager"));
- }
-
@Test
public void testParseSecretOptionWithInvalidGcpSecretParameter() {
String secretOption = "type:gcpsecret;invalid_param:some_value";
@@ -107,27 +87,7 @@ public class SecretTest {
}
@Test
- public void testRawSecretStr() {
- Secret secret = new RawSecret("STATIC_SECRET_");
- assertEquals("STATIC_SECRET_", secret.getString(true));
- assertEquals("STATIC_SECRET_", secret.getString(false));
- assertEquals("STATIC_SECRET_", secret.getString());
- assertArrayEquals("STATIC_SECRET_".getBytes(StandardCharsets.UTF_8),
secret.getBytes(true));
- assertArrayEquals("STATIC_SECRET_".getBytes(StandardCharsets.UTF_8),
secret.getBytes());
- }
-
- @Test
- public void testRawSecretBytes() {
- byte[] bytes = "STATIC_BYTES_".getBytes(StandardCharsets.UTF_8);
- Secret secret = new RawSecret(bytes);
- assertEquals("STATIC_BYTES_", secret.getString(true));
- assertEquals("STATIC_BYTES_", secret.getString());
- assertArrayEquals(bytes, secret.getBytes(true));
- assertArrayEquals(bytes, secret.getBytes());
- }
-
- @Test
- public void testSecretFactory() {
+ public void testSecretFactoryGcp() {
String spec = "{\"name\": \"test-secret\", \"project\": \"proj\"}";
Secret secretGcp = Secret.fromJson(spec, "GoogleCloudSecretManager");
@@ -143,9 +103,11 @@ public class SecretTest {
"projects/proj/secrets/test-secret/versions/latest",
((GcpSecret) secretSingleQuoted).getVersionName());
- Secret secretRaw = Secret.fromJson("STATIC_SECRET_", null);
- assertTrue(secretRaw instanceof RawSecret);
- assertEquals("STATIC_SECRET_", secretRaw.getString());
+ // Case-insensitive secret manager lookup in fromJson
+ Secret secretGcpLower = Secret.fromJson(spec, "googlecloudsecretmanager");
+ assertTrue(secretGcpLower instanceof GcpSecret);
+ Secret secretShortLower = Secret.fromJson(spec, "gcpsecret");
+ assertTrue(secretShortLower instanceof GcpSecret);
Exception exception =
assertThrows(
@@ -154,12 +116,32 @@ public class SecretTest {
exception.getMessage().contains("Unsupported secret manager:
'unsupported_provider'"));
assertTrue(exception.getMessage().contains("GoogleCloudSecretManager"));
assertTrue(exception.getMessage().contains("GcpSecret"));
+ }
- // Case-insensitive secret manager lookup in fromJson
- Secret secretGcpLower = Secret.fromJson(spec, "googlecloudsecretmanager");
- assertTrue(secretGcpLower instanceof GcpSecret);
- Secret secretShortLower = Secret.fromJson(spec, "gcpsecret");
- assertTrue(secretShortLower instanceof GcpSecret);
+ @Test
+ public void testParseSecretOptionWithUnsupportedTypeListsGcpSecrets() {
+ String secretOption =
"type:unsupported;version_name:my_secret/versions/latest";
+ Exception exception =
+ assertThrows(IllegalArgumentException.class, () ->
Secret.parseSecretOption(secretOption));
+ assertTrue(exception.getMessage().contains("Invalid secret type
unsupported"));
+ assertTrue(exception.getMessage().contains("GcpSecret"));
+ assertTrue(exception.getMessage().contains("GoogleCloudSecretManager"));
+ }
+
+ @Test
+ public void testLoadServicesOrderedDiscoversSecretRegistrars() {
+ Iterable<org.apache.beam.sdk.util.SecretRegistrar> registrars =
+ org.apache.beam.sdk.util.common.ReflectHelpers.loadServicesOrdered(
+ org.apache.beam.sdk.util.SecretRegistrar.class);
+ org.junit.Assert.assertNotNull(registrars);
+ boolean foundGcp = false;
+ for (org.apache.beam.sdk.util.SecretRegistrar registrar : registrars) {
+ if (registrar instanceof GcpSecretRegistrar) {
+ foundGcp = true;
+ break;
+ }
+ }
+ assertTrue("Expected GcpSecretRegistrar to be discovered", foundGcp);
}
@Test
@@ -171,23 +153,9 @@ public class SecretTest {
assertEquals("p", ((GcpHsmGeneratedSecret) secretHsm).getProjectId());
}
- @Test
- public void testJsonSecretWithoutSecretManagerWarning() {
- String jsonSpec = "{\"name\": \"my-secret\", \"project\": \"my-proj\"}";
- Secret secret = Secret.fromJson(jsonSpec, null);
- assertTrue(secret instanceof RawSecret);
- }
-
@Test
public void testEquality() {
RawSecret raw1 = new RawSecret("secret_value");
- RawSecret raw2 = new RawSecret("secret_value");
- RawSecret raw3 = new RawSecret("other_value");
- assertEquals(raw1, raw2);
- assertEquals(raw1.hashCode(), raw2.hashCode());
- assertNotEquals(raw1, raw3);
- assertFalse(raw1.equals("secret_value"));
- assertFalse(raw1.equals(null));
Map<String, String> spec1 = new HashMap<>();
spec1.put("name", "sec");
@@ -246,10 +214,6 @@ public class SecretTest {
@Test
public void testSerialization() {
- RawSecret raw = new RawSecret("test_secret");
- RawSecret deserializedRaw = SerializableUtils.clone(raw);
- assertEquals(raw, deserializedRaw);
-
GcpSecret gcp = new GcpSecret("projects/p/secrets/s/versions/1");
GcpSecret deserializedGcp = SerializableUtils.clone(gcp);
assertEquals(gcp, deserializedGcp);
@@ -258,80 +222,4 @@ public class SecretTest {
GcpHsmGeneratedSecret deserializedHsm = SerializableUtils.clone(hsm);
assertEquals(hsm, deserializedHsm);
}
-
- @Test
- public void testLoadSecretFactoriesNullList() {
- Map<String, SecretRegistrar.SecretFactory> factories =
Secret.loadSecretFactories(null);
- assertTrue(factories.isEmpty());
- }
-
- @Test
- public void testLoadSecretFactoriesHandlesNullRegistrarAndNullFactories() {
- SecretRegistrar nullFactoriesRegistrar = () -> null;
- Map<String, SecretRegistrar.SecretFactory> factories =
- Secret.loadSecretFactories(java.util.Arrays.asList(null,
nullFactoriesRegistrar));
- assertTrue(factories.isEmpty());
- }
-
- @Test
- public void testLoadSecretFactoriesHandlesThrowingRegistrar() {
- SecretRegistrar throwingRegistrar =
- () -> {
- throw new RuntimeException("Simulated failure in registrar");
- };
- SecretRegistrar validRegistrar =
- () -> Collections.singletonMap("valid", spec -> new RawSecret("test"));
-
- Map<String, SecretRegistrar.SecretFactory> factories =
- Secret.loadSecretFactories(java.util.Arrays.asList(throwingRegistrar,
validRegistrar));
- assertEquals(1, factories.size());
- assertTrue(factories.containsKey("valid"));
- }
-
- @Test
- public void testLoadSecretFactoriesHandlesMalformedEntries() {
- Map<String, SecretRegistrar.SecretFactory> malformedMap = new HashMap<>();
- malformedMap.put(null, spec -> new RawSecret("val"));
- malformedMap.put("", spec -> new RawSecret("val"));
- malformedMap.put(" ", spec -> new RawSecret("val"));
- malformedMap.put("null_factory", null);
- malformedMap.put("good", spec -> new RawSecret("good_val"));
-
- SecretRegistrar registrar = () -> malformedMap;
- Map<String, SecretRegistrar.SecretFactory> factories =
- Secret.loadSecretFactories(Collections.singletonList(registrar));
- assertEquals(1, factories.size());
- assertTrue(factories.containsKey("good"));
- }
-
- @Test
- public void testLoadSecretFactoriesDuplicateKeysFirstWins() {
- SecretRegistrar.SecretFactory factory1 = spec -> new RawSecret("first");
- SecretRegistrar.SecretFactory factory2 = spec -> new RawSecret("second");
-
- SecretRegistrar registrar1 = () ->
Collections.singletonMap("duplicate_key", factory1);
- SecretRegistrar registrar2 = () ->
Collections.singletonMap("DUPLICATE_KEY", factory2);
-
- Set<String> supportedTypes = new TreeSet<>();
- Map<String, SecretRegistrar.SecretFactory> factories =
- Secret.loadSecretFactories(java.util.Arrays.asList(registrar1,
registrar2), supportedTypes);
- assertEquals(1, factories.size());
- assertEquals(factory1, factories.get("duplicate_key"));
- assertEquals(Collections.singleton("duplicate_key"), supportedTypes);
- }
-
- @Test
- public void testLoadServicesOrderedDiscoversSecretRegistrars() {
- Iterable<SecretRegistrar> registrars =
- ReflectHelpers.loadServicesOrdered(SecretRegistrar.class);
- org.junit.Assert.assertNotNull(registrars);
- boolean foundGcp = false;
- for (SecretRegistrar registrar : registrars) {
- if (registrar instanceof GcpSecretRegistrar) {
- foundGcp = true;
- break;
- }
- }
- assertTrue("Expected GcpSecretRegistrar to be discovered", foundGcp);
- }
}
diff --git a/sdks/java/io/components/build.gradle
b/sdks/java/io/components/build.gradle
index 97342009834..4d3fe55033d 100644
--- a/sdks/java/io/components/build.gradle
+++ b/sdks/java/io/components/build.gradle
@@ -37,7 +37,10 @@ dependencies {
implementation library.java.vendored_guava_32_1_2_jre
implementation library.java.vendored_grpc_1_69_0
+ testImplementation
platform(library.java.google_cloud_platform_libraries_bom)
testImplementation project(path: ":sdks:java:core", configuration:
"shadowTest")
+ testImplementation library.java.grpc_inprocess
+ testRuntimeOnly library.java.grpc_netty_shaded
testImplementation library.java.guava_testlib
testImplementation library.java.junit
testImplementation library.java.hamcrest
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 b0b0b2dd2bf..67ca6501880 100644
--- a/sdks/python/apache_beam/runners/portability/expansion_service_test.py
+++ b/sdks/python/apache_beam/runners/portability/expansion_service_test.py
@@ -39,8 +39,14 @@ from apache_beam.transforms import
fully_qualified_named_transform
from apache_beam.transforms import ptransform
from apache_beam.transforms.environments import PyPIArtifactRegistry
from apache_beam.transforms.external import ImplicitSchemaPayloadBuilder
+from apache_beam.utils import secret as beam_secret
from apache_beam.utils import thread_pool_executor
+# Register TestSecret for cross-language GBEK validation tests.
+beam_secret._SECRET_TYPE_TO_SECRET_MANAGER["testsecret"] = "TestSecretManager"
+beam_secret._SECRET_CLASSES["testsecretmanager"] = (
+ lambda secret: beam_secret.RawSecret(secret.encode("utf-8")))
+
# This script provides an expansion service and example ptransforms for running
# external transform test cases. See external_test.py for details.
