slachiewicz commented on code in PR #305:
URL:
https://github.com/apache/flink-connector-kafka/pull/305#discussion_r3972075822
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumeratorTest.java:
##########
@@ -632,6 +632,100 @@ public void testPartitionChangeChecking() throws
Throwable {
}
}
+ /** Verifies that in-flight partitions are not rediscovered as new on the
next cycle. */
+ @Test
+ public void testInFlightPartitionsNotRediscoveredAsNew() throws Throwable {
+ try (MockSplitEnumeratorContext<KafkaPartitionSplit> context =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS);
+ KafkaSourceEnumerator enumerator =
+ createEnumerator(context,
ENABLE_PERIODIC_PARTITION_DISCOVERY)) {
+ enumerator.start();
+
+ // Register readers so partitions can be assigned.
+ registerReader(context, enumerator, READER0);
+ registerReader(context, enumerator, READER1);
+ registerReader(context, enumerator, READER2);
+
+ // Run discovery once and leave the worker callable pending to
simulate a slow
+ // worker.
+ context.runPeriodicCallable(PARTITION_DISCOVERY_CALLABLE_INDEX);
+
+ // The worker callable is now pending.
+ assertThat(context.getOneTimeCallables())
+ .as("initializePartitionSplits should be pending on the
worker")
+ .hasSize(1);
+
+ // Fire a second cycle before the first init completes.
+ context.runPeriodicCallable(PARTITION_DISCOVERY_CALLABLE_INDEX);
+
+ // The second cycle dispatches nothing because partitions are in
flight.
+ assertThat(context.getOneTimeCallables())
+ .as(
+ "No additional initializePartitionSplits should be
dispatched while partitions are in flight")
+ .hasSize(1);
+
+ // Run the pending init.
+ context.runNextOneTimeCallable();
+
+ // Verify a single assignment without duplicates.
+ List<SplitsAssignment<KafkaPartitionSplit>> assignments =
+ context.getSplitsAssignmentSequence();
+ assertThat(assignments).as("Exactly one assignment batch
expected").hasSize(1);
+
+ // Count assigned partitions across topics.
+ int totalAssignedPartitions =
+
assignments.get(0).assignment().values().stream().mapToInt(List::size).sum();
+ int expectedPartitions =
+
KafkaSourceTestEnv.getPartitionsForTopics(PRE_EXISTING_TOPICS).size();
+ assertThat(totalAssignedPartitions)
+ .as("Each partition should be assigned exactly once, not
duplicated")
+ .isEqualTo(expectedPartitions);
+ }
+ }
+
+ /** Verifies that a migrated split in flight is not rediscovered on the
next cycle. */
Review Comment:
Great test coverage reproducing the multi-discovery-cycle scenario! Adding
both testInFlightPartitionsNotRediscoveredAsNew and
testMigratedSplitInFlightNotRediscovered ensures both fresh split assignment
and migration paths are guarded.
--
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]
