You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "RamanVerma (via GitHub)" <gi...@apache.org> on 2023/03/16 19:13:55 UTC
[GitHub] [flink-connector-kafka] RamanVerma commented on a diff in pull request #8: [FLINK-31319][connectors/kafka] Fix kafka new source partitionDiscoveryIntervalMs=0 cause bounded source can not quit
RamanVerma commented on code in PR #8:
URL: https://github.com/apache/flink-connector-kafka/pull/8#discussion_r1139238012
##########
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:
I think you haven't done partition discovery in this test. You need to call `runPeriodicPartitionDiscovery` at least once before check on line 215
##########
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
+ // handlePartitionSplitChanges it still be false
+ assertThat((Boolean) Whitebox.getInternalState(enumerator, "noMoreNewPartitionSplits"))
+ .isFalse();
+ }
+ }
+
+ @Test
+ public void testRunWithDiscoverPartitionsOnceWithZeroMsToCheckNoMoreSplit() throws Throwable {
+ try (MockSplitEnumeratorContext<KafkaPartitionSplit> context =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS);
+ // set partitionDiscoveryIntervalMs = 0
+ KafkaSourceEnumerator enumerator = createEnumerator(context, 0L)) {
+
+ // Start the enumerator, and it should schedule a one time task to discover and assign
+ // partitions.
+ enumerator.start();
+ runOneTimePartitionDiscovery(context);
Review Comment:
It would be better to check that one time and periodical callables are properly populated before calling `runOneTimePartitionDiscovery`. You are doing this in lines 208 and 209. Please copy that check (with appropriate changes) here and in the first test.
--
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