Extract Topic Operator's Cruise Control client (#10092)

Signed-off-by: Federico Valeri <[email protected]>
strimzi · May 23, 2024 · 6c24f81 · 6c24f81
1 parent 57a4a21
commit 6c24f81
Show file tree

Hide file tree

Showing 14 changed files with 743 additions and 339 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);
  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, 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, 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);
  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);
  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);
  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);
  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);
  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,8 +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),
- AUTH_HEADER)
+ .withHeaders(header(USER_TASK_ID_HEADER, STATE_PROPOSAL_NOT_READY), AUTH_HEADER)
  .withSecure(true))
  .respond(
  response()
@@ -204,7 +204,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, REBALANCE_NOT_ENOUGH_VALID_WINDOWS_ERROR_RESPONSE_UTID))
  .withDelay(TimeUnit.SECONDS, 0));
  }
 
@@ -231,7 +231,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, 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,15 @@
+/*
+ * 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;
+
+/**
+ * Cruise Control headers.
+ */
+public class CruiseControlHeaders {
+ /**
+ * User task id.
+ */
+ public static final String USER_TASK_ID_HEADER = "User-Task-ID";
+}
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
@@ -349,7 +349,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;