Added SCRAM-SHA-512 support to dataflow kafka-to-bigquery

v2/kafka-common/src/main/java/com/google/cloud/teleport/v2/kafka/options/KafkaReadOptions.java

@@ -98,14 +98,16 @@ final class Offset {
             KafkaAuthenticationMethod.APPLICATION_DEFAULT_CREDENTIALS),
         @TemplateParameter.TemplateEnumOption(KafkaAuthenticationMethod.SASL_PLAIN),
         @TemplateParameter.TemplateEnumOption(KafkaAuthenticationMethod.TLS),
+        @TemplateParameter.TemplateEnumOption(KafkaAuthenticationMethod.SASL_SCRAM_512),
         @TemplateParameter.TemplateEnumOption(KafkaAuthenticationMethod.NONE),
       },
       description = "Kafka Source Authentication Mode",
       helpText =
           "The mode of authentication to use with the Kafka cluster. "
               + "Use `KafkaAuthenticationMethod.NONE` for no authentication, `KafkaAuthenticationMethod.SASL_PLAIN` for SASL/PLAIN username and password, "
-              + "and `KafkaAuthenticationMethod.TLS` for certificate-based authentication. `KafkaAuthenticationMethod.APPLICATION_DEFAULT_CREDENTIALS` "
-              + "should be used only for Google Cloud Apache Kafka for BigQuery cluster, it allows to authenticate using application default credentials.")
+              + "`KafkaAuthenticationMethod.SASL_SCRAM_512` for SASL_SCRAM_512 authentication and `KafkaAuthenticationMethod.TLS` for certificate-based "
+              + "authentication. `KafkaAuthenticationMethod.APPLICATION_DEFAULT_CREDENTIALS` should be used only for Google Cloud Apache Kafka for BigQuery cluster, "
+              + "it allows to authenticate using application default credentials.")
   @Default.String(KafkaAuthenticationMethod.SASL_PLAIN)
   String getKafkaReadAuthenticationMode();
 
@@ -215,4 +217,63 @@ final class Offset {
   String getKafkaReadKeyPasswordSecretId();
 
   void setKafkaReadKeyPasswordSecretId(String sourceKeyPasswordSecretId);
+
+  @TemplateParameter.Text(
+      order = 13,
+      groupName = "Source",
+      parentName = "kafkaReadAuthenticationMode",
+      parentTriggerValues = {KafkaAuthenticationMethod.SASL_SCRAM_512},
+      optional = true,
+      description = "Secret Version ID For Kafka SASL_SCRAM Username",
+      helpText =
+          "The Google Cloud Secret Manager secret ID that contains the Kafka username "
+              + "to use with `SASL_SCRAM` authentication.",
+      example = "projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>")
+  String getKafkaReadSaslScramUsernameSecretId();
+
+  void setKafkaReadSaslScramUsernameSecretId(String value);
+
+  @TemplateParameter.Text(
+      order = 14,
+      groupName = "Source",
+      parentName = "kafkaReadAuthenticationMode",
+      parentTriggerValues = KafkaAuthenticationMethod.SASL_SCRAM_512,
+      optional = true,
+      description = "Secret Version ID For Kafka SASL_SCRAM Password",
+      helpText =
+          "The Google Cloud Secret Manager secret ID that contains the Kafka password to use with `SASL_SCRAM` authentication.",
+      example = "projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>")
+  String getKafkaReadSaslScramPasswordSecretId();
+
+  void setKafkaReadSaslScramPasswordSecretId(String value);
+
+  @TemplateParameter.GcsReadFile(
+      order = 15,
+      optional = true,
+      groupName = "Source",
+      parentName = "kafkaReadAuthenticationMode",
+      parentTriggerValues = {KafkaAuthenticationMethod.SASL_SCRAM_512},
+      description = "Truststore File Location",
+      helpText =
+          "The Google Cloud Storage path to the Java TrustStore (JKS) file that contains"
+              + " the trusted certificates to use to verify the identity of the Kafka broker.")
+  String getKafkaReadSaslScramTruststoreLocation();
+
+  void setKafkaReadSaslScramTruststoreLocation(String sourceSaslScramTruststoreLocation);
+
+  @TemplateParameter.Text(
+      order = 16,
+      optional = true,
+      groupName = "Source",
+      parentName = "kafkaReadAuthenticationMode",
+      parentTriggerValues = {KafkaAuthenticationMethod.SASL_SCRAM_512},
+      description = "Secret Version ID for Truststore Password",
+      helpText =
+          "The Google Cloud Secret Manager secret ID that contains the password to "
+              + "use to access the Java TrustStore (JKS) file for Kafka SASL_SCRAM authentication",
+      example = "projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>")
+  String getKafkaReadSaslScramTruststorePasswordSecretId();
+
+  void setKafkaReadSaslScramTruststorePasswordSecretId(
+      String sourceSaslScramTruststorePasswordSecretId);
 }

v2/kafka-common/src/main/java/com/google/cloud/teleport/v2/kafka/options/KafkaWriteOptions.java

@@ -43,12 +43,14 @@ public interface KafkaWriteOptions extends PipelineOptions {
         @TemplateParameter.TemplateEnumOption(
             KafkaAuthenticationMethod.APPLICATION_DEFAULT_CREDENTIALS),
         @TemplateParameter.TemplateEnumOption(KafkaAuthenticationMethod.SASL_PLAIN),
+        @TemplateParameter.TemplateEnumOption(KafkaAuthenticationMethod.SASL_SCRAM_512),
         @TemplateParameter.TemplateEnumOption(KafkaAuthenticationMethod.TLS),
         @TemplateParameter.TemplateEnumOption(KafkaAuthenticationMethod.NONE),
       },
       helpText =
           "The mode of authentication to use with the Kafka cluster. "
-              + "Use NONE for no authentication, SASL_PLAIN for SASL/PLAIN username and password, and"
+              + "Use NONE for no authentication, SASL_PLAIN for SASL/PLAIN username and password, "
+              + " SASL_SCRAM_512 for SASL_SCRAM_512 based authentication and"
               + " TLS for certificate-based authentication.")
   @Default.String(KafkaAuthenticationMethod.APPLICATION_DEFAULT_CREDENTIALS)
   String getKafkaWriteAuthenticationMethod();
@@ -160,4 +162,63 @@ public interface KafkaWriteOptions extends PipelineOptions {
   String getKafkaWriteKeyPasswordSecretId();
 
   void setKafkaWriteKeyPasswordSecretId(String destinationKeyPasswordSecretId);
+
+  @TemplateParameter.Text(
+      order = 13,
+      groupName = "Destination",
+      parentName = "kafkaReadAuthenticationMode",
+      parentTriggerValues = {KafkaAuthenticationMethod.SASL_SCRAM_512},
+      optional = true,
+      description = "Secret Version ID For Kafka SASL_SCRAM Username",
+      helpText =
+          "The Google Cloud Secret Manager secret ID that contains the Kafka username "
+              + "to use with `SASL_SCRAM` authentication.",
+      example = "projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>")
+  String getKafkaReadSaslScramUsernameSecretId();
+
+  void setKafkaReadSaslScramUsernameSecretId(String value);
+
+  @TemplateParameter.Text(
+      order = 14,
+      groupName = "Destination",
+      parentName = "kafkaReadAuthenticationMode",
+      parentTriggerValues = KafkaAuthenticationMethod.SASL_SCRAM_512,
+      optional = true,
+      description = "Secret Version ID For Kafka SASL_SCRAM Password",
+      helpText =
+          "The Google Cloud Secret Manager secret ID that contains the Kafka password to use with `SASL_SCRAM` authentication.",
+      example = "projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>")
+  String getKafkaReadSaslScramPasswordSecretId();
+
+  void setKafkaReadSaslScramPasswordSecretId(String value);
+
+  @TemplateParameter.GcsReadFile(
+      order = 15,
+      optional = true,
+      groupName = "Destination",
+      parentName = "kafkaReadAuthenticationMode",
+      parentTriggerValues = {KafkaAuthenticationMethod.SASL_SCRAM_512},
+      description = "Truststore File Location",
+      helpText =
+          "The Google Cloud Storage path to the Java TrustStore (JKS) file that contains"
+              + " the trusted certificates to use to verify the identity of the Kafka broker.")
+  String getKafkaReadSaslScramTruststoreLocation();
+
+  void setKafkaReadSaslScramTruststoreLocation(String sourceSaslScramTruststoreLocation);
+
+  @TemplateParameter.Text(
+      order = 16,
+      optional = true,
+      groupName = "Destination",
+      parentName = "kafkaReadAuthenticationMode",
+      parentTriggerValues = {KafkaAuthenticationMethod.SASL_SCRAM_512},
+      description = "Secret Version ID for Truststore Password",
+      helpText =
+          "The Google Cloud Secret Manager secret ID that contains the password to "
+              + "use to access the Java TrustStore (JKS) file for Kafka SASL_SCRAM authentication",
+      example = "projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>")
+  String getKafkaReadSaslScramTruststorePasswordSecretId();
+
+  void setKafkaReadSaslScramTruststorePasswordSecretId(
+      String sourceSaslScramTruststorePasswordSecretId);
 }

v2/kafka-common/src/main/java/com/google/cloud/teleport/v2/kafka/utils/KafkaConfig.java

@@ -44,7 +44,11 @@
             options.getKafkaReadKeystorePasswordSecretId(),
             options.getKafkaReadKeyPasswordSecretId(),
             options.getKafkaReadUsernameSecretId(),
-            options.getKafkaReadPasswordSecretId());
+            options.getKafkaReadPasswordSecretId(),
+            options.getKafkaReadSaslScramUsernameSecretId(),
+            options.getKafkaReadSaslScramPasswordSecretId(),
+            options.getKafkaReadSaslScramTruststoreLocation(),
+            options.getKafkaReadSaslScramTruststorePasswordSecretId());
 
     properties.putAll(KafkaCommonUtils.configureKafkaOffsetCommit(options));
 
@@ -60,7 +64,11 @@
         options.getKafkaWriteKeystorePasswordSecretId(),
         options.getKafkaWriteKeyPasswordSecretId(),
         options.getKafkaWriteUsernameSecretId(),
-        options.getKafkaWritePasswordSecretId());
+        options.getKafkaWritePasswordSecretId(),
+        options.getKafkaReadSaslScramUsernameSecretId(),
+        options.getKafkaReadSaslScramPasswordSecretId(),
+        options.getKafkaReadSaslScramTruststoreLocation(),
+        options.getKafkaReadSaslScramTruststorePasswordSecretId());
   }
 
   public static Map<String, Object> fromSchemaRegistryOptions(SchemaRegistryOptions options) {
@@ -122,7 +130,11 @@
       String keystorePasswordSecretId,
       String keyPasswordSecretId,
       String usernameSecretId,
-      String passwordSecretId) {
+      String passwordSecretId,
+      String scramUsernameSecretId,
+      String scramPasswordSecretId,
+      String scramTruststoreLocation,
+      String scramTruststorePasswordSecretId) {
     Map<String, Object> properties = new HashMap<>();
     if (authMode == null || authMode.equals(KafkaAuthenticationMethod.NONE)) {
       return properties;
@@ -154,6 +166,22 @@
               + " password=\'"
               + SecretManagerUtils.getSecret(passwordSecretId)
               + "\';");
+    } else if (authMode.equals(KafkaAuthenticationMethod.SASL_SCRAM_512)) {
+      properties.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-512");
+      properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
+      properties.put(
+          SaslConfigs.SASL_JAAS_CONFIG,
+          "org.apache.kafka.common.security.scram.ScramLoginModule required"
+              + " username=\'"
+              + SecretManagerUtils.getSecret(scramUsernameSecretId)
+              + "\'"
+              + " password=\'"
+              + SecretManagerUtils.getSecret(scramPasswordSecretId)
+              + "\';");
+      properties.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, scramTruststoreLocation);
+      properties.put(
+          SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG,
+          FileAwareFactoryFn.SECRET_MANAGER_VALUE_PREFIX + scramTruststorePasswordSecretId);
     } else if (authMode.equals(KafkaAuthenticationMethod.APPLICATION_DEFAULT_CREDENTIALS)) {
       properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
       properties.put(SaslConfigs.SASL_MECHANISM, "OAUTHBEARER");

v2/kafka-common/src/main/java/com/google/cloud/teleport/v2/kafka/values/KafkaAuthenticationMethod.java

@@ -22,6 +22,7 @@ public class KafkaAuthenticationMethod {
   public static final String NONE = "NONE";
   public static final String TLS = "TLS";
   public static final String SASL_PLAIN = "SASL_PLAIN";
+  public static final String SASL_SCRAM_512 = "SASL_SCRAM_512";
   public static final String APPLICATION_DEFAULT_CREDENTIALS = "APPLICATION_DEFAULT_CREDENTIALS";
   public static final String OAUTH = "OAUTH";
 }