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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
