Added segmentSize info to cluster, broker, topic and topicPartition level

docker/kafka-clusters-only.yaml

@@ -5,10 +5,10 @@ services:
   zookeeper0:
     image: confluentinc/cp-zookeeper:5.1.0
     environment:
-      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_CLIENT_PORT: 2183
       ZOOKEEPER_TICK_TIME: 2000
     ports:
-     - 2181:2181
+     - 2183:2183
 
   kafka0:
     image: confluentinc/cp-kafka:5.1.0
@@ -19,11 +19,28 @@ services:
       - 9997:9997
     environment:
       KAFKA_BROKER_ID: 1
-      KAFKA_ZOOKEEPER_CONNECT: zookeeper0:2181
-      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:29091,PLAINTEXT_HOST://localhost:9091,PLAIN://kafka0:29090
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper0:2183
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka0:9092,PLAINTEXT_HOST://localhost:29091 #,PLAIN://kafka0:29090
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT #,PLAIN:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 2
+      JMX_PORT: 9997
+      KAFKA_JMX_OPTS: -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=localhost -Dcom.sun.management.jmxremote.rmi.port=9997
+
+  kafka01:
+    image: confluentinc/cp-kafka:5.1.0
+    depends_on:
+      - zookeeper0
+    ports:
+      - 29093:29093
+      - 9999:9999
+    environment:
+      KAFKA_BROKER_ID: 2
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper0:2183
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka01:9092,PLAINTEXT_HOST://localhost:29093,PLAIN://kafka0:29090
       KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,PLAIN:PLAINTEXT
       KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
-      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 2
       JMX_PORT: 9997
       KAFKA_JMX_OPTS: -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=localhost -Dcom.sun.management.jmxremote.rmi.port=9997
 
@@ -32,9 +49,9 @@ services:
     depends_on:
       - kafka0
     command: "bash -c 'echo Waiting for Kafka to be ready... && \
-                cub kafka-ready -b kafka0:29090 1 20 && \
-                kafka-topics --create --topic users --partitions 2 --replication-factor 2 --if-not-exists --zookeeper zookeeper0:2181 && \
-                kafka-topics --create --topic messages --partitions 3 --replication-factor 2 --if-not-exists --zookeeper zookeeper0:2181'"
+                cub kafka-ready -b kafka0:9092 1 20 && \
+                kafka-topics --create --topic users --partitions 2 --replication-factor 2 --if-not-exists --zookeeper zookeeper0:2183 && \
+                kafka-topics --create --topic messages --partitions 3 --replication-factor 2 --if-not-exists --zookeeper zookeeper0:2183'"
     environment:
       KAFKA_BROKER_ID: ignored
       KAFKA_ZOOKEEPER_CONNECT: ignored
@@ -44,10 +61,10 @@ services:
   zookeeper1:
     image: confluentinc/cp-zookeeper:5.1.0
     environment:
-      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_CLIENT_PORT: 2182
       ZOOKEEPER_TICK_TIME: 2000
     ports:
-      - 2182:2181
+      - 2182:2182
 
   kafka1:
     image: confluentinc/cp-kafka:5.1.0
@@ -58,8 +75,8 @@ services:
       - 9998:9998
     environment:
       KAFKA_BROKER_ID: 1
-      KAFKA_ZOOKEEPER_CONNECT: zookeeper1:2181
-      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:29092,PLAINTEXT_HOST://localhost:9092,PLAIN://kafka1:29090
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper1:2182
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:9092,PLAINTEXT_HOST://localhost:29092,PLAIN://localhost:29090
       KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,PLAIN:PLAINTEXT
       KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 2
@@ -69,13 +86,13 @@ services:
   kafka-init-topics1:
     image: confluentinc/cp-kafka:5.1.0
     depends_on:
-      - kafka1
+     - kafka1
     command: "bash -c 'echo Waiting for Kafka to be ready... && \
-                cub kafka-ready -b kafka1:29090 1 20 && \
-                kafka-topics --create --topic users --partitions 3 --replication-factor 2 --if-not-exists --zookeeper zookeeper1:2181 && \
-                kafka-topics --create --topic messages --partitions 2 --replication-factor 2 --if-not-exists --zookeeper zookeeper1:2181'"
+               cub kafka-ready -b kafka1:9092 20 && \
+               kafka-topics --create --topic secondUsers --partitions 3 --replication-factor 2 --if-not-exists --zookeeper zookeeper1:2182 && \
+               kafka-topics --create --topic secondMessages --partitions 2 --replication-factor 2 --if-not-exists --zookeeper zookeeper1:2182'"
     environment:
