Revert "Add pipeline option to force GBEK (Java) (#36346)"

This reverts commit c8df4da.

Author: damccorm

.github/trigger_files/beam_PostCommit_Java.json

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

.github/trigger_files/beam_PostCommit_Java_DataflowV1.json

@@ -3,6 +3,6 @@
   "https://github.com/apache/beam/pull/34902": "Introducing OutputBuilder",
   "https://github.com/apache/beam/pull/35177": "Introducing WindowedValueReceiver to runners",
     "comment": "Modify this file in a trivial way to cause this test suite to run",
-    "modification": 2,
+    "modification": 1,
   "https://github.com/apache/beam/pull/35159": "moving WindowedValue and making an interface"
 }

.github/trigger_files/beam_PostCommit_Java_DataflowV2.json

@@ -1,4 +1,8 @@
 {
-  "modification": 4,
+  "https://github.com/apache/beam/pull/36138": "Cleanly separating v1 worker and v2 sdk harness container image handling",
+  "https://github.com/apache/beam/pull/34902": "Introducing OutputBuilder",
+  "https://github.com/apache/beam/pull/35177": "Introducing WindowedValueReceiver to runners",
+    "comment": "Modify this file in a trivial way to cause this test suite to run",
+    "modification": 3,
   "https://github.com/apache/beam/pull/35159": "moving WindowedValue and making an interface"
 }

runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/internal/DataflowGroupByKey.java

@@ -25,21 +25,17 @@
 import org.apache.beam.sdk.coders.Coder.NonDeterministicException;
 import org.apache.beam.sdk.coders.IterableCoder;
 import org.apache.beam.sdk.coders.KvCoder;
-import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.runners.AppliedPTransform;
-import org.apache.beam.sdk.transforms.GroupByEncryptedKey;
 import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.transforms.windowing.DefaultTrigger;
 import org.apache.beam.sdk.transforms.windowing.GlobalWindows;
-import org.apache.beam.sdk.util.Secret;
 import org.apache.beam.sdk.util.construction.PTransformTranslation;
 import org.apache.beam.sdk.util.construction.SdkComponents;
 import org.apache.beam.sdk.util.construction.TransformPayloadTranslatorRegistrar;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PCollection.IsBounded;
 import org.apache.beam.sdk.values.WindowingStrategy;
-import org.checkerframework.checker.nullness.qual.Nullable;
 
 /**
  * Specialized implementation of {@code GroupByKey} for translating Redistribute transform into
@@ -50,13 +46,9 @@ public class DataflowGroupByKey<K, V>
 
   // Plumbed from Redistribute transform.
   private final boolean allowDuplicates;
-  private boolean insideGBEK;
-  private boolean surroundsGBEK;
 
   private DataflowGroupByKey(boolean allowDuplicates) {
     this.allowDuplicates = allowDuplicates;
-    this.insideGBEK = false;
-    this.surroundsGBEK = false;
   }
 
   /**
@@ -87,22 +79,6 @@ public boolean allowDuplicates() {
     return allowDuplicates;
   }
 
-  /**
-   * For Beam internal use only. Tells runner that this is an inner GBK inside of a
-   * GroupByEncryptedKey
-   */
-  public void setInsideGBEK() {
-    this.insideGBEK = true;
-  }
-
-  /**
-   * For Beam internal use only. Tells runner that this is a GBK wrapped around of a
-   * GroupByEncryptedKey
-   */
-  public boolean surroundsGBEK() {
-    return this.surroundsGBEK;
-  }
-
   /////////////////////////////////////////////////////////////////////////////
 
   public static void applicableTo(PCollection<?> input) {
@@ -141,20 +117,6 @@ public PCollection<KV<K, Iterable<V>>> expand(PCollection<KV<K, V>> input) {
           "the keyCoder of a DataflowGroupByKey must be deterministic", e);
     }
 
