[KafkaIO] Remove beam_fn_api requirement for dynamic reads (#39735)

Author: goutamadwant

CHANGES.md

@@ -81,6 +81,7 @@
 ## Bugfixes
 
 * (Python) Fixed incorrect profiler options handling on portable runners ([#39613](https://github.com/apache/beam/issues/39613)).
+* (Java) KafkaIO dynamic reads no longer require the obsolete `beam_fn_api` experiment ([#29998](https://github.com/apache/beam/issues/29998)).
 
 ## Security Fixes
 

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

@@ -1651,18 +1651,13 @@ public PCollection<KafkaRecord<K, V>> expand(PBegin input) {
       checkArgument(
           getConsumerConfig().get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG) != null,
           "withBootstrapServers() is required");
-      // With dynamic read, we no longer require providing topic/partition during pipeline
-      // construction time. But it requires enabling beam_fn_api.
+      // With dynamic read, topics and partitions are discovered during pipeline execution.
       if (!isDynamicRead()) {
         checkArgument(
             (getTopics() != null && getTopics().size() > 0)
                 || (getTopicPartitions() != null && getTopicPartitions().size() > 0)
                 || getTopicPattern() != null,
             "Either withTopic(), withTopics(), withTopicPartitions() or withTopicPattern() is required");
-      } else {
-        checkArgument(
-            ExperimentalOptions.hasExperiment(input.getPipeline().getOptions(), "beam_fn_api"),
-            "Kafka Dynamic Read requires enabling experiment beam_fn_api.");
       }
       checkArgument(getKeyDeserializerProvider() != null, "withKeyDeserializer() is required");
       checkArgument(getValueDeserializerProvider() != null, "withValueDeserializer() is required");

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

@@ -17,6 +17,7 @@
  */
 package org.apache.beam.sdk.io.kafka;
 
+import static org.apache.beam.sdk.io.kafka.KafkaIOReadImplementationCompatibility.KafkaIOReadImplementation.SDF;
 import static org.apache.beam.sdk.io.kafka.KafkaIOTest.mkKafkaReadTransformWithOffsetDedup;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.containsInAnyOrder;
@@ -30,6 +31,7 @@
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
+import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.PipelineResult;
 import org.apache.beam.sdk.io.kafka.KafkaIOReadImplementationCompatibility.KafkaIOReadProperties;
 import org.apache.beam.sdk.io.kafka.KafkaIOTest.ValueAsTimestampFn;
@@ -162,6 +164,26 @@ public void testReadTransformCreationWithSdfImplementationBoundProperty() {
     assertThat(Lineage.query(r.metrics(), Lineage.Type.SOURCE), containsInAnyOrder(expect));
   }
 
+  @Test
+  public void testDynamicReadUsesSdfWithoutBeamFnApiExperiment() {
+    KafkaIO.Read<Integer, Long> read =
+        KafkaIOTest.mkKafkaReadTransform(
+                1000,
+                null,
+                new ValueAsTimestampFn(),
+                false, /* redistribute */
+                false, /* allowDuplicates */
+                0, /* numKeys */
+                null, /* offsetDeduplication */
+                null, /* topics */
+                null /* redistributeByRecordKey */)
+            .withDynamicRead(Duration.standardMinutes(1));
+
+    assertThat(
+        KafkaIOReadImplementationCompatibility.getCompatibility(read).supportsOnly(SDF), is(true));
+    Pipeline.create().apply(read);
+  }
+
   @Test
   public void testReadTransformCreationWithBothImplementationBoundProperties() {
     thrown.expect(IllegalStateException.class);