Merge pull request #35214: Allow STRING format in Kafka config

Author: chamikaramj

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaReadSchemaTransformConfiguration.java

@@ -71,6 +71,9 @@ public void validate() {
     } else if (dataFormat != null && dataFormat.equals("RAW")) {
       checkArgument(
           inputSchema == null, "To read from Kafka in RAW format, you can't provide a schema.");
+    } else if (dataFormat != null && dataFormat.equals("STRING")) {
+      checkArgument(
+          inputSchema == null, "To read from Kafka in STRING format, you can't provide a schema.");
     } else if (dataFormat != null && dataFormat.equals("JSON")) {
       checkNotNull(inputSchema, "To read from Kafka in JSON format, you must provide a schema.");
     } else if (dataFormat != null && dataFormat.equals("PROTO")) {

sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaReadSchemaTransformProviderTest.java

@@ -197,6 +197,24 @@ public void testBuildTransformWithRawFormat() {
             .build());
   }
 
+  @Test
+  public void testBuildTransformWithStringFormat() {
+    ServiceLoader<SchemaTransformProvider> serviceLoader =
+        ServiceLoader.load(SchemaTransformProvider.class);
+    List<SchemaTransformProvider> providers =
+        StreamSupport.stream(serviceLoader.spliterator(), false)
+            .filter(provider -> provider.getClass() == KafkaReadSchemaTransformProvider.class)
+            .collect(Collectors.toList());
+    KafkaReadSchemaTransformProvider kafkaProvider =
+        (KafkaReadSchemaTransformProvider) providers.get(0);
+    kafkaProvider.from(
+        KafkaReadSchemaTransformConfiguration.builder()
+            .setTopic("anytopic")
+            .setBootstrapServers("anybootstrap")
+            .setFormat("STRING")
+            .build());
+  }
+
   @Test
   public void testBuildTransformWithProtoFormat() {
     ServiceLoader<SchemaTransformProvider> serviceLoader =
@@ -300,15 +318,16 @@ public void testBuildTransformWithManaged() {
     List<String> configs =
         Arrays.asList(
             "topic: topic_1\n" + "bootstrap_servers: some bootstrap\n" + "format: RAW",
-            "topic: topic_2\n"
+            "topic: topic_2\n" + "bootstrap_servers: some bootstrap\n" + "format: STRING",
+            "topic: topic_3\n"
                 + "bootstrap_servers: some bootstrap\n"
                 + "schema: '{\"type\":\"record\",\"name\":\"my_record\",\"fields\":[{\"name\":\"bool\",\"type\":\"boolean\"}]}'",
-            "topic: topic_3\n"
+            "topic: topic_4\n"
                 + "bootstrap_servers: some bootstrap\n"
                 + "schema_registry_url: some-url\n"
                 + "schema_registry_subject: some-subject\n"
                 + "format: RAW",
-            "topic: topic_4\n"
+            "topic: topic_5\n"
                 + "bootstrap_servers: some bootstrap\n"
                 + "format: PROTO\n"
                 + "schema: '"