fixed dirty mistakes

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/service/MetricsUpdateService.java

@@ -6,9 +6,6 @@ import com.provectus.kafka.ui.zookeeper.ZookeeperService;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
 import org.springframework.stereotype.Service;
-import reactor.core.publisher.Mono;
-
-import java.util.logging.Level;
 
 @Service
 @RequiredArgsConstructor
@@ -19,10 +16,8 @@ public class MetricsUpdateService {
     private final ZookeeperService zookeeperService;
 
     public void updateMetrics(KafkaCluster kafkaCluster) {
-        Mono.just(true)
-                .doOnNext(s -> {
-                    log.debug("Start getting metrics for kafkaCluster: {}", kafkaCluster.getName());
-                    kafkaService.loadClusterMetrics(kafkaCluster);
-                }).doOnNext(s -> zookeeperService.checkZookeeperStatus(kafkaCluster)).subscribe();
+        log.debug("Start getting metrics for kafkaCluster: {}", kafkaCluster.getName());
+        kafkaService.loadClusterMetrics(kafkaCluster);
+        zookeeperService.checkZookeeperStatus(kafkaCluster);
     }
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/kafka/KafkaService.java

@@ -27,25 +27,22 @@ public class KafkaService {
 
     @SneakyThrows
     public void loadClusterMetrics(KafkaCluster kafkaCluster) {
-        Mono.just(false)
-            .map(b -> {
-                log.debug("Start getting metrics for kafkaCluster: {}", kafkaCluster.getName());
-                if (kafkaCluster.getAdminClient() != null) {
-                    b = isAdminClientConnected(kafkaCluster);
-                }
-                if (kafkaCluster.getAdminClient() == null || !b) {
-                    b = createAdminClient(kafkaCluster);
-                }
-                if (!b) {
-                    kafkaCluster.getCluster().setStatus(ServerStatus.OFFLINE);
-                }
-                return b;
-            }).filter(s -> s)
-            .doOnNext(s -> {
-                kafkaCluster.getCluster().setId(kafkaCluster.getId());
-                kafkaCluster.getCluster().setStatus(ServerStatus.ONLINE);
-            }).doOnNext(s -> loadMetrics(kafkaCluster))
-            .doOnNext(s -> loadTopicsData(kafkaCluster)).subscribe();
+        var isConnected = false;
+        log.debug("Start getting metrics for kafkaCluster: {}", kafkaCluster.getName());
+        if (kafkaCluster.getAdminClient() != null) {
+            isConnected = isAdminClientConnected(kafkaCluster);
+        }
+        if (kafkaCluster.getAdminClient() == null || !isConnected) {
+            isConnected = createAdminClient(kafkaCluster);
+        }
+        if (!isConnected) {
+            kafkaCluster.getCluster().setStatus(ServerStatus.OFFLINE);
+            return;
+        }
+        kafkaCluster.getCluster().setId(kafkaCluster.getId());
+        kafkaCluster.getCluster().setStatus(ServerStatus.ONLINE);
+        loadMetrics(kafkaCluster);
+        loadTopicsData(kafkaCluster);
     }
 
 
@@ -114,19 +111,19 @@ public class KafkaService {
         ListTopicsOptions listTopicsOptions = new ListTopicsOptions();
         listTopicsOptions.listInternal(true);
         ClusterUtil.toMono(adminClient.listTopics(listTopicsOptions).names())
-                .map(tl -> {
-            kafkaCluster.getCluster().setTopicCount(tl.size());
-            DescribeTopicsResult topicDescriptionsWrapper = adminClient.describeTopics(tl);
-            Map<String, KafkaFuture<TopicDescription>> topicDescriptionFuturesMap = topicDescriptionsWrapper.values();
-            resetMetrics(kafkaCluster);
-            return topicDescriptionFuturesMap.entrySet();
-        })
-        .flatMapMany(Flux::fromIterable)
-        .flatMap(s -> ClusterUtil.toMono(s.getValue()))
-        .flatMap(e -> collectTopicData(kafkaCluster, e))
-        .collectList()
-        .doOnNext(kafkaCluster::setTopics)
-        .subscribe();
+            .map(tl -> {
+                kafkaCluster.getCluster().setTopicCount(tl.size());
+                DescribeTopicsResult topicDescriptionsWrapper = adminClient.describeTopics(tl);
+                Map<String, KafkaFuture<TopicDescription>> topicDescriptionFuturesMap = topicDescriptionsWrapper.values();
+                resetMetrics(kafkaCluster);
+                return topicDescriptionFuturesMap.entrySet();
+            })
+            .flatMapMany(Flux::fromIterable)
+            .flatMap(s -> ClusterUtil.toMono(s.getValue()))
+            .flatMap(e -> collectTopicData(kafkaCluster, e))
+            .collectList()
+            .doOnNext(kafkaCluster::setTopics)
+            .subscribe();
     }
 
     private void resetMetrics(KafkaCluster kafkaCluster) {
@@ -268,6 +265,6 @@ public class KafkaService {
                 .values()
                 .values()
                 .iterator()
-                .next()).map(s -> Mono.just(true));
+                .next()).subscribe();
     }
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/zookeeper/ZookeeperService.java

@@ -5,7 +5,6 @@ import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
 import org.I0Itec.zkclient.ZkClient;
 import org.springframework.stereotype.Service;
-import reactor.core.publisher.Mono;
 
 @Service
 @RequiredArgsConstructor
@@ -13,21 +12,19 @@ import reactor.core.publisher.Mono;
 public class ZookeeperService {
 
     public void checkZookeeperStatus(KafkaCluster kafkaCluster) {
-        Mono.just(false)
-                .doOnNext(isConnected -> {
-                    log.debug("Start getting Zookeeper metrics for kafkaCluster: {}", kafkaCluster.getName());
-                    if (kafkaCluster.getZkClient() != null) {
-                        isConnected = isZkClientConnected(kafkaCluster);
-                    }
-                    if (kafkaCluster.getZkClient() == null || !isConnected) {
-                        isConnected = createZookeeperConnection(kafkaCluster);
-                    }
-                    if (!isConnected) {
-                        kafkaCluster.getBrokersMetrics().setZooKeeperStatus(ZooKeeperConstants.OFFLINE);
-                        return;
-                    }
-                    kafkaCluster.getBrokersMetrics().setZooKeeperStatus(ZooKeeperConstants.ONLINE);
-                }).subscribe();
+        var isConnected = false;
+        log.debug("Start getting Zookeeper metrics for kafkaCluster: {}", kafkaCluster.getName());
+        if (kafkaCluster.getZkClient() != null) {
+            isConnected = isZkClientConnected(kafkaCluster);
+        }
+        if (kafkaCluster.getZkClient() == null || !isConnected) {
+            isConnected = createZookeeperConnection(kafkaCluster);
+        }
+        if (!isConnected) {
+            kafkaCluster.getBrokersMetrics().setZooKeeperStatus(ZooKeeperConstants.OFFLINE);
+            return;
+        }
+        kafkaCluster.getBrokersMetrics().setZooKeeperStatus(ZooKeeperConstants.ONLINE);
     }
 
     private boolean createZookeeperConnection(KafkaCluster kafkaCluster) {