divijvaidya commented on code in PR #13284:
URL: https://github.com/apache/kafka/pull/13284#discussion_r1248164637
##########
connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/DedicatedMirrorIntegrationTest.java:
##########
@@ -273,6 +288,30 @@ private void writeToTopic(EmbeddedKafkaCluster cluster,
String topic, int numMes
}
}
+ private void awaitMirrorMakerStart(final SourceAndTarget sourceAndTarget)
throws InterruptedException {
+ waitForCondition(() -> {
+ try {
+ return mirrorMakers.values().stream().allMatch(
+ mm -> CONNECTOR_CLASSES.stream().allMatch(
+ connectorClazz ->
isConnectorRunningForMirrorMaker(connectorClazz, mm, sourceAndTarget)));
+ } catch (Exception ex) {
+ log.error("Something unexpected occurred. Unable to check for
startup status for mirror maker for {}", sourceAndTarget, ex);
+ throw ex;
+ }
+ }, MM_START_UP_TIMEOUT_MS, "MirrorMaker instances did not transition
to running in time");
+ }
+
+ private <T extends SourceConnector> void awaitConnectorTasksStart(final
Class<T> clazz, final String source, String target) throws InterruptedException
{
+ waitForCondition(() -> {
+ try {
+ return mirrorMakers.values().stream().allMatch(mm ->
isTaskRunningForMirrorMakerConnector(clazz, mm, source, target));
Review Comment:
Done. Only checking for one node now.
##########
connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/DedicatedMirrorIntegrationTest.java:
##########
@@ -273,6 +288,30 @@ private void writeToTopic(EmbeddedKafkaCluster cluster,
String topic, int numMes
}
}
+ private void awaitMirrorMakerStart(final SourceAndTarget sourceAndTarget)
throws InterruptedException {
+ waitForCondition(() -> {
+ try {
+ return mirrorMakers.values().stream().allMatch(
Review Comment:
Done in latest revision.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
