You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "C0urante (via GitHub)" <gi...@apache.org> on 2023/06/21 20:30:28 UTC
[GitHub] [kafka] C0urante commented on a diff in pull request #13284: KAFKA-14718: Fix flaky DedicatedMirrorIntegrationTest
C0urante commented on code in PR #13284:
URL: https://github.com/apache/kafka/pull/13284#discussion_r1237643809
##########
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:
Same thought RE only performing assertions with a single `MirrorMaker` instance instead of all of them.
##########
connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/DedicatedMirrorIntegrationTest.java:
##########
@@ -288,4 +327,28 @@ private void awaitTopicContent(EmbeddedKafkaCluster cluster, String clusterName,
}
}
+ /**
+ * Validates that the underlying connector are running for the given MirrorMaker.
+ */
+ private boolean isConnectorRunningForMirrorMaker(final Class<?> connectorClazz, final MirrorMaker mm, final SourceAndTarget sourceAndTarget) {
+ final String connName = connectorClazz.getSimpleName();
+ final ConnectorStateInfo connectorStatus = mm.connectorStatus(sourceAndTarget, connName);
+ return connectorStatus != null
+ // verify that connector state is set to running
+ && connectorStatus.connector().state().equals(AbstractStatus.State.RUNNING.toString());
+ }
+
+ /**
+ * Validates that the tasks are associated with the connector and they are running for the given MirrorMaker.
+ */
+ private <T extends SourceConnector> boolean isTaskRunningForMirrorMakerConnector(final Class<T> connectorClazz, final MirrorMaker mm, final String source, final String target) {
+ final SourceAndTarget sourceAndTarget = new SourceAndTarget(source, target);
+ final String connName = connectorClazz.getSimpleName();
+ final ConnectorStateInfo connectorStatus = mm.connectorStatus(sourceAndTarget, connName);
+ return isConnectorRunningForMirrorMaker(connectorClazz, mm, sourceAndTarget)
+ // verify that at least one task exists
+ && !connectorStatus.tasks().isEmpty()
+ // verify that tasks are set to running
+ && connectorStatus.tasks().stream().allMatch(s -> s.state().equals(AbstractStatus.State.RUNNING.toString()));
Review Comment:
Same thought RE throwing a `NoRetryException` if a connector (or, here, task) is `FAILED`.
##########
connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/DedicatedMirrorIntegrationTest.java:
##########
@@ -94,6 +100,11 @@ private MirrorMaker startMirrorMaker(String name, Map<String, String> mmProps) {
result.start();
+ // wait for connectors to start
+ String[] clusterNames = mmProps.get("clusters").split(",");
+ SourceAndTarget sourceAndTarget = new SourceAndTarget(clusterNames[0].trim(), clusterNames[1].trim());
+ awaitMirrorMakerStart(sourceAndTarget);
Review Comment:
This assumes we define a one-way flow with two clusters, from the first cluster to the second.
This assumption may not hold in future updates and people may think we did this intentionally (instead of out of convenience).
I think it'd be better to move the calls to `awaitMirrorMakerStart` into the test cases themselves, since that's where the clusters for each run are currently defined.
##########
connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/DedicatedMirrorIntegrationTest.java:
##########
@@ -288,4 +327,28 @@ private void awaitTopicContent(EmbeddedKafkaCluster cluster, String clusterName,
}
}
+ /**
+ * Validates that the underlying connector are running for the given MirrorMaker.
+ */
+ private boolean isConnectorRunningForMirrorMaker(final Class<?> connectorClazz, final MirrorMaker mm, final SourceAndTarget sourceAndTarget) {
+ final String connName = connectorClazz.getSimpleName();
+ final ConnectorStateInfo connectorStatus = mm.connectorStatus(sourceAndTarget, connName);
+ return connectorStatus != null
+ // verify that connector state is set to running
+ && connectorStatus.connector().state().equals(AbstractStatus.State.RUNNING.toString());
Review Comment:
Can we throw a [NoRetryException](https://github.com/apache/kafka/blob/3c059133d3008d87f018f2efa4af27027fd5d18e/clients/src/test/java/org/apache/kafka/test/NoRetryException.java) if we see that the connector's status is `FAILED`? That way, developers won't have to wait for two minutes to learn that startup has failed. We can also include (or at least log) the stack trace for the failed connector.
##########
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:
We don't have to perform this assertion for every MM2 node; it's enough to verify that it succeeds with at least one.
I think we may want to refactor this method to accept a `MirrorMaker` instance and require the caller to provide a single instance that can be used to query the status of connectors/tasks. This would work well if we also move the call sites for this method into individual test cases, so that we only have to call it once per replication flow instead of once per MM2 node.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org