Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Extract Topic Operator's Cruise Control client #10092

Merged
merged 4 commits into from
May 23, 2024
Merged

Extract Topic Operator's Cruise Control client #10092

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
25ddddd
Extract Topic Operator's Cruise Control client
fvaleri May 8, 2024
1d9ea17
Convert headers enum to a string constant
fvaleri May 15, 2024
641a33b
Address comments from Tom
fvaleri May 16, 2024
efadc9b
Address more comments from Tom
fvaleri May 20, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 0 additions & 5 deletions ...in/java/io/strimzi/operator/cluster/operator/resource/cruisecontrol/CruiseControlApi.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
scholzj marked this conversation as resolved.
Show resolved Hide resolved

/**
* Gets the state of the Cruise Control server.
*
Expand Down
18 changes: 10 additions & 8 deletions ...ava/io/strimzi/operator/cluster/operator/resource/cruisecontrol/CruiseControlApiImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
*/
Expand Down Expand Up @@ -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)) {
Expand Down Expand Up @@ -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);
}
});
});
Expand All @@ -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) {
Expand All @@ -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)) {
Expand All @@ -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 ...
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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)) {
Expand Down
8 changes: 4 additions & 4 deletions ...t/java/io/strimzi/operator/cluster/operator/resource/cruisecontrol/MockCruiseControl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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));
}

Expand All @@ -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));
}

Expand Down
15 changes: 15 additions & 0 deletions ...on/src/main/java/io/strimzi/operator/common/model/cruisecontrol/CruiseControlHeaders.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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";
}
ppatierno marked this conversation as resolved.
Show resolved Hide resolved
1 change: 0 additions & 1 deletion systemtest/src/main/java/io/strimzi/systemtest/utils/kafkaUtils/KafkaTopicUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
Loading
Loading