[FLINK-36780] Kafka source disable partition discovery unexpectedly (#…

…136)
apache · Dec 3, 2024 · f6a077a · f6a077a
1 parent 59baacc
commit f6a077a
Show file tree

Hide file tree

Showing 2 changed files with 26 additions and 4 deletions.
diff --git a/...ector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java b/...ector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java
@@ -474,10 +474,9 @@ private void parseAndSetRequiredProperties() {
                 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/...r-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilderTest.java b/...r-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilderTest.java
@@ -217,6 +217,29 @@ public void testSettingInvalidCustomDeserializers(
                 .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")