Add HTTP timeouts to KafkaAgentClient to prevent KafkaRoller from blocking

KafkaAgentClient.getBrokerState() relies on a java.net.http.HttpClient that had neither a connect timeout nor a request timeout configured. When a broker is alive but stuck on IO (for example because the underlying storage has zero IOPS), TCP connection establishment succeeds but the Kafka Agent never produces an HTTP response. The call therefore blocks the KafkaRoller's single-threaded executor indefinitely, preventing any other broker from being processed and inflating reconciliation time.

Add a bounded timeout (30s) on both the HttpClient connectTimeout and the HttpRequest timeout. On timeout the resulting HttpTimeoutException is wrapped as RuntimeException by doGet() and is already handled gracefully by getBrokerState(), which returns BrokerState(-1, null) and lets the roller move on to other brokers.

Fixes #12513

Signed-off-by: chon3806 <93464148+chon3806@users.noreply.github.com>

Author: chon3806

CHANGELOG.md

@@ -2,7 +2,7 @@
 
 ## 1.1.0
 
-* _Nothing here yet, but we will surely develop something new pretty soon_ 😉
+* Add HTTP request and connect timeouts to the Kafka Agent client so that a broker stuck on IO can no longer block the rolling update.
 
 ### Major changes, deprecations, and removals
 

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/resource/KafkaAgentClient.java

@@ -23,6 +23,7 @@
 import java.net.http.HttpRequest;
 import java.net.http.HttpResponse;
 import java.security.GeneralSecurityException;
+import java.time.Duration;
 
 /**
  * Creates HTTP client and interacts with Kafka Agent's REST endpoint
@@ -34,6 +35,10 @@ public class KafkaAgentClient {
     private static final String BROKER_STATE_REST_PATH = "/v1/broker-state/";
     private static final int KAFKA_AGENT_HTTPS_PORT = 8443;
     private static final char[] KEYSTORE_PASSWORD = "changeit".toCharArray();
+    // Bounds the connect and full HTTP request lifecycle so that a broker which accepts the TCP connection but
+    // never produces a response (e.g. alive but stuck on IO) cannot block the KafkaRoller's single-threaded
+    // executor indefinitely. See https://github.com/strimzi/strimzi-kafka-operator/issues/12513.
+    /* test */ static final Duration HTTP_REQUEST_TIMEOUT = Duration.ofSeconds(30);
     private final String namespace;
     private final Reconciliation reconciliation;
     private final String cluster;
@@ -92,20 +97,39 @@ private HttpClient createHttpClient() {
             SSLContext sslContext = SSLContext.getInstance("TLS");
             sslContext.init(keyManagerFactory.getKeyManagers(), trustManagerFactory.getTrustManagers(), null);
 
-            return HttpClient.newBuilder()
+            return httpClientBuilder()
                     .sslContext(sslContext)
                     .build();
         } catch (GeneralSecurityException | IOException e) {
             throw new RuntimeException("Failed to configure HTTP client", e);
         }
     }
 
+    /**
+     * Returns an {@link HttpClient.Builder} pre-configured with {@link #HTTP_REQUEST_TIMEOUT} as the connect timeout.
+     * Centralising the timeout wiring here keeps {@link #createHttpClient()} focused on TLS setup and lets tests
+     * assert that the configured connect timeout is applied without needing a TLS identity.
+     */
+    /* test */ static HttpClient.Builder httpClientBuilder() {
+        return HttpClient.newBuilder().connectTimeout(HTTP_REQUEST_TIMEOUT);
+    }
+
+    /**
+     * Builds a {@code GET} {@link HttpRequest} for the given URI with {@link #HTTP_REQUEST_TIMEOUT} applied as the
+     * full request timeout. Centralising the timeout wiring here lets tests assert the request timeout without
+     * needing a live HTTP endpoint.
+     */
+    /* test */ static HttpRequest buildRequest(URI uri) {
+        return HttpRequest.newBuilder()
+                .uri(uri)
+                .timeout(HTTP_REQUEST_TIMEOUT)
+                .GET()
+                .build();
+    }
+
     String doGet(URI uri) {
         try {
-            HttpRequest req = HttpRequest.newBuilder()
-                    .uri(uri)
-                    .GET()
-                    .build();
+            HttpRequest req = buildRequest(uri);
 
             var response = httpClient.send(req, HttpResponse.BodyHandlers.ofString());
             if (response.statusCode() != 200) {

cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/resource/KafkaAgentClientTest.java

@@ -7,6 +7,11 @@
 import io.strimzi.operator.common.Reconciliation;
 import org.junit.jupiter.api.Test;
 
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.time.Duration;
+
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
@@ -60,4 +65,32 @@ public void testErrorResponse() {
         assertEquals(0, actual.remainingLogsToRecover());
         assertEquals(0, actual.remainingSegmentsToRecover());
     }
+
+    /**
+     * Regression guard for https://github.com/strimzi/strimzi-kafka-operator/issues/12513: a previous version of
+     * KafkaAgentClient configured no timeout on the per-request side, which let a broker stuck on IO block the
+     * KafkaRoller's single-threaded executor indefinitely. Verify that requests built via the package-private
+     * helper carry the configured timeout.
+     */
+    @Test
+    public void testBuildRequestAppliesHttpRequestTimeout() {
+        HttpRequest request = KafkaAgentClient.buildRequest(URI.create("https://example.invalid/v1/broker-state/"));
+        Duration timeout = request.timeout().orElseThrow(() -> new AssertionError("HTTP request timeout was not configured"));
+        assertEquals(KafkaAgentClient.HTTP_REQUEST_TIMEOUT, timeout, "Request timeout must equal HTTP_REQUEST_TIMEOUT");
+        assertTrue(timeout.toMillis() > 0, "HTTP request timeout must be positive but was " + timeout);
+    }
+
+    /**
+     * Regression guard for https://github.com/strimzi/strimzi-kafka-operator/issues/12513: a previous version of
+     * KafkaAgentClient configured no connect timeout, which let an unresponsive broker block the KafkaRoller's
+     * single-threaded executor indefinitely. Verify that clients built via the package-private helper carry the
+     * configured connect timeout.
+     */
+    @Test
+    public void testHttpClientBuilderAppliesConnectTimeout() {
+        HttpClient client = KafkaAgentClient.httpClientBuilder().build();
+        Duration connectTimeout = client.connectTimeout().orElseThrow(() -> new AssertionError("HTTP connect timeout was not configured"));
+        assertEquals(KafkaAgentClient.HTTP_REQUEST_TIMEOUT, connectTimeout, "Connect timeout must equal HTTP_REQUEST_TIMEOUT");
+        assertTrue(connectTimeout.toMillis() > 0, "HTTP connect timeout must be positive but was " + connectTimeout);
+    }
 }