-      KAFKA_BROKER_ID: ignored
-      KAFKA_ZOOKEEPER_CONNECT: ignored
+     KAFKA_BROKER_ID: ignored
+     KAFKA_ZOOKEEPER_CONNECT: ignored
     networks:
-      - default
+     - default

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/mapper/BrokersMetricsMapper.java

@@ -1,13 +0,0 @@
-package com.provectus.kafka.ui.cluster.mapper;
-
-import com.provectus.kafka.ui.cluster.model.InternalClusterMetrics;
-import com.provectus.kafka.ui.model.BrokersMetrics;
-import org.mapstruct.Mapper;
-
-@Mapper(componentModel = "spring")
-public interface BrokersMetricsMapper {
-
-    InternalClusterMetrics toBrokersMetricsDto (BrokersMetrics brokersMetrics);
-
-    BrokersMetrics toBrokersMetrics (InternalClusterMetrics brokersMetrics);
-}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/mapper/ClusterDtoMapper.java

@@ -1,12 +0,0 @@
-package com.provectus.kafka.ui.cluster.mapper;
-
-import com.provectus.kafka.ui.cluster.model.KafkaCluster;
-import com.provectus.kafka.ui.model.Cluster;
-import org.mapstruct.Mapper;
-
-@Mapper(componentModel = "spring")
-public interface ClusterDtoMapper {
-
-    KafkaCluster toInternalCluster(Cluster cluster);
-    Cluster toClusterDto(KafkaCluster cluster);
-}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/model/InternalSegmentSize.java

@@ -1,12 +0,0 @@
-package com.provectus.kafka.ui.cluster.model;
-
-import lombok.Data;
-import lombok.RequiredArgsConstructor;
-
-@Data
-@RequiredArgsConstructor
-public class InternalSegmentSize {
-
-    private final String replicaName;
-    private final long segmentSize;
-}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/model/InternalSegmentSizeDto.java

@@ -0,0 +1,14 @@
+package com.provectus.kafka.ui.cluster.model;
+
+import lombok.Builder;
+import lombok.Data;
+
+import java.util.Map;
+
+@Data
+@Builder(toBuilder = true)
+public class InternalSegmentSizeDto {
+
+    private final Map<String, InternalTopic> internalTopicWithSegmentSize;
+    private final InternalClusterMetrics clusterMetricsWithSegmentSize;
+}

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

@@ -16,7 +16,7 @@ public class MetricsUpdateService {
     private final KafkaService kafkaService;
 
     public Mono<KafkaCluster> updateMetrics(KafkaCluster kafkaCluster) {
-        log.debug("Start getting metrics for kafkaCluster: {}", kafkaCluster);
+        log.debug("Start getting metrics for kafkaCluster: {}", kafkaCluster.getName());
         return kafkaService.getUpdatedCluster(kafkaCluster);
     }
 }

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

@@ -43,8 +43,8 @@ public class KafkaService {
                             getTopicsData(ac).flatMap( topics ->
                                 loadTopicsConfig(ac, topics.stream().map(InternalTopic::getName).collect(Collectors.toList()))
                                         .map( configs -> mergeWithConfigs(topics, configs))
-                                    .flatMap(it -> updateTopicSegmentSize(ac, clusterMetrics, it))
-                            ).map( topics -> buildFromData(cluster, clusterMetrics, topics))
+                                    .flatMap(it -> updateSegmentMetrics(ac, clusterMetrics, it))
+                            ).map( segmentSizeDto -> buildFromData(cluster, segmentSizeDto))
                         )
         ).onErrorResume(
                 e -> Mono.just(cluster.toBuilder()
@@ -54,7 +54,10 @@ public class KafkaService {
         );
     }
 
-    private KafkaCluster buildFromData(KafkaCluster currentCluster, InternalClusterMetrics brokersMetrics, Map<String, InternalTopic> topics) {
+    private KafkaCluster buildFromData(KafkaCluster currentCluster, InternalSegmentSizeDto segmentSizeDto) {
+
+        var topics = segmentSizeDto.getInternalTopicWithSegmentSize();
+        var brokersMetrics = segmentSizeDto.getClusterMetricsWithSegmentSize();
 
         InternalClusterMetrics.InternalClusterMetricsBuilder metricsBuilder = brokersMetrics.toBuilder();
 
@@ -142,7 +145,7 @@ public class KafkaService {
                             return metricsBuilder.build();
                         }
                     )
-                ).flatMap(c -> updateClusterSegmentSize(client, c));
+                );
     }
 
 
