hachikuji commented on a change in pull request #9996:
URL: https://github.com/apache/kafka/pull/9996#discussion_r569029166
##########
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);
Review comment:
I think you can just raise the exception? It will get caught and
`Call.fail` will be called. This does all the `tries` bookkeeping and will
resubmit it if the exception is retriable (which `TimeoutException` is).
----------------------------------------------------------------
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