-    PipelineOptions options = input.getPipeline().getOptions();
-    String gbekOveride = options.getGbek();
-    if (!this.insideGBEK && gbekOveride != null && !gbekOveride.trim().isEmpty()) {
-      this.surroundsGBEK = true;
-      Secret hmacSecret = Secret.parseSecretOption(gbekOveride);
-      DataflowGroupByKey<byte[], KV<byte[], byte[]>> gbk = DataflowGroupByKey.create();
-      if (this.allowDuplicates) {
-        gbk = DataflowGroupByKey.createWithAllowDuplicates();
-      }
-      gbk.setInsideGBEK();
-      GroupByEncryptedKey<K, V> gbek = GroupByEncryptedKey.createWithCustomGbk(hmacSecret, gbk);
-      return input.apply(gbek);
-    }
-
     // This primitive operation groups by the combination of key and window,
     // merging windows as needed, using the windows assigned to the
     // key/value input elements and the window merge operation of the
@@ -209,22 +171,10 @@ public String getUrn() {
       return PTransformTranslation.GROUP_BY_KEY_TRANSFORM_URN;
     }
 
-    @Override
-    public String getUrn(DataflowGroupByKey<?, ?> transform) {
-      if (transform.surroundsGBEK()) {
-        return PTransformTranslation.GROUP_BY_KEY_WRAPPER_TRANSFORM_URN;
-      }
-      return PTransformTranslation.GROUP_BY_KEY_TRANSFORM_URN;
-    }
-
     @Override
     @SuppressWarnings("nullness")
-    public RunnerApi.@Nullable FunctionSpec translate(
+    public RunnerApi.FunctionSpec translate(
         AppliedPTransform<?, ?, DataflowGroupByKey<?, ?>> transform, SdkComponents components) {
-      if (transform.getTransform().surroundsGBEK()) {
-        // Can use null for spec for empty composite.
-        return null;
-      }
       return RunnerApi.FunctionSpec.newBuilder().setUrn(getUrn(transform.getTransform())).build();
     }
   }

sdks/java/build-tools/src/main/resources/beam/checkstyle/suppressions.xml

@@ -58,8 +58,6 @@
   <!-- 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.*GroupByEncryptedKeyTest.*" />
-  <suppress id="ForbidNonVendoredGrpcProtobuf" files=".*sdk.*core.*GroupByKeyTest.*" />
-  <suppress id="ForbidNonVendoredGrpcProtobuf" files=".*sdk.*core.*GroupByKeyIT.*" />
   <suppress id="ForbidNonVendoredGrpcProtobuf" files=".*sdk.*extensions.*ml.*" />
   <suppress id="ForbidNonVendoredGrpcProtobuf" files=".*sdk.*io.*gcp.*" />
   <suppress id="ForbidNonVendoredGrpcProtobuf" files=".*sdk.*io.*googleads.*DummyRateLimitPolicy\.java" />

sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java

@@ -37,7 +37,6 @@
 import org.apache.beam.sdk.util.ReleaseInfo;
 import org.apache.beam.sdk.util.common.ReflectHelpers;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects;
-import org.checkerframework.checker.nullness.qual.Nullable;
 import org.joda.time.DateTimeUtils;
 import org.joda.time.DateTimeZone;
 import org.joda.time.format.DateTimeFormat;