@@ -229,7 +232,6 @@ public class KafkaService {
 
     public Mono<List<ConsumerGroup>> getConsumerGroups(KafkaCluster cluster) {
         var adminClient =  this.createAdminClient(cluster);
-
         return ClusterUtil.toMono(adminClient.listConsumerGroups().all())
                 .flatMap(s -> ClusterUtil.toMono(adminClient
                         .describeConsumerGroups(s.stream().map(ConsumerGroupListing::groupId).collect(Collectors.toList())).all()))
@@ -247,64 +249,57 @@ public class KafkaService {
                     .next());
     }
 
-    private Mono<Map<String, InternalTopic>> updateTopicSegmentSize(AdminClient ac, InternalClusterMetrics clusterMetrics, Map<String, InternalTopic> internalTopic) {
-            return ClusterUtil.toMono(ac.describeTopics(internalTopic.keySet()).all()).flatMap(topic ->
-                {
-                    Map<TopicPartition, Integer> leaders = topic.values().stream()
-                            .flatMap(t -> t.partitions().stream()
-                                .flatMap(t1 -> {
-                                    Map <TopicPartition, Integer > result = new HashMap<>();
-                                    result.put(new TopicPartition(t.name(), t1.partition()), t1.leader().id());
-                                    return Stream.of(result);
-                                }))
-                            .reduce((map1, map2) -> Stream.concat(map1.entrySet().stream(), map2.entrySet().stream())
-                                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))).orElseThrow();
-                    return ClusterUtil.toMono(ac.describeLogDirs(clusterMetrics.getInternalBrokerMetrics().keySet()).all())
-                        .map(log -> {
-                            Map<TopicPartition, Long> partitionSegmentSize = leaders.entrySet().stream()
+    private Mono<InternalSegmentSizeDto> updateSegmentMetrics(AdminClient ac, InternalClusterMetrics clusterMetrics, Map<String, InternalTopic> internalTopic) {
+        return ClusterUtil.toMono(ac.describeTopics(internalTopic.keySet()).all()).flatMap(topic -> {
+            Map<TopicPartition, Integer> leaders = topic.values().stream()
+                    .flatMap(t -> t.partitions().stream()
+                            .flatMap(t1 -> {
+                                Map<TopicPartition, Integer> result = new HashMap<>();
+                                result.put(new TopicPartition(t.name(), t1.partition()), t1.leader().id());
+                                return Stream.of(result);
+                            }))
+                    .reduce((map1, map2) -> Stream.concat(map1.entrySet().stream(), map2.entrySet().stream())
+                            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))).orElseThrow();
+
+            return ClusterUtil.toMono(ac.describeLogDirs(clusterMetrics.getInternalBrokerMetrics().keySet()).all())
+                    .map(log -> {
+                        var partitionSegmentSize = leaders.entrySet().stream()
                                 .flatMap(l -> {
                                     Map<TopicPartition, Long> result = new HashMap<>();
                                     result.put(l.getKey(), log.get(l.getValue()).values().stream().mapToLong(e -> e.replicaInfos.get(l.getKey()).size).sum());
                                     return Stream.of(result);
                                 }).reduce((map1, map2) -> Stream.concat(map1.entrySet().stream(), map2.entrySet().stream())
-                                            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))).orElseThrow();
-                            return internalTopic.keySet().stream().flatMap(k -> {
-                                Map<String, InternalTopic> result = new HashMap<>();
-                                result.put(k, internalTopic.get(k).toBuilder()
-                                        .segmentSize(partitionSegmentSize.entrySet().stream().filter(e -> e.getKey().topic().equals(k)).mapToLong(Map.Entry::getValue).sum())
-                                        .partitionSegmentSize(partitionSegmentSize.entrySet().stream().filter(e -> e.getKey().topic().equals(k)).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))).build());
-                                return Stream.of(result);
-                            }).reduce((map1, map2) -> Stream.concat(map1.entrySet().stream(), map2.entrySet().stream())
-                                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))).orElseThrow();
+                                        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))).orElseThrow();
+
+                        var resultTopicMetrics = internalTopic.keySet().stream().flatMap(k -> {
+                            Map<String, InternalTopic> result = new HashMap<>();
+                            result.put(k, internalTopic.get(k).toBuilder()
+                                    .segmentSize(partitionSegmentSize.entrySet().stream().filter(e -> e.getKey().topic().equals(k)).mapToLong(Map.Entry::getValue).sum())
+                                    .partitionSegmentSize(partitionSegmentSize.entrySet().stream().filter(e -> e.getKey().topic().equals(k)).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))).build());
+                            return Stream.of(result);
+                        }).reduce((map1, map2) -> Stream.concat(map1.entrySet().stream(), map2.entrySet().stream())
+                                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))).orElseThrow();
 
