Apply GCP Application Default authentication when using a Managed Service for Apache Kafka cluster. (#2925)

* Add logic to apply GCP Application Default authentication when using a Managed Service for Apache Kafka cluster

* Tighten the regex for better matching on google clusters.

Author: fozzie15

v2/streaming-data-generator/src/main/java/com/google/cloud/teleport/v2/transforms/StreamingDataGeneratorWriteToKafka.java

@@ -25,6 +25,7 @@
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
+import java.util.regex.Pattern;
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericDatumReader;
 import org.apache.avro.generic.GenericDatumWriter;
@@ -49,6 +50,9 @@
 /** A {@link PTransform} that converts generatedMessages to write to Spanner table. */
 public final class StreamingDataGeneratorWriteToKafka {
 
+  private static final Pattern ManagedKafkaRegex =
+      Pattern.compile("bootstrap\\..*\\.managedkafka\\..*\\.cloud\\.goog.*");
+
   private StreamingDataGeneratorWriteToKafka() {}
 
   /** Creates Kafka message with JSON (UTF-8) encoded Payload. */
@@ -179,13 +183,15 @@ public PDone expand(PCollection<byte[]> generatedMessages) {
                           StreamingDataGenerator.OutputType.AVRO.name())));
       }
 
-      return kafkaMessages.apply(
-          "writeSuccessMessages",
+      KafkaIO.Write<Void, String> writeTransform =
           KafkaIO.<Void, String>write()
               .withBootstrapServers(getPipelineOptions().getBootstrapServer())
               .withTopic(getPipelineOptions().getKafkaTopic())
-              .withValueSerializer(StringSerializer.class)
-              .values());
+              .withValueSerializer(StringSerializer.class);
+      if (ManagedKafkaRegex.matcher(getPipelineOptions().getBootstrapServer()).matches()) {
+        writeTransform = writeTransform.withGCPApplicationDefaultCredentials();
+      }
+      return kafkaMessages.apply("writeSuccessMessages", writeTransform.values());
     }
   }
 }