MartijnVisser commented on code in PR #309:
URL:
https://github.com/apache/flink-connector-kafka/pull/309#discussion_r4368012002
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorTest.java:
##########
@@ -286,6 +286,68 @@ public void
testBoundedSourceCompletesReadersAfterActiveAndRetainedReassignment(
}
}
+ @Test
+ public void testBoundedSourceCompletesReadersAgainAfterMetadataChange()
throws Throwable {
+ // The stream moves off cluster 0 and onto cluster 1.
+ KafkaStream kafkaStreamOnFirstCluster =
DynamicKafkaSourceTestHelper.getKafkaStream(TOPIC);
+ kafkaStreamOnFirstCluster
+ .getClusterMetadataMap()
+ .remove(DynamicKafkaSourceTestHelper.getKafkaClusterId(1));
+ KafkaStream kafkaStreamOnSecondCluster =
DynamicKafkaSourceTestHelper.getKafkaStream(TOPIC);
+ kafkaStreamOnSecondCluster
+ .getClusterMetadataMap()
+ .remove(DynamicKafkaSourceTestHelper.getKafkaClusterId(0));
+
+ MockKafkaMetadataService mockKafkaMetadataService =
+ new
MockKafkaMetadataService(Collections.singleton(kafkaStreamOnFirstCluster));
+
+ try (RecordingSplitEnumeratorContext context = new
RecordingSplitEnumeratorContext();
+ DynamicKafkaSourceEnumerator enumerator =
+ createBoundedEnumerator(
+ context,
+ mockKafkaMetadataService,
+ (properties) ->
+ properties.setProperty(
+ DynamicKafkaSourceOptions
+
.STREAM_METADATA_DISCOVERY_INTERVAL_MS
+ .key(),
+ "1"),
+ new DynamicKafkaSourceEnumState())) {
+ enumerator.start();
+ // Discovery is periodic at this interval, so the first round has
to be driven before
+ // any reader can be told that there is nothing more coming.
+ context.runPeriodicCallable(0);
+ runAllOneTimeCallables(context);
+ for (int reader = 0; reader < NUM_SUBTASKS; reader++) {
+ mockRegisterReaderAndSendReaderStartupEvent(context,
enumerator, reader);
+ }
+ runAllOneTimeCallables(context);
+
+ for (int reader = 0; reader < NUM_SUBTASKS; reader++) {
+ assertThat(context.splitsAtCompletion.get(reader))
+ .as(
+ "reader %s should have been told once before
the metadata change",
+ reader)
+ .hasSize(1);
+ }
+
+ // The reader recreates its sub readers on a metadata change, so
each must be told
+ // again.
+ mockKafkaMetadataService.setKafkaStreams(
+ Collections.singleton(kafkaStreamOnSecondCluster));
+ context.runPeriodicCallable(0);
+ runAllOneTimeCallables(context);
+
+ for (int reader = 0; reader < NUM_SUBTASKS; reader++) {
+ assertThat(context.splitsAtCompletion.get(reader))
+ .as(
+ "reader %s should have been told again after
the metadata change",
+ reader)
+ .hasSize(2);
Review Comment:
Could this also assert that the second entry contains the reader's cluster 1
split? That pins that the signal follows the assignment.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorTest.java:
##########
@@ -286,6 +286,68 @@ public void
testBoundedSourceCompletesReadersAfterActiveAndRetainedReassignment(
}
}
+ @Test
+ public void testBoundedSourceCompletesReadersAgainAfterMetadataChange()
throws Throwable {
+ // The stream moves off cluster 0 and onto cluster 1.
Review Comment:
Nit: since FLINK-31006 landed, adding cluster 1 next to cluster 0 also fails
without the clear, so I think a `@ParameterizedTest` over both changes is worth
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
