Fix KafkaToGcsIT (#1618)

* Fix test * Mark old kafka to bq template hidden * Spotless * Change if condition to check for schema path or schema registry * Add exception * Revert "Add exception" This reverts commit a198f49. * Dummy change
GoogleCloudPlatform · May 30, 2024 · 4319ce6 · 4319ce6
1 parent 2ce397f
commit 4319ce6
Show file tree

Hide file tree

Showing 4 changed files with 10 additions and 6 deletions.
diff --git a/...fka-to-bigquery/src/main/java/com/google/cloud/teleport/v2/templates/KafkaToBigQuery.java b/...fka-to-bigquery/src/main/java/com/google/cloud/teleport/v2/templates/KafkaToBigQuery.java
@@ -123,7 +123,8 @@
     },
     streaming = true,
     supportsAtLeastOnce = true,
-    supportsExactlyOnce = true)
+    supportsExactlyOnce = true,
+    hidden = true)
 public class KafkaToBigQuery {
 
   /* Logger for class. */

diff --git a/...to-bigquery/src/main/java/com/google/cloud/teleport/v2/templates/KafkaToBigQueryFlex.java b/...to-bigquery/src/main/java/com/google/cloud/teleport/v2/templates/KafkaToBigQueryFlex.java
@@ -108,8 +108,8 @@
       "keystoreLocation",
       "sourceSSL",
       "sourceTruststorePassword",
+      "sourceKey",
       "sourceKeystorePassword",
-      "sourceKey"
     })
 public class KafkaToBigQueryFlex {
 

diff --git a/...afka-to-gcs/src/main/java/com/google/cloud/teleport/v2/transforms/AvroWriteTransform.java b/...afka-to-gcs/src/main/java/com/google/cloud/teleport/v2/transforms/AvroWriteTransform.java
@@ -85,12 +85,14 @@ public WriteFilesResult<AvroDestination> expand(
         KafkaTemplateParamters.MessageFormatConstants.AVRO_CONFLUENT_WIRE_FORMAT)) {
       String schemaRegistryURL = schemaRegistryURL();
       String schemaPath = schemaPath();
-      if (schemaRegistryURL == null && schemaPath == null) {
+      assert schemaRegistryURL != null;
+      assert schemaPath != null;
+      if (schemaRegistryURL.isBlank() && schemaPath.isBlank()) {
         throw new UnsupportedOperationException(
-            "A Schema Registry URL or static schemas are required for CONFLUENT_WIRE_FORMAT messages");
+            "A Schema Registry URL or static schema is required for CONFLUENT_WIRE_FORMAT messages");
       }
       DoFn<KafkaRecord<byte[], byte[]>, GenericRecord> convertToBytes;
-      if (schemaRegistryURL != null) {
+      if (!schemaRegistryURL.isBlank()) {
         convertToBytes = new ConvertBytesToGenericRecord(schemaRegistryURL);
       } else {
         convertToBytes = new ConvertBytesToGenericRecord(schemaPath, true);

diff --git a/v2/kafka-to-gcs/src/test/java/com/google/cloud/teleport/v2/templates/KafkaToGcsIT.java b/v2/kafka-to-gcs/src/test/java/com/google/cloud/teleport/v2/templates/KafkaToGcsIT.java
@@ -111,7 +111,8 @@ private void baseKafkaToGcs(Function<LaunchConfig.Builder, LaunchConfig.Builder>
                         + ";"
                         + topicName)
                 .addParameter("windowDuration", "10s")
-                .addParameter("schemaPath", getGcsPath("avro_schema.avsc"))
+                .addParameter("schemaFormat", "SINGLE_SCHEMA_FILE")
+                .addParameter("confluentAvroSchemaPath", getGcsPath("avro_schema.avsc"))
                 .addParameter("kafkaReadOffset", "earliest")
                 .addParameter("outputDirectory", getGcsPath(testName))
                 .addParameter("outputFilenamePrefix", testName + "-")