Remove empty result in describeLogDirsByBrokerAndTopic

Author: tinawenqiao

src/main/java/org/gnuhpc/bigdata/controller/KafkaController.java

@@ -5,7 +5,6 @@
 import io.swagger.annotations.ApiResponse;
 import io.swagger.annotations.ApiResponses;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
@@ -86,15 +85,16 @@ public Map<Integer, List<String>> listLogDirs(
     return kafkaAdminService.listLogDirsByBroker(brokerList);
   }
 
-  @GetMapping(value = "/brokers/logdirs/detail")
+  @PostMapping(value = "/brokers/logdirs/detail")
   @ApiOperation(value = "Describe log dirs by broker list and topic list")
   public Map<Integer, Map<String, LogDirInfo>> describeLogDirs(
       @RequestParam(required = false) List<Integer> brokerList,
       @RequestParam(required = false) List<String> logDirList,
-      @RequestParam(required = false) Map<String, List<Integer>> topicPartitionMap) {
+      @RequestBody(required = false) Map<String, List<Integer>> topicPartitionMap) {
     return kafkaAdminService.describeLogDirsByBrokerAndTopic(brokerList, logDirList, topicPartitionMap);
   }
 
+
   @GetMapping(value = "/brokers/replicalogdirs")
   @ApiOperation(value = "Describe replica log dirs.")
   public Map<TopicPartitionReplica, ReplicaLogDirInfo> describeReplicaLogDirs(

src/main/java/org/gnuhpc/bigdata/service/KafkaAdminService.java

@@ -101,7 +101,6 @@
 import org.apache.kafka.common.internals.Topic;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.requests.DescribeLogDirsResponse.LogDirInfo;
-import org.apache.kafka.common.serialization.Serdes;
 import org.gnuhpc.bigdata.CollectionConvertor;
 import org.gnuhpc.bigdata.componet.OffsetStorage;
 import org.gnuhpc.bigdata.config.KafkaConfig;
@@ -502,11 +501,13 @@ public Map<Integer, Map<String, LogDirInfo>> describeLogDirsByBrokerAndTopic(
       log.warn("Describe log dirs exception:" + exception);
       throw new ApiException("Describe log dirs exception:" + exception);
     } finally {
+      log.info("After describe log dir, result is:" + logDirInfosByBroker);
       if (logDirList != null && !logDirList.isEmpty()) {
         logDirInfosByBroker.entrySet().forEach(e -> {
           e.getValue().entrySet().removeIf(m -> !logDirList.contains(m.getKey()));
         });
       }
+      log.info("After describe log dir filtered by logdirList, result is:" + logDirInfosByBroker);
       if (topicPartitionMap != null && !topicPartitionMap.isEmpty()) {
         logDirInfosByBroker
             .entrySet()
@@ -528,6 +529,10 @@ public Map<Integer, Map<String, LogDirInfo>> describeLogDirsByBrokerAndTopic(
       }
     }
 
+    log.info("After describe log dir filtered by topicPartitionMap, result is:" + logDirInfosByBroker);
+    logDirInfosByBroker.entrySet().forEach(e->e.getValue().entrySet().removeIf(m->m.getValue().replicaInfos.isEmpty()));
+    logDirInfosByBroker.entrySet().removeIf(e->e.getValue().isEmpty());
+
     return logDirInfosByBroker;
   }