[GitHub] [flink-connector-kafka] chucheng92 commented on a diff in pull request #8: [FLINK-31319][connectors/kafka] Fix kafka new source partitionDiscoveryIntervalMs=0 cause bounded source can not quit

["chucheng92 (via GitHub)" <gi...@apache.org>]

chucheng92 commented on code in PR #8:
URL: https://github.com/apache/flink-connector-kafka/pull/8#discussion_r1139572140


##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaEnumeratorTest.java:
##########
@@ -166,6 +174,68 @@ public void testReaderRegistrationTriggersAssignments() throws Throwable {
         }
     }
 
+    @Test
+    public void testRunWithDiscoverPartitionsOnceToCheckNoMoreSplit() throws Throwable {
+        try (MockSplitEnumeratorContext<KafkaPartitionSplit> context =
+                        new MockSplitEnumeratorContext<>(NUM_SUBTASKS);
+                KafkaSourceEnumerator enumerator =
+                        createEnumerator(context, DISABLE_PERIODIC_PARTITION_DISCOVERY)) {
+
+            // Start the enumerator and it should schedule a one time task to discover and assign
+            // partitions.
+            enumerator.start();
+
+            // Run the partition discover callable and check the partition assignment.
+            runOneTimePartitionDiscovery(context);
+
+            // enumerator noMoreNewPartitionSplits first will be false, when execute
+            // handlePartitionSplitChanges will be set true
+            assertThat((Boolean) Whitebox.getInternalState(enumerator, "noMoreNewPartitionSplits"))
+                    .isTrue();
+        }
+    }
+
+    @Test
+    public void testRunWithPeriodicPartitionDiscoveryOnceToCheckNoMoreSplit() throws Throwable {
+        try (MockSplitEnumeratorContext<KafkaPartitionSplit> context =
+                        new MockSplitEnumeratorContext<>(NUM_SUBTASKS);
+                KafkaSourceEnumerator enumerator =
+                        createEnumerator(context, ENABLE_PERIODIC_PARTITION_DISCOVERY)) {
+
+            // Start the enumerator and it should schedule a one time task to discover and assign
+            // partitions.
+            enumerator.start();
+            assertThat(context.getOneTimeCallables()).isEmpty();
+            assertThat(context.getPeriodicCallables())
+                    .as("A periodic partition discovery callable should have been scheduled")
+                    .hasSize(1);
+
+            // enumerator noMoreNewPartitionSplits first will be false, even when execute

Review Comment:
   @RamanVerma thanks. i added it.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org