(beam) branch master updated: [KafkaIO] Remove beam_fn_api requirement for dynamic reads (#39735)

[yhu]

This is an automated email from the ASF dual-hosted git repository.

Abacn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new a0e27149ee1 [KafkaIO] Remove beam_fn_api requirement for dynamic reads 
(#39735)
a0e27149ee1 is described below

commit a0e27149ee119b1bf6bff12c8c08c803f6757946
Author: Goutam Adwant <8672451+goutamadw...@users.noreply.github.com>
AuthorDate: Wed Aug 12 09:36:37 2026 -0700

    [KafkaIO] Remove beam_fn_api requirement for dynamic reads (#39735)
---
 CHANGES.md                                         |  1 +
 .../java/org/apache/beam/sdk/io/kafka/KafkaIO.java |  7 +------
 ...KafkaIOReadImplementationCompatibilityTest.java | 22 ++++++++++++++++++++++
 3 files changed, 24 insertions(+), 6 deletions(-)

diff --git a/CHANGES.md b/CHANGES.md
index e2c3f262536..73966a48313 100644
--- a/CHANGES.md
+++ b/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
 
diff --git 
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java 
b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
index 4e8059e689b..33fdc76ca58 100644
--- a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
+++ b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
@@ -1651,18 +1651,13 @@ public class KafkaIO {
       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");
diff --git 
a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOReadImplementationCompatibilityTest.java
 
b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOReadImplementationCompatibilityTest.java
index 33870794e77..55ae55ceca8 100644
--- 
a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOReadImplementationCompatibilityTest.java
+++ 
b/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.Set;
 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 class KafkaIOReadImplementationCompatibilityTest {
     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);