> input) {
- PipelineOptions options = input.getPipeline().getOptions();
- String gbekOveride = options.getGbek();
- if (gbekOveride != null && !gbekOveride.trim().isEmpty()) {
- // Don't replace this transform if we're using GBEK since the runner may insert
- // its own GBK which doesn't perform encryption.
- this.shouldSkipReplacement = true;
- }
return input
.apply(fewKeys ? GroupByKey.createWithFewKeys() : GroupByKey.create())
.apply(
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 1f4b7535d89e..6ed0a31b3b95 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 base64 encoded 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 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(
- java.util.Base64.getUrlDecoder().decode(this.hmacKey.getSecretBytes()), "AES");
+ this.secretKeySpec = new SecretKeySpec(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/main/java/org/apache/beam/sdk/transforms/GroupByKey.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByKey.java
index 95ff73f55e74..1970b42a658a 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByKey.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByKey.java
@@ -122,7 +122,7 @@ public class GroupByKey
private GroupByKey(boolean fewKeys) {
this.fewKeys = fewKeys;
this.insideGBEK = false;
- surroundsGBEK = false;
+ this.surroundsGBEK = false;
}
/**
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/CombineTranslation.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/CombineTranslation.java
index 73a3ed84d820..1a1913d87f39 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/CombineTranslation.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/CombineTranslation.java
@@ -61,25 +61,12 @@ public String getUrn() {
return PTransformTranslation.COMBINE_PER_KEY_TRANSFORM_URN;
}
- @Override
- public String getUrn(Combine.PerKey transform) {
- if (transform.shouldSkipReplacement()) {
- return "beam:transform:combine_per_key_wrapper:v1";
- }
- return PTransformTranslation.COMBINE_PER_KEY_TRANSFORM_URN;
- }
-
@Override
public FunctionSpec translate(
AppliedPTransform> transform, SdkComponents components)
throws IOException {
- Combine.PerKey underlyingCombine = transform.getTransform();
- if (underlyingCombine.shouldSkipReplacement()) {
- // Can use null for spec for generic composite.
- return null;
- }
- if (underlyingCombine.getSideInputs().isEmpty()) {
- GlobalCombineFn combineFn = underlyingCombine.getFn();
+ if (transform.getTransform().getSideInputs().isEmpty()) {
+ GlobalCombineFn combineFn = transform.getTransform().getFn();
Coder accumulatorCoder =
extractAccumulatorCoder(combineFn, (AppliedPTransform) transform);
return FunctionSpec.newBuilder()
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 3a2fc2f08c04..ba4c50e5a41e 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 =
- "YUt3STJQbXFZRnQycDV0TktDeUJTNXFZV0hoSHNHWmM".getBytes(Charset.defaultCharset());
+ "aKwI2PmqYFt2p5tNKCyBS5qYmHhHsGZc".getBytes(Charset.defaultCharset());
@Override
public byte[] getSecretBytes() {
@@ -123,10 +123,7 @@ public static void setup() throws IOException {
byte[] secretBytes = new byte[32];
new SecureRandom().nextBytes(secretBytes);
client.addSecretVersion(
- secretName,
- SecretPayload.newBuilder()
- .setData(ByteString.copyFrom(java.util.Base64.getUrlEncoder().encode(secretBytes)))
- .build());
+ secretName, SecretPayload.newBuilder().setData(ByteString.copyFrom(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 1c8168a42a03..60477a4c242f 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,10 +82,7 @@ public static void setup() throws IOException {
byte[] secretBytes = new byte[32];
new SecureRandom().nextBytes(secretBytes);
client.addSecretVersion(
- secretName,
- SecretPayload.newBuilder()
- .setData(ByteString.copyFrom(java.util.Base64.getUrlEncoder().encode(secretBytes)))
- .build());
+ secretName, SecretPayload.newBuilder().setData(ByteString.copyFrom(secretBytes)).build());
}
gcpSecretVersionName = secretName.toString() + "/versions/latest";
}
@@ -151,7 +148,7 @@ public void testGroupByKeyWithInvalidGcpSecretOption() throws Exception {
// Redistribute depends on GBK under the hood and can have runner-specific implementations
@Test
- public void testRedistributeWithValidGcpSecretOption() throws Exception {
+ public void redistributeWithValidGcpSecretOption() throws Exception {
if (gcpSecretVersionName == null) {
// Skip test if we couldn't set up secret manager
return;
@@ -192,44 +189,4 @@ public void testRedistributeWithInvalidGcpSecretOption() throws Exception {
thrown.expect(RuntimeException.class);
p.run();
}
-
- // Combine.PerKey depends on GBK under the hood, but can be overriden by a runner. This can
- // fail unless it is handled specially, so we should test it specifically
- @Test
- public void testCombinePerKeyWithValidGcpSecretOption() throws Exception {
- 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));
- Pipeline p = Pipeline.create(options);
-
- List> ungroupedPairs =
- Arrays.asList(
- KV.of("k1", 3), KV.of("k2", 66), KV.of("k1", 4), KV.of("k2", -33), KV.of("k3", 0));
- List> sums = Arrays.asList(KV.of("k1", 7), KV.of("k2", 33), KV.of("k3", 0));
- PCollection> input =
- p.apply(
- Create.of(ungroupedPairs)
- .withCoder(KvCoder.of(StringUtf8Coder.of(), VarIntCoder.of())));
- PCollection> output = input.apply(Combine.perKey(Sum.ofIntegers()));
- PAssert.that(output).containsInAnyOrder(sums);
-
- p.run();
- }
-
- @Test
- public void testCombinePerKeyWithInvalidGcpSecretOption() throws Exception {
- if (gcpSecretVersionName == null) {
- // Skip test if we couldn't set up secret manager
- return;
- }
- PipelineOptions options = TestPipeline.testingPipelineOptions();
- options.setGbek("type:gcpsecret;version_name:bad_path/versions/latest");
- Pipeline p = Pipeline.create(options);
- p.apply(Create.of(KV.of("k1", 1))).apply(Combine.perKey(Sum.ofIntegers()));
- thrown.expect(RuntimeException.class);
- p.run();
- }
}
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 d9a3e3ed20d4..cdc405c94cce 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
@@ -123,51 +123,49 @@ public abstract static class SharedTestBase {
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);
- }
- }
+ // @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(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);
+ // }
+ // }
}
/** Tests validating basic {@link GroupByKey} scenarios. */
@@ -674,54 +672,54 @@ public void testLargeKeys100MB() throws Exception {
runLargeKeysTest(p, 100 << 20);
}
- @Test
- @Category(NeedsRunner.class)
- public void testGroupByKeyWithValidGcpSecretOption() {
- if (gcpSecretVersionName == null) {
- // Skip test if we couldn't set up secret manager
- return;
- }
- List> 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> 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>> output = input.apply(GroupByKey.create());
-
- SerializableFunction>>, 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());
- }
+ // @Test
+ // @Category(NeedsRunner.class)
+ // public void testGroupByKeyWithValidGcpSecretOption() {
+ // if (gcpSecretVersionName == null) {
+ // // Skip test if we couldn't set up secret manager
+ // return;
+ // }
+ // List> 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> 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>> output = input.apply(GroupByKey.create());
+
+ // SerializableFunction>>, 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. */
diff --git a/sdks/python/apache_beam/options/pipeline_options.py b/sdks/python/apache_beam/options/pipeline_options.py
index 3fc5151156f1..2d3b8b49d8d7 100644
--- a/sdks/python/apache_beam/options/pipeline_options.py
+++ b/sdks/python/apache_beam/options/pipeline_options.py
@@ -1726,12 +1726,8 @@ 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. The secret should be a url safe base64 encoded '
- '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. ' # pylint: disable=line-too-long
- 'Runners with this behavior include the Dataflow, '
- 'Flink, and Spark runners. The option should be '
+ 'not unencrypted. 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'))
diff --git a/settings.gradle.kts b/settings.gradle.kts
index 72c5194ec93d..562d9aba804d 100644
--- a/settings.gradle.kts
+++ b/settings.gradle.kts
@@ -56,11 +56,11 @@ buildCache {
url = uri("https://beam-cache.apache.org/cache/")
isAllowUntrustedServer = false
credentials {
- username = System.getenv("GRADLE_ENTERPRISE_CACHE_USERNAME")
- password = System.getenv("GRADLE_ENTERPRISE_CACHE_PASSWORD")
+ username = System.getenv("GRADLE_ENTERPRISE_CACHE_USERNAME_DISABLED")
+ password = System.getenv("GRADLE_ENTERPRISE_CACHE_PASSWORD_DISABLED")
}
- isEnabled = !System.getenv("GRADLE_ENTERPRISE_CACHE_USERNAME").isNullOrBlank()
- isPush = isCi && !System.getenv("GRADLE_ENTERPRISE_CACHE_USERNAME").isNullOrBlank()
+ isEnabled = !System.getenv("GRADLE_ENTERPRISE_CACHE_USERNAME_DISABLED").isNullOrBlank()
+ isPush = isCi && !System.getenv("GRADLE_ENTERPRISE_CACHE_USERNAME_DISABLED").isNullOrBlank()
}
}