-                            });
-                });
+                        var resultBrokerMetrics = clusterMetrics.getInternalBrokerMetrics().entrySet().stream().map(
+                                e -> {
+                                    var brokerSegmentSize = log.get(e.getKey()).values().stream()
+                                            .mapToLong(v -> v.replicaInfos.values().stream()
+                                                    .mapToLong(r -> r.size).sum()).sum();
+                                    InternalBrokerMetrics tempBrokerMetrics = InternalBrokerMetrics.builder().segmentSize(brokerSegmentSize).build();
+                                    return Collections.singletonMap(e.getKey(), tempBrokerMetrics);
+                                })
+                                .reduce((map1, map2) -> Stream.concat(map1.entrySet().stream(), map2.entrySet().stream())
+                                        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))).orElseThrow();
+
+                        var resultClusterMetrics = clusterMetrics.toBuilder()
+                                .internalBrokerMetrics(resultBrokerMetrics)
+                                .segmentSize(partitionSegmentSize.values().stream().reduce(Long::sum).orElseThrow())
+                                .build();
+
+                        return InternalSegmentSizeDto.builder()
+                                .clusterMetricsWithSegmentSize(resultClusterMetrics)
+                                .internalTopicWithSegmentSize(resultTopicMetrics).build();
+                    });
+        });
     }
-
-    private Mono<InternalClusterMetrics> updateClusterSegmentSize (AdminClient client, InternalClusterMetrics clusterMetrics) {
-        return ClusterUtil.toMono(client.describeLogDirs(clusterMetrics.getInternalBrokerMetrics().keySet()).all())
-                .map(l -> {
-                    var replicasInfo = l.values().stream()
-                            .flatMap(v -> Stream.of(v.values()))
-                            .flatMap(v -> Stream.of(v.stream().map(s -> s.replicaInfos)))
-                            .findFirst().orElseThrow().collect(Collectors.toList());
-
-                    var segmentSize = replicasInfo.stream()
-                            .mapToLong(t -> t.values().stream()
-                                    .mapToLong(e -> e.size).sum())
-                            .sum();
-
-                    var internalBrokerMetrics = clusterMetrics.getInternalBrokerMetrics().entrySet().stream().map(
-                            e -> {
-                                var brokerSegmentSize = l.get(e.getKey()).values().stream()
-                                        .mapToLong(v -> v.replicaInfos.values().stream()
-                                                .mapToLong(r -> r.size).sum()).sum();
-                                InternalBrokerMetrics tempBrokerMetrics = InternalBrokerMetrics.builder().segmentSize(brokerSegmentSize).build();
-                                return Collections.singletonMap(e.getKey(), tempBrokerMetrics);
-                            })
-                            .reduce((map1, map2) -> Stream.concat(map1.entrySet().stream(), map2.entrySet().stream())
-                                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
-                    return clusterMetrics.toBuilder().segmentSize(segmentSize).internalBrokerMetrics(internalBrokerMetrics.orElseThrow()).build();
-                });
-        }
 }

kafka-ui-api/src/main/resources/application-local.yml

@@ -3,8 +3,12 @@ kafka:
     -
       name: local
       bootstrapServers: localhost:29091
-      zookeeper: localhost:2181
+      zookeeper: localhost:2183
     -
       name: secondLocal
       bootstrapServers: localhost:29092
-      zookeeper: localhost:2182
+      zookeeper: localhost:2182
+    -
+      name: localReplica
+      bootstrapServers: localhost:29093
+      zookeeper: localhost:2183