KAFKA-14594: Move LogDirsCommand to tools module (#13122)

Reviewers: Mickael Maison <mickael.maison@gmail.com>

bin/kafka-log-dirs.sh

@@ -14,4 +14,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-exec $(dirname $0)/kafka-run-class.sh kafka.admin.LogDirsCommand "$@"
+exec $(dirname $0)/kafka-run-class.sh org.apache.kafka.tools.LogDirsCommand "$@"

bin/windows/kafka-log-dirs.bat

@@ -14,4 +14,4 @@ rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 rem See the License for the specific language governing permissions and
 rem limitations under the License.
 
-"%~dp0kafka-run-class.bat" kafka.admin.LogDirsCommand %*
+"%~dp0kafka-run-class.bat" org.apache.kafka.tools.LogDirsCommand %*

clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java

@@ -945,7 +945,37 @@ synchronized public AlterReplicaLogDirsResult alterReplicaLogDirs(
     @Override
     synchronized public DescribeLogDirsResult describeLogDirs(Collection<Integer> brokers,
                                                               DescribeLogDirsOptions options) {
-        throw new UnsupportedOperationException("Not implemented yet");
+        Map<Integer, Map<String, LogDirDescription>> unwrappedResults = new HashMap<>();
+
+        for (Integer broker : brokers) {
+            unwrappedResults.putIfAbsent(broker, new HashMap<>());
+        }
+
+        for (Map.Entry<String, TopicMetadata> entry : allTopics.entrySet()) {
+            String topicName = entry.getKey();
+            TopicMetadata topicMetadata = entry.getValue();
+            // For tests, we make the assumption that there will always be only 1 entry.
+            List<String> partitionLogDirs = topicMetadata.partitionLogDirs;
+            List<TopicPartitionInfo> topicPartitionInfos = topicMetadata.partitions;
+            for (TopicPartitionInfo topicPartitionInfo : topicPartitionInfos) {
+                List<Node> nodes = topicPartitionInfo.replicas();
+                for (Node node : nodes) {
+                    Map<String, LogDirDescription> logDirDescriptionMap = unwrappedResults.get(node.id());
+                    LogDirDescription logDirDescription = logDirDescriptionMap.getOrDefault(partitionLogDirs.get(0), new LogDirDescription(null, new HashMap<>()));
+                    logDirDescription.replicaInfos().put(new TopicPartition(topicName, topicPartitionInfo.partition()), new ReplicaInfo(0, 0, false));
+                }
+            }
+        }
+
+        Map<Integer, KafkaFuture<Map<String, LogDirDescription>>> results = new HashMap<>();
+
+        for (Map.Entry<Integer, Map<String, LogDirDescription>> entry : unwrappedResults.entrySet()) {
+            KafkaFutureImpl<Map<String, LogDirDescription>> kafkaFuture = new KafkaFutureImpl<>();
+            kafkaFuture.complete(entry.getValue());
+            results.put(entry.getKey(), kafkaFuture);
+        }
+
+        return new DescribeLogDirsResult(results);
     }
 
     @Override