Make Connect integration tests more reliable (#91)

Signed-off-by: Mickael Maison <[email protected]>

Author: mimaison

client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/MetricsUtils.java

@@ -8,6 +8,7 @@
 import io.prometheus.metrics.model.snapshots.InfoSnapshot;
 import io.prometheus.metrics.model.snapshots.Labels;
 import io.prometheus.metrics.model.snapshots.MetricSnapshot;
+import io.strimzi.test.container.StrimziConnectCluster;
 import org.junit.jupiter.api.function.ThrowingConsumer;
 import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.Network;
@@ -18,7 +19,11 @@
 import java.io.IOException;
 import java.io.InputStreamReader;
 import java.net.HttpURLConnection;
+import java.net.URI;
 import java.net.URL;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
@@ -149,4 +154,32 @@ public static GenericContainer<?> clientContainer(Map<String, String> env, int p
                 .waitingFor(Wait.forHttp("/metrics").forStatusCode(200));
     }
 
+    /**
+     * Start a connector
+     * @param connect the Connect cluster
+     * @param name the name of the connector
+     * @param config the connector configuration
+     */
+    public static void startConnector(StrimziConnectCluster connect, String name, String config) {
+        assertTimeoutPreemptively(TIMEOUT, () -> {
+            while (true) {
+                HttpClient httpClient = HttpClient.newHttpClient();
+                URI uri = new URI(connect.getRestEndpoint() + "/connectors/" + name + "/config");
+                HttpRequest request = HttpRequest.newBuilder()
+                        .PUT(HttpRequest.BodyPublishers.ofString(config))
+                        .setHeader("Content-Type", "application/json")
+                        .uri(uri)
+                        .build();
+                HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
+                try {
+                    assertEquals(HttpURLConnection.HTTP_CREATED, response.statusCode());
+                    break;
+                } catch (Throwable t) {
+                    assertInstanceOf(AssertionError.class, t);
+                    TimeUnit.MILLISECONDS.sleep(100L);
+                }
+            }
+        });
+    }
+
 }

client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/integration/TestConnectMetricsIT.java

@@ -24,18 +24,12 @@
 import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.utility.MountableFile;
 
-import java.net.HttpURLConnection;
-import java.net.URI;
-import java.net.http.HttpClient;
-import java.net.http.HttpRequest;
-import java.net.http.HttpResponse;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
 import static io.strimzi.kafka.metrics.prometheus.ClientMetricsReporterConfig.ALLOWLIST_CONFIG;
-import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -158,7 +152,7 @@ private void setupConnect(Map<String, String> overrides) {
     }
 
     @Test
-    public void testConnectMetrics() throws Exception {
+    public void testConnectMetrics() {
         setupConnect(Map.of(
                 "consumer.metric.reporters", ClientMetricsReporter.class.getName(),
                 "producer.metric.reporters", ClientMetricsReporter.class.getName(),
@@ -175,7 +169,7 @@ public void testConnectMetrics() throws Exception {
                 "  \"topics\": \"" + TOPIC + "\",\n" +
                 "  \"file\": \"" + FILE + "\"\n" +
                 "}";
-        httpPut("/connectors/" + SINK_CONNECTOR + "/config", connectorConfig);
+        MetricsUtils.startConnector(connect, SINK_CONNECTOR, connectorConfig);
         checkMetricsExist(SINK_PATTERNS);
 
         // Start a source connector metrics and check its metrics
@@ -185,12 +179,12 @@ public void testConnectMetrics() throws Exception {
                 "  \"topic\": \"" + TOPIC + "\",\n" +
                 "  \"file\": \"" + FILE + "\"\n" +
                 "}";
-        httpPut("/connectors/" + SOURCE_CONNECTOR + "/config", connectorConfig);
+        MetricsUtils.startConnector(connect, SOURCE_CONNECTOR, connectorConfig);
         checkMetricsExist(SOURCE_PATTERNS);
     }
 
     @Test
-    public void testConnectMetricsWithAllowlist() throws Exception {
+    public void testConnectMetricsWithAllowlist() {
         setupConnect(Map.of(
                 "consumer.metric.reporters", ClientMetricsReporter.class.getName(),
                 "producer.metric.reporters", ClientMetricsReporter.class.getName(),
@@ -223,7 +217,7 @@ public void testConnectMetricsWithAllowlist() throws Exception {
                 "  \"topics\": \"" + TOPIC + "\",\n" +
                 "  \"file\": \"" + FILE + "\"\n" +
                 "}";
-        httpPut("/connectors/" + SINK_CONNECTOR + "/config", connectorConfig);
+        MetricsUtils.startConnector(connect, SINK_CONNECTOR, connectorConfig);
         List<String> allowedSinkPatterns = List.of(
                 "kafka_connect_connector_metrics_.*" + SINK_CONNECTOR_PATTERN,
                 "kafka_connect_connect_worker_metrics_connector_count 1.0",
@@ -241,7 +235,7 @@ public void testConnectMetricsWithAllowlist() throws Exception {
                 "  \"topic\": \"" + TOPIC + "\",\n" +
                 "  \"file\": \"" + FILE + "\"\n" +
                 "}";
-        httpPut("/connectors/" + SOURCE_CONNECTOR + "/config", connectorConfig);
+        MetricsUtils.startConnector(connect, SOURCE_CONNECTOR, connectorConfig);
         List<String> allowedSourcePatterns = List.of(
                 "kafka_connect_connector_metrics_.*" + SOURCE_CONNECTOR_PATTERN,
                 "kafka_connect_connect_worker_metrics_connector_count 2.0",
@@ -264,16 +258,4 @@ private void checkMetricsDontExist(List<String> patterns) {
             MetricsUtils.verify(worker, patterns, PORT, metrics -> assertTrue(metrics.isEmpty()));
         }
     }
-
-    private void httpPut(String path, String body) throws Exception {
-        HttpClient httpClient = HttpClient.newHttpClient();
-        URI uri = new URI(connect.getRestEndpoint() + path);
-        HttpRequest request = HttpRequest.newBuilder()
-                .PUT(HttpRequest.BodyPublishers.ofString(body))
-                .setHeader("Content-Type", "application/json")
-                .uri(uri)
-                .build();
-        HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
-        assertEquals(HttpURLConnection.HTTP_CREATED, response.statusCode());
-    }
 }