@@ -414,40 +413,6 @@ public Long create(PipelineOptions options) {
 
   void setUserAgent(String userAgent);
 
-  /**
-   * A string defining whether GroupByKey transforms should be replaced by GroupByEncryptedKey
-   *
-   * <p>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:
-   *
-   * <pre><code>
-   * --gbek=type:<secret_type>;<secret_param>:<value>
-   * </code></pre>
-   *
-   * for example:
-   *
-   * <pre><code>
-   * --gbek=type:GcpSecret;version_name:my_secret/versions/latest"
-   * </code></pre>
-   *
-   * All variables should use snake case to allow consistency across languages.
-   */
-  @Description(
-      "When set, will replace all GroupByKey transforms in the pipeline the option. 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:"
-          + " --gbek=type:<secret_type>;<secret_param>:<value>, for example "
-          + " --gbek=type:GcpSecret;version_name:my_secret/versions/latest. All variables "
-          + " should use snake case to allow consistency across languages.")
-  @Nullable
-  String getGbek();
-
-  void setGbek(String gbek);
-
   /**
    * Returns a user agent string constructed from {@link ReleaseInfo#getName()} and {@link
    * ReleaseInfo#getVersion()}, in the format {@code [name]/[version]}.

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByEncryptedKey.java

@@ -53,19 +53,9 @@ public class GroupByEncryptedKey<K, V>
     extends PTransform<PCollection<KV<K, V>>, PCollection<KV<K, Iterable<V>>>> {
 
   private final Secret hmacKey;
-  private final PTransform<
-          PCollection<KV<byte[], KV<byte[], byte[]>>>,
-          PCollection<KV<byte[], Iterable<KV<byte[], byte[]>>>>>
-      gbk;
 
-  private GroupByEncryptedKey(
-      Secret hmacKey,
-      PTransform<
-              PCollection<KV<byte[], KV<byte[], byte[]>>>,
-              PCollection<KV<byte[], Iterable<KV<byte[], byte[]>>>>>
-          gbk) {
+  private GroupByEncryptedKey(Secret hmacKey) {
     this.hmacKey = hmacKey;
-    this.gbk = gbk;
   }
 
   /**
@@ -77,25 +67,7 @@ private GroupByEncryptedKey(
    * @return A {@link GroupByEncryptedKey} transform.
    */
   public static <K, V> GroupByEncryptedKey<K, V> create(Secret hmacKey) {
-    return new GroupByEncryptedKey<>(hmacKey, GroupByKey.create());
-  }
-
-  /**
-   * Creates a {@link GroupByEncryptedKey} transform with a custom GBK in the middle.
-   *
-   * @param hmacKey The {@link Secret} key to use for encryption.
-   * @param gbk The custom GBK transform to use in the middle of the GBEK.
-   * @param <K> The type of the keys in the input PCollection.
-   * @param <V> The type of the values in the input PCollection.
-   * @return A {@link GroupByEncryptedKey} transform.
-   */
-  public static <K, V> GroupByEncryptedKey<K, V> createWithCustomGbk(
-      Secret hmacKey,
-      PTransform<
-              PCollection<KV<byte[], KV<byte[], byte[]>>>,
-              PCollection<KV<byte[], Iterable<KV<byte[], byte[]>>>>>
-          gbk) {
-    return new GroupByEncryptedKey<>(hmacKey, gbk);
+    return new GroupByEncryptedKey<>(hmacKey);
   }
 
   @Override
@@ -121,7 +93,7 @@ public PCollection<KV<K, Iterable<V>>> expand(PCollection<KV<K, V>> input) {
             .apply(
                 "EncryptMessage",
                 ParDo.of(new EncryptMessage<>(this.hmacKey, keyCoder, valueCoder)))
-            .apply(this.gbk);
+            .apply(GroupByKey.create());
 
     return grouped
         .apply("DecryptMessage", ParDo.of(new DecryptMessage<>(this.hmacKey, keyCoder, valueCoder)))

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByKey.java

@@ -32,7 +32,6 @@
 import org.apache.beam.sdk.transforms.windowing.TimestampCombiner;
 import org.apache.beam.sdk.transforms.windowing.Window;
 import org.apache.beam.sdk.transforms.windowing.WindowFn;
-import org.apache.beam.sdk.util.Secret;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PCollection.IsBounded;
@@ -116,13 +115,9 @@ public class GroupByKey<K, V>
     extends PTransform<PCollection<KV<K, V>>, PCollection<KV<K, Iterable<V>>>> {
 
   private final boolean fewKeys;
-  private boolean insideGBEK;
-  private boolean surroundsGBEK;
 
   private GroupByKey(boolean fewKeys) {
     this.fewKeys = fewKeys;
-    this.insideGBEK = false;
-    surroundsGBEK = false;
   }
 
   /**
@@ -153,21 +148,6 @@ public boolean fewKeys() {
     return fewKeys;
   }
 
-  /**
-   * For Beam internal use only. Tells runner that this is an inner GBK inside a GroupByEncryptedKey
-   */
-  public void setInsideGBEK() {
-    this.insideGBEK = true;
-  }
-
-  /**
-   * For Beam internal use only. Tells runner that this is a GBK wrapped around of a
-   * GroupByEncryptedKey
-   */
-  public boolean surroundsGBEK() {
-    return this.surroundsGBEK;
-  }
-
   /////////////////////////////////////////////////////////////////////////////
 
   public static void applicableTo(PCollection<?> input) {
@@ -264,20 +244,6 @@ public PCollection<KV<K, Iterable<V>>> expand(PCollection<KV<K, V>> input) {
       throw new IllegalStateException("the keyCoder of a GroupByKey must be deterministic", e);
     }
 
-    PipelineOptions options = input.getPipeline().getOptions();
-    String gbekOveride = options.getGbek();
-    if (!this.insideGBEK && gbekOveride != null && !gbekOveride.trim().isEmpty()) {
-      this.surroundsGBEK = true;
-      Secret hmacSecret = Secret.parseSecretOption(gbekOveride);
-      GroupByKey<byte[], KV<byte[], byte[]>> gbk = GroupByKey.create();
-      if (this.fewKeys) {
-        gbk = GroupByKey.createWithFewKeys();
-      }
-      gbk.setInsideGBEK();
-      GroupByEncryptedKey<K, V> gbek = GroupByEncryptedKey.createWithCustomGbk(hmacSecret, gbk);
-      return input.apply(gbek);
-    }
-
     // This primitive operation groups by the combination of key and window,
     // merging windows as needed, using the windows assigned to the
     // key/value input elements and the window merge operation of the

sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpSecret.java

@@ -55,13 +55,4 @@ public byte[] getSecretBytes() {
       throw new RuntimeException("Failed to retrieve secret bytes", e);
     }
   }
-
-  /**
-   * Returns the version name of the secret.
-   *
-   * @return The version name as a String.
-   */
-  public String getVersionName() {
-    return versionName;
-  }
 }

sdks/java/core/src/main/java/org/apache/beam/sdk/util/Secret.java

@@ -18,11 +18,6 @@
 package org.apache.beam.sdk.util;
 
 import java.io.Serializable;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
 
 /**
  * A secret management interface used for handling sensitive data.
@@ -38,48 +33,4 @@ public interface Secret extends Serializable {
    * @return The secret as a byte array.
    */
   byte[] getSecretBytes();
-
-  static Secret parseSecretOption(String secretOption) {
-    Map<String, String> paramMap = new HashMap<>();
-    for (String param : secretOption.split(";", -1)) {
-      String[] parts = param.split(":", 2);
-      if (parts.length == 2) {
-        paramMap.put(parts[0], parts[1]);
-      }
-    }
-
-    if (!paramMap.containsKey("type")) {
-      throw new RuntimeException("Secret string must contain a valid type parameter");
-    }
-
-    String secretType = paramMap.get("type");
-    paramMap.remove("type");
-
-    if (secretType == null) {
-      throw new RuntimeException("Secret string must contain a valid value for type parameter");
-    }
-
-    switch (secretType.toLowerCase()) {
-      case "gcpsecret":
-        Set<String> gcpSecretParams = new HashSet<>(Arrays.asList("version_name"));
-        for (String paramName : paramMap.keySet()) {
-          if (!gcpSecretParams.contains(paramName)) {
-            throw new RuntimeException(
-                String.format(
-                    "Invalid secret parameter %s, GcpSecret only supports the following parameters: %s",
-                    paramName, gcpSecretParams));
-          }
-        }
-        String versionName = paramMap.get("version_name");
-        if (versionName == null) {
-          throw new RuntimeException(
-              "version_name must contain a valid value for versionName parameter");
-        }
-        return new GcpSecret(versionName);
-      default:
-        throw new RuntimeException(
-            String.format(
-                "Invalid secret type %s, currently only GcpSecret is supported", secretType));
-    }
-  }
 }