[GitHub] [kafka] skaundinya15 commented on a change in pull request #8894: KAFKA-9509: Add retries for mirrorClient consume records to fix flaky test

[GitBox <gi...@apache.org>]

skaundinya15 commented on a change in pull request #8894:
URL: https://github.com/apache/kafka/pull/8894#discussion_r442445728



##########
File path: connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorConnectorsIntegrationTest.java
##########
@@ -207,23 +212,45 @@ public void close() {
         backup.stop();
     }
 
+    // throw exception after 3 retries, and print expected error messages
+    private void assertEqualsWithConsumeRetries(final String errorMsg,
+                                                final int numRecordsProduces,
+                                                final int timeout,
+                                                final ClusterType clusterType,
+                                                final String... topics) throws InterruptedException {
+        int retries = 3;
+        while (retries-- > 0) {
+            try {
+                int actualNum = clusterType == ClusterType.PRIMARY ?
+                        primary.kafka().consume(numRecordsProduces, timeout, topics).count() :
+                        backup.kafka().consume(numRecordsProduces, timeout, topics).count();
+                if (numRecordsProduces == actualNum)
+                    return;
+            } catch (Throwable e) {
+                log.error("Could not find enough records with {} retries left", retries, e);
+            }
+        }
+        throw new InterruptedException(errorMsg);
+    }
+
     @Test
     public void testReplication() throws InterruptedException {
         MirrorClient primaryClient = new MirrorClient(mm2Config.clientConfig("primary"));
         MirrorClient backupClient = new MirrorClient(mm2Config.clientConfig("backup"));
 
-        assertEquals("Records were not produced to primary cluster.", NUM_RECORDS_PRODUCED,

Review comment:
       I'd agree with @ryannedolan here. We could use the `waitForCondition` in `TestUtils.java` instead to wait for the condition necessary instead. More details on that is here: https://github.com/apache/kafka/blob/d8cc6fe8e36329c647736773d9d66de89c447409/clients/src/test/java/org/apache/kafka/test/TestUtils.java#L370-L371




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org