Extract Cruise Control client

The long term objective is to move this client to operator-common and share it with rebalance operator. Signed-off-by: Federico Valeri <fedevaleri@gmail.com>
strimzi · May 12, 2024 · abd22af · abd22af
1 parent 4b4e3da
commit abd22af
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;