hachikuji commented on a change in pull request #9996:
URL: https://github.com/apache/kafka/pull/9996#discussion_r569029333
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -4605,6 +4609,56 @@ void handleFailure(Throwable throwable) {
return new UpdateFeaturesResult(new HashMap<>(updateFutures));
}
+ private Call
getDecommissionBrokerCall(DecommissionBrokerOperationContext<Void> context) {
+ return new Call("decommissionBroker", context.deadline(),
+ new LeastLoadedNodeProvider()) {
+
+ @Override
+ DecommissionBrokerRequest.Builder createRequest(int timeoutMs) {
+ DecommissionBrokerRequestData data =
+ new
DecommissionBrokerRequestData().setBrokerId(context.brokerId());
+ return new DecommissionBrokerRequest.Builder(data);
+ }
+
+ @Override
+ void handleResponse(AbstractResponse abstractResponse) {
+ final DecommissionBrokerResponse response =
+ (DecommissionBrokerResponse) abstractResponse;
+ Errors error = Errors.forCode(response.data().errorCode());
+ switch (error) {
+ case NONE:
+ context.future().complete(null);
+ break;
+ case REQUEST_TIMED_OUT:
+ Call nextCall = getDecommissionBrokerCall(context);
+ nextCall.tries = super.tries + 1;
+ nextCall.nextAllowedTryMs =
calculateNextAllowedRetryMs();
+ runnable.call(nextCall, time.milliseconds());
+ break;
+ default:
+
context.future().completeExceptionally(error.exception());
Review comment:
Can we log an error message here which includes the request information
and the unexpected error?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
