Merge pull request #1537 from an2x:master

PiperOrigin-RevId: 634841553
GoogleCloudPlatform · May 17, 2024 · dbff166 · dbff166
2 parents 4b06b8b + 5c92cc8
commit dbff166
Show file tree

Hide file tree

Showing 6 changed files with 74 additions and 2 deletions.
diff --git a/metadata/src/main/java/com/google/cloud/teleport/metadata/TemplateParameter.java b/metadata/src/main/java/com/google/cloud/teleport/metadata/TemplateParameter.java
@@ -701,4 +701,44 @@ public final class TemplateParameter {
     /** Parameter visibility in the UI. */
     boolean hiddenUi() default false;
   }
+
+  /**
+   * Template parameter containing a Kafka Topic.
+   *
+   * <p>The parameter specifies the fully-qualified name of an Apache Kafka topic. This can be
+   * either a Google Managed Kafka topic or a non-managed Kafka topic.
+   */
+  @Retention(RetentionPolicy.RUNTIME)
+  @Target({ElementType.FIELD, ElementType.METHOD})
+  public @interface KafkaTopic {
+    /** Order of appearance. */
+    int order() default 999;
+
+    /** Name of the parameter. */
+    String name() default "";
+
+    /** Group Name of the parameter. */
+    String groupName() default "";
+
+    /** Parent Name of the parameter. */
+    String parentName() default "";
+
+    /** List of parent trigger values. */
+    String[] parentTriggerValues() default "";
+
+    /** If parameter is optional. */
+    boolean optional() default false;
+
+    /** Description of the parameter. */
+    String description();
+
+    /** Help text of the parameter. */
+    String helpText();
+
+    /** Example of the parameter. */
+    String example() default "";
+
+    /** Parameter visibility in the UI. */
+    boolean hiddenUi() default false;
+  }
 }
diff --git a/metadata/src/main/java/com/google/cloud/teleport/metadata/util/MetadataUtils.java b/metadata/src/main/java/com/google/cloud/teleport/metadata/util/MetadataUtils.java
@@ -42,6 +42,7 @@ public final class MetadataUtils {
         TemplateParameter.GcsWriteFile.class,
         TemplateParameter.GcsWriteFolder.class,
         TemplateParameter.Integer.class,
+        TemplateParameter.KafkaTopic.class,
         TemplateParameter.KmsEncryptionKey.class,
         TemplateParameter.Long.class,
         TemplateParameter.Password.class,

diff --git a/.../core-plugin/src/main/java/com/google/cloud/teleport/plugin/model/ImageSpecParameter.java b/.../core-plugin/src/main/java/com/google/cloud/teleport/plugin/model/ImageSpecParameter.java
@@ -468,6 +468,23 @@ public void processParamType(Annotation parameterAnnotation) {
         this.setHiddenUi(durationParam.hiddenUi());
         this.setParamType(ImageSpecParameterType.TEXT);
         break;
+      case "KafkaTopic":
+        TemplateParameter.KafkaTopic kafkaTopic =
+            (TemplateParameter.KafkaTopic) parameterAnnotation;
+        if (!kafkaTopic.name().isEmpty()) {
+          this.setName(kafkaTopic.name());
+        }
+        processDescriptions(
+            kafkaTopic.groupName(),
+            kafkaTopic.description(),
+            kafkaTopic.helpText(),
+            kafkaTopic.example());
+        this.setParentName(kafkaTopic.parentName());
+        this.setParentTriggerValues(kafkaTopic.parentTriggerValues());
+        this.setOptional(kafkaTopic.optional());
+        this.setHiddenUi(kafkaTopic.hiddenUi());
+        this.setParamType(ImageSpecParameterType.KAFKA_TOPIC);
+        break;
       default:
         throw new IllegalArgumentException("Invalid type " + parameterAnnotation);
     }

diff --git a/...e-plugin/src/main/java/com/google/cloud/teleport/plugin/model/ImageSpecParameterType.java b/...e-plugin/src/main/java/com/google/cloud/teleport/plugin/model/ImageSpecParameterType.java
@@ -49,5 +49,8 @@ public enum ImageSpecParameterType {
   ENUM,
 
   /** Number parameter. */
-  NUMBER;
+  NUMBER,
+
+  /** Kafka Topic parameter. */
+  KAFKA_TOPIC;
 }
diff --git a/...-plugin/src/test/java/com/google/cloud/teleport/plugin/model/TemplateDefinitionsTest.java b/...-plugin/src/test/java/com/google/cloud/teleport/plugin/model/TemplateDefinitionsTest.java
@@ -63,6 +63,9 @@ public void testSampleAtoBOk() {
     ImageSpecParameter to = metadata.getParameter("to").get();
     assertEquals(ImageSpecParameterType.BIGQUERY_TABLE, to.getParamType());
 
+    ImageSpecParameter inputKafkaTopic = metadata.getParameter("inputKafkaTopic").get();
+    assertEquals(ImageSpecParameterType.KAFKA_TOPIC, inputKafkaTopic.getParamType());
+
     ImageSpecParameter logical = metadata.getParameter("logical").get();
     assertEquals(ImageSpecParameterType.BOOLEAN, logical.getParamType());
     assertEquals("^(true|false)$", logical.getRegexes().get(0));

diff --git a/plugins/core-plugin/src/test/java/com/google/cloud/teleport/plugin/sample/AtoBOk.java b/plugins/core-plugin/src/test/java/com/google/cloud/teleport/plugin/sample/AtoBOk.java
@@ -99,12 +99,20 @@ public interface AtoBOptions {
     Boolean getParamWithGroupName();
 
     @TemplateParameter.Text(
-        order = 8,
+        order = 9,
         parentName = "paramWithGroupName",
         parentTriggerValues = {"true"},
         description = "N/A",
         helpText = "Text that has parent name and parent trigger value")
     @Default.Boolean(false)
     Boolean getParamWithParentName();
+
+    @TemplateParameter.KafkaTopic(
+        order = 10,
+        description = "Kafka input topic",
+        helpText = "Kafka topic to read from",
+        example =
+            "projects/project-foo/locations/us-central1/clusters/cluster-bar/topics/topic-baz")
+    String getInputKafkaTopic();
   }
 }