Extract Topic Operator's Cruise Control client

This is the first step in the direction of having a shared Cruise Control client based on Java HTTP client. For the moment changes are limited to the Topic Operator. If we agree on the strategy, I can then move the Cruise Control client to operator-common and also use it in CruiseControlApi (different PR). --- In my view, the shared Cruise Control client shouldn't be concerned with response handling. This client should simply send the request and return a response POJO. Much like Kafka admin and Kube clients do. The response handling is specific to the functionality we are implementing (the same response could be processed in different ways), so it should be the responsibility of the caller (CruiseControlApi in CO, ReplicasChangeHandler in TO). That way, each object in the call chain triggered by the controller is focused on doing one thing, and the code is easier to read and maintain. This would leave a Vertx dependency in CruiseControlApi required by KafkaRebalanceAssemblyOperator. Of course we can also get rid of that, but I would leave it for a dedicated PR. Signed-off-by: Federico Valeri <fedevaleri@gmail.com>
strimzi · May 14, 2024 · f226884 · f226884
1 parent 4b4e3da
commit f226884
Show file tree

Hide file tree

Showing 14 changed files with 739 additions and 338 deletions.
diff --git a/...in/java/io/strimzi/operator/cluster/operator/resource/cruisecontrol/CruiseControlApi.java b/...in/java/io/strimzi/operator/cluster/operator/resource/cruisecontrol/CruiseControlApi.java
@@ -20,11 +20,6 @@ public interface CruiseControlApi {
      */
     String CC_REST_API_PROGRESS_KEY = "progress";
 
-    /**
-     * User ID header key
-     */
-    String CC_REST_API_USER_ID_HEADER = "User-Task-ID";
-
     /**
      *  Gets the state of the Cruise Control server.
      *

diff --git a/...ava/io/strimzi/operator/cluster/operator/resource/cruisecontrol/CruiseControlApiImpl.java b/...ava/io/strimzi/operator/cluster/operator/resource/cruisecontrol/CruiseControlApiImpl.java
@@ -30,6 +30,8 @@
 import java.net.NoRouteToHostException;
 import java.util.concurrent.TimeoutException;
 
+import static io.strimzi.operator.common.model.cruisecontrol.CruiseControlHeaders.USER_TASK_ID_HEADER;
+
 /**
  * Implementation of the Cruise Control API client
  */
@@ -121,7 +123,7 @@ private Future<CruiseControlResponse> getCruiseControlState(String host, int por
                     request.result().send(response -> {
                         if (response.succeeded()) {
                             if (response.result().statusCode() == 200 || response.result().statusCode() == 201) {
-                                String userTaskID = response.result().getHeader(CC_REST_API_USER_ID_HEADER);
+                                String userTaskID = response.result().getHeader(USER_TASK_ID_HEADER.toString());
                                 response.result().bodyHandler(buffer -> {
                                     JsonObject json = buffer.toJsonObject();
                                     if (json.containsKey(CC_REST_API_ERROR_KEY)) {
@@ -151,7 +153,7 @@ private Future<CruiseControlResponse> getCruiseControlState(String host, int por
                 }
 
                 if (userTaskId != null) {
-                    request.result().putHeader(CC_REST_API_USER_ID_HEADER, userTaskId);
+                    request.result().putHeader(USER_TASK_ID_HEADER.toString(), userTaskId);
                 }
             });
         });
@@ -165,7 +167,7 @@ private void internalRebalance(String host, int port, String path, String userTa
             }
 
             if (userTaskId != null) {
-                request.result().putHeader(CC_REST_API_USER_ID_HEADER, userTaskId);
+                request.result().putHeader(USER_TASK_ID_HEADER.toString(), userTaskId);
             }
 
             if (authHttpHeader != null) {
@@ -176,14 +178,14 @@ private void internalRebalance(String host, int port, String path, String userTa
                 if (response.succeeded()) {
                     if (response.result().statusCode() == 200 || response.result().statusCode() == 201) {
                         response.result().bodyHandler(buffer -> {
-                            String userTaskID = response.result().getHeader(CC_REST_API_USER_ID_HEADER);
+                            String userTaskID = response.result().getHeader(USER_TASK_ID_HEADER.toString());
                             JsonObject json = buffer.toJsonObject();
                             CruiseControlRebalanceResponse ccResponse = new CruiseControlRebalanceResponse(userTaskID, json);
                             result.complete(ccResponse);
                         });
                     } else if (response.result().statusCode() == 202) {
                         response.result().bodyHandler(buffer -> {
-                            String userTaskID = response.result().getHeader(CC_REST_API_USER_ID_HEADER);
+                            String userTaskID = response.result().getHeader(USER_TASK_ID_HEADER.toString());
                             JsonObject json = buffer.toJsonObject();
                             CruiseControlRebalanceResponse ccResponse = new CruiseControlRebalanceResponse(userTaskID, json);
                             if (json.containsKey(CC_REST_API_PROGRESS_KEY)) {
@@ -199,7 +201,7 @@ private void internalRebalance(String host, int port, String path, String userTa
                         });
                     } else if (response.result().statusCode() == 500) {
                         response.result().bodyHandler(buffer -> {
-                            String userTaskID = response.result().getHeader(CC_REST_API_USER_ID_HEADER);
+                            String userTaskID = response.result().getHeader(USER_TASK_ID_HEADER.toString());
                             JsonObject json = buffer.toJsonObject();
                             if (json.containsKey(CC_REST_API_ERROR_KEY)) {
                                 // If there was a client side error, check whether it was due to not enough data being available ...
@@ -321,7 +323,7 @@ public Future<CruiseControlResponse> getUserTaskStatus(String host, int port, St
                     request.result().send(response -> {
                         if (response.succeeded()) {
                             if (response.result().statusCode() == 200 || response.result().statusCode() == 201) {
-                                String userTaskID = response.result().getHeader(CC_REST_API_USER_ID_HEADER);
+                                String userTaskID = response.result().getHeader(USER_TASK_ID_HEADER.toString());
                                 response.result().bodyHandler(buffer -> {
                                     JsonObject json = buffer.toJsonObject();
                                     JsonObject jsonUserTask = json.getJsonArray("userTasks").getJsonObject(0);
@@ -418,7 +420,7 @@ public Future<CruiseControlResponse> stopExecution(String host, int port) {
                     request.result().send(response -> {
                         if (response.succeeded()) {
                             if (response.result().statusCode() == 200 || response.result().statusCode() == 201) {
-                                String userTaskID = response.result().getHeader(CC_REST_API_USER_ID_HEADER);
+                                String userTaskID = response.result().getHeader(USER_TASK_ID_HEADER.toString());
                                 response.result().bodyHandler(buffer -> {
                                     JsonObject json = buffer.toJsonObject();
                                     if (json.containsKey(CC_REST_API_ERROR_KEY)) {

diff --git a/...t/java/io/strimzi/operator/cluster/operator/resource/cruisecontrol/MockCruiseControl.java b/...t/java/io/strimzi/operator/cluster/operator/resource/cruisecontrol/MockCruiseControl.java
@@ -31,6 +31,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.logging.LogManager;
 
+import static io.strimzi.operator.common.model.cruisecontrol.CruiseControlHeaders.USER_TASK_ID_HEADER;
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.mockserver.model.Header.header;
 import static org.mockserver.model.HttpRequest.request;
@@ -134,7 +135,7 @@ public void setupCCStateResponse() {
                                 .withQueryStringParameter(Parameter.param(CruiseControlParameters.JSON.toString(), "true|false"))
                                 .withQueryStringParameter(Parameter.param(CruiseControlParameters.VERBOSE.toString(), "true|false"))
                                 .withPath(CruiseControlEndpoints.STATE.toString())
-                                .withHeaders(header(CruiseControlApi.CC_REST_API_USER_ID_HEADER, STATE_PROPOSAL_NOT_READY),
+                                .withHeaders(header(USER_TASK_ID_HEADER.toString(), STATE_PROPOSAL_NOT_READY),
                                         AUTH_HEADER)
                                 .withSecure(true))
                 .respond(
@@ -204,7 +205,7 @@ public void setupCCRebalanceNotEnoughDataError(CruiseControlEndpoints endpoint)
                         response()
                                 .withStatusCode(500)
                                 .withBody(jsonError)
-                                .withHeaders(header(CruiseControlApi.CC_REST_API_USER_ID_HEADER, REBALANCE_NOT_ENOUGH_VALID_WINDOWS_ERROR_RESPONSE_UTID))
+                                .withHeaders(header(USER_TASK_ID_HEADER.toString(), REBALANCE_NOT_ENOUGH_VALID_WINDOWS_ERROR_RESPONSE_UTID))
                                 .withDelay(TimeUnit.SECONDS, 0));
     }
 
@@ -231,7 +232,7 @@ public void setupCCBrokerDoesNotExist(CruiseControlEndpoints endpoint) {
                         response()
                                 .withStatusCode(500)
                                 .withBody(jsonError)
-                                .withHeaders(header(CruiseControlApi.CC_REST_API_USER_ID_HEADER, BROKERS_NOT_EXIST_ERROR_RESPONSE_UTID))
+                                .withHeaders(header(USER_TASK_ID_HEADER.toString(), BROKERS_NOT_EXIST_ERROR_RESPONSE_UTID))
                                 .withDelay(TimeUnit.SECONDS, 0));
     }
 

diff --git a/...on/src/main/java/io/strimzi/operator/common/model/cruisecontrol/CruiseControlHeaders.java b/...on/src/main/java/io/strimzi/operator/common/model/cruisecontrol/CruiseControlHeaders.java
@@ -0,0 +1,31 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package io.strimzi.operator.common.model.cruisecontrol;
+
+/**
+ * Enum with Cruise Control headers
+ */
+public enum CruiseControlHeaders {
+    /**
+     * User task id
+     */
+    USER_TASK_ID_HEADER("User-Task-ID");
+
+    private final String name;
+
+    /**
+     * Creates the Enum from String
+     *
+     * @param name String with the path
+     */
+    CruiseControlHeaders(String name) {
+        this.name = name;
+    }
+
+    @Override
+    public String toString() {
+        return name;
+    }
+}
diff --git a/systemtest/src/main/java/io/strimzi/systemtest/utils/kafkaUtils/KafkaTopicUtils.java b/systemtest/src/main/java/io/strimzi/systemtest/utils/kafkaUtils/KafkaTopicUtils.java
@@ -347,7 +347,6 @@ private static boolean checkReplicaChangeFailureDueToInsufficientBrokers(KafkaTo
         if (kafkaTopic != null && kafkaTopic.getStatus() != null && kafkaTopic.getStatus().getReplicasChange() != null) {
             String message = kafkaTopic.getStatus().getReplicasChange().getMessage();
             return message != null &&
-                    message.contains("Replicas change failed (500), Error processing POST request") &&
                     message.contains("Requested RF cannot be more than number of alive brokers") &&
                     kafkaTopic.getStatus().getReplicasChange().getState().toValue().equals("pending") &&
                     kafkaTopic.getStatus().getReplicasChange().getTargetReplicas() == targetReplicas;