Broker增加服务是否存活接口

km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/broker/BrokerService.java

@@ -67,4 +67,8 @@ public interface BrokerService {
      * 获取总的Broker数
      */
     Integer countAllBrokers();
+
+    boolean allServerDown(Long clusterPhyId);
+
+    boolean existServerDown(Long clusterPhyId);
 }

km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/broker/impl/BrokerServiceImpl.java

@@ -262,14 +262,32 @@ public class BrokerServiceImpl extends BaseVersionControlService implements Brok
         return version;
     }
 
-
-
     @Override
     public Integer countAllBrokers() {
         LambdaQueryWrapper<BrokerPO> lambdaQueryWrapper = new LambdaQueryWrapper<>();
         return brokerDAO.selectCount(lambdaQueryWrapper);
     }
 
+    @Override
+    public boolean allServerDown(Long clusterPhyId) {
+        List<BrokerPO> poList = this.getAllBrokerPOsFromDB(clusterPhyId);
+        if (ValidateUtils.isEmptyList(poList)) {
+            return false;
+        }
+
+        return poList.stream().filter(elem -> elem.getStatus().equals(Constant.DOWN)).count() == poList.size();
+    }
+
+    @Override
+    public boolean existServerDown(Long clusterPhyId) {
+        List<BrokerPO> poList = this.getAllBrokerPOsFromDB(clusterPhyId);
+        if (ValidateUtils.isEmptyList(poList)) {
+            return false;
+        }
+
+        return poList.stream().filter(elem -> elem.getStatus().equals(Constant.DOWN)).count() > 0;
+    }
+
     /**************************************************** private method ****************************************************/
 
     private List<Broker> listAllBrokersAndUpdateCache(Long clusterPhyId) {