Merge cecbc237ec into 4a5aa37169

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -3106,17 +3106,26 @@ public class KafkaAdminClient extends AdminClient {
                 @Override
                 public void handleResponse(AbstractResponse abstractResponse) {
                     DescribeLogDirsResponse response = (DescribeLogDirsResponse) abstractResponse;
-                    for (Map.Entry<String, LogDirDescription> responseEntry : logDirDescriptions(response).entrySet()) {
+
+                    Set<TopicPartition> pendingPartitions = new HashSet<>(replicaDirInfoByPartition.keySet());
+                    Map<String, Throwable> directoryFailures = new HashMap<>();
+                    Map<String, LogDirDescription> descriptions = logDirDescriptions(response);
+                    if (descriptions.isEmpty()) {
+                        Errors error = response.data().errorCode() == Errors.NONE.code()
+                                ? Errors.CLUSTER_AUTHORIZATION_FAILED
+                                : Errors.forCode(response.data().errorCode());
+                        handleFailure(error.exception(), pendingPartitions);
+                    }
+
+                    for (Map.Entry<String, LogDirDescription> responseEntry : descriptions.entrySet()) {
                         String logDir = responseEntry.getKey();
                         LogDirDescription logDirInfo = responseEntry.getValue();
 
                         // No replica info will be provided if the log directory is offline
                         if (logDirInfo.error() instanceof KafkaStorageException)
                             continue;
-                        if (logDirInfo.error() != null)
-                            handleFailure(new IllegalStateException(
-                                "The error " + logDirInfo.error().getClass().getName() + " for log directory " + logDir + " in the response from broker " + brokerId + " is illegal"));
 
+                        if (logDirInfo.error() == null) {
                             for (Map.Entry<TopicPartition, ReplicaInfo> replicaInfoEntry : logDirInfo.replicaInfos().entrySet()) {
                                 TopicPartition tp = replicaInfoEntry.getKey();
                                 ReplicaInfo replicaInfo = replicaInfoEntry.getValue();
@@ -3134,7 +3143,18 @@ public class KafkaAdminClient extends AdminClient {
                                             replicaLogDirInfo.getFutureReplicaLogDir(),
                                             replicaLogDirInfo.getFutureReplicaOffsetLag()));
                                 }
+                                pendingPartitions.remove(tp);
                             }
+                        } else {
+                            directoryFailures.put(logDir, logDirInfo.error());
+                        }
+                    }
+
+                    if (!pendingPartitions.isEmpty() && !directoryFailures.isEmpty()) {
+                        List<String> errorAtDir = new ArrayList<>();
+                        directoryFailures.forEach((k, v) -> errorAtDir.add(v.getClass().getName() + " at " + k));
+                        Throwable error = new IllegalStateException("The error " + String.join(", ", errorAtDir) + " in the response from broker " + brokerId + " is illegal");
+                        handleFailure(error, pendingPartitions);
                     }
 
                     for (Map.Entry<TopicPartition, ReplicaLogDirInfo> entry : replicaDirInfoByPartition.entrySet()) {
@@ -3148,6 +3168,13 @@ public class KafkaAdminClient extends AdminClient {
                 void handleFailure(Throwable throwable) {
                     completeAllExceptionally(futures.values(), throwable);
                 }
+
+                void handleFailure(Throwable throwable, Collection<TopicPartition> topicPartitions) {
+                    for (TopicPartition tp: topicPartitions) {
+                        KafkaFutureImpl<ReplicaLogDirInfo> future = futures.get(new TopicPartitionReplica(tp.topic(), tp.partition(), brokerId));
+                        future.completeExceptionally(throwable);
+                    }
+                }
             }, now);
         }