(flink-connector-kafka) branch main updated: [FLINK-36780] Kafka source disable partition discovery unexpectedly (#136)

[arvid]

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

arvid pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git


The following commit(s) were added to refs/heads/main by this push:
     new f6a077a9 [FLINK-36780] Kafka source disable partition discovery 
unexpectedly (#136)
f6a077a9 is described below

commit f6a077a9dd8d1d5e43fc545cc9baab227d8438a0
Author: Mingliang Liu <lium...@apache.org>
AuthorDate: Tue Dec 3 05:11:18 2024 -0800

    [FLINK-36780] Kafka source disable partition discovery unexpectedly (#136)
---
 .../connector/kafka/source/KafkaSourceBuilder.java |  7 +++----
 .../kafka/source/KafkaSourceBuilderTest.java       | 23 ++++++++++++++++++++++
 2 files changed, 26 insertions(+), 4 deletions(-)

diff --git 
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java
 
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java
index 78a4b0b6..0709afe0 100644
--- 
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java
+++ 
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java
@@ -474,10 +474,9 @@ public class KafkaSourceBuilder<OUT> {
                 true);
 
         // If the source is bounded, do not run periodic partition discovery.
-        maybeOverride(
-                KafkaSourceOptions.PARTITION_DISCOVERY_INTERVAL_MS.key(),
-                "-1",
-                boundedness == Boundedness.BOUNDED);
+        if (boundedness == Boundedness.BOUNDED) {
+            
maybeOverride(KafkaSourceOptions.PARTITION_DISCOVERY_INTERVAL_MS.key(), "-1", 
true);
+        }
 
         // If the client id prefix is not set, reuse the consumer group id as 
the client id prefix,
         // or generate a random string if consumer group id is not specified.
diff --git 
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilderTest.java
 
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilderTest.java
index 2829f01e..ca777bc7 100644
--- 
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilderTest.java
+++ 
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilderTest.java
@@ -217,6 +217,29 @@ public class KafkaSourceBuilderTest {
                 .hasMessageContaining(expectedError);
     }
 
+    @Test
+    public void testDefaultPartitionDiscovery() {
+        final KafkaSource<String> kafkaSource = getBasicBuilder().build();
+        // Commit on checkpoint and auto commit should be disabled because 
group.id is not specified
+        assertThat(
+                        kafkaSource
+                                .getConfiguration()
+                                
.get(KafkaSourceOptions.PARTITION_DISCOVERY_INTERVAL_MS))
+                
.isEqualTo(KafkaSourceOptions.PARTITION_DISCOVERY_INTERVAL_MS.defaultValue());
+    }
+
+    @Test
+    public void testPeriodPartitionDiscovery() {
+        final KafkaSource<String> kafkaSource =
+                
getBasicBuilder().setBounded(OffsetsInitializer.latest()).build();
+        // Commit on checkpoint and auto commit should be disabled because 
group.id is not specified
+        assertThat(
+                        kafkaSource
+                                .getConfiguration()
+                                
.get(KafkaSourceOptions.PARTITION_DISCOVERY_INTERVAL_MS))
+                .isEqualTo(-1L);
+    }
+
     private KafkaSourceBuilder<String> getBasicBuilder() {
         return new KafkaSourceBuilder<String>()
                 .setBootstrapServers("testServer")