ISSUE-2962: skew calculation fixes

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

@@ -1,26 +1,32 @@
 package com.provectus.kafka.ui.model;
 
 import java.math.BigDecimal;
+import java.math.MathContext;
 import java.util.HashMap;
 import java.util.Map;
 import javax.annotation.Nullable;
 import lombok.AccessLevel;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.admin.TopicDescription;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartitionInfo;
 
 @RequiredArgsConstructor(access = AccessLevel.PRIVATE)
 @Getter
+@Slf4j
 public class PartitionDistributionStats {
 
   // avg skew will show unuseful results on low number of partitions
   private static final int MIN_PARTITIONS_FOR_SKEW_CALCULATION = 50;
 
+  private static final MathContext ROUNDING_MATH_CTX = new MathContext(3);
+
   private final Map<Node, Integer> partitionLeaders;
   private final Map<Node, Integer> partitionsCount;
   private final Map<Node, Integer> inSyncPartitions;
+  private final double avgLeadersCntPerBroker;
   private final double avgPartitionsPerBroker;
   private final boolean skewCanBeCalculated;
 
@@ -32,10 +38,8 @@ public class PartitionDistributionStats {
     var partitionLeaders = new HashMap<Node, Integer>();
     var partitionsReplicated = new HashMap<Node, Integer>();
     var isr = new HashMap<Node, Integer>();
-    int partitionsCnt = 0;
     for (TopicDescription td : stats.getTopicDescriptions().values()) {
       for (TopicPartitionInfo tp : td.partitions()) {
-        partitionsCnt++;
         tp.replicas().forEach(r -> incr(partitionsReplicated, r));
         tp.isr().forEach(r -> incr(isr, r));
         if (tp.leader() != null) {
@@ -43,14 +47,21 @@ public class PartitionDistributionStats {
         }
       }
     }
-    int nodesCount = stats.getClusterDescription().getNodes().size();
-    double avgPartitionsPerBroker = nodesCount == 0 ? 0 : ((double) partitionsCnt) / nodesCount;
+    int nodesWithPartitions = partitionsReplicated.size();
+    int partitionReplications = partitionsReplicated.values().stream().mapToInt(i -> i).sum();
+    var avgPartitionsPerBroker = nodesWithPartitions == 0 ? 0 : ((double) partitionReplications) / nodesWithPartitions;
+
+    int nodesWithLeaders = partitionLeaders.size();
+    int leadersCnt = partitionLeaders.values().stream().mapToInt(i -> i).sum();
+    var avgLeadersCntPerBroker = nodesWithLeaders == 0 ? 0 : ((double) leadersCnt) / nodesWithLeaders;
+
     return new PartitionDistributionStats(
         partitionLeaders,
         partitionsReplicated,
         isr,
+        avgLeadersCntPerBroker,
         avgPartitionsPerBroker,
-        partitionsCnt >= minPartitionsForSkewCalculation
+        partitionReplications >= minPartitionsForSkewCalculation
     );
   }
 
@@ -65,7 +76,7 @@ public class PartitionDistributionStats {
 
   @Nullable
   public BigDecimal leadersSkew(Node node) {
-    return calculateAvgSkew(partitionLeaders.get(node), avgPartitionsPerBroker);
+    return calculateAvgSkew(partitionLeaders.get(node), avgLeadersCntPerBroker);
   }
 
   // Returns difference (in percents) from average value, null if it can't be calculated
@@ -75,6 +86,6 @@ public class PartitionDistributionStats {
       return null;
     }
     value = value == null ? 0 : value;
-    return new BigDecimal((value - avgValue) / avgValue * 100.0);
+    return new BigDecimal((value - avgValue) / avgValue * 100.0).round(ROUNDING_MATH_CTX);
   }
 }

kafka-ui-api/src/test/java/com/provectus/kafka/ui/model/PartitionDistributionStatsTest.java

@@ -20,6 +20,7 @@ class PartitionDistributionStatsTest {
     Node n1 = new Node(1, "n1", 9092);
     Node n2 = new Node(2, "n2", 9092);
     Node n3 = new Node(3, "n3", 9092);
+    Node n4 = new Node(4, "n4", 9092);
 
     var stats = PartitionDistributionStats.create(
         Statistics.builder()
@@ -53,25 +54,30 @@ class PartitionDistributionStatsTest {
     assertThat(stats.getInSyncPartitions())
         .containsExactlyInAnyOrderEntriesOf(Map.of(n1, 3, n2, 3, n3, 1));
 
-    // 4 partitions, 3 brokers = avg partition cnt per broker is 1.333.
+    // Node(partitions): n1(3), n2(4), n3(1), n4(0)
+    // average partitions cnt = (3+4+1) / 3 = 2.666 (counting only nodes with partitions!)
     assertThat(stats.getAvgPartitionsPerBroker())
-        .isCloseTo(1.333, Percentage.withPercentage(1));
+        .isCloseTo(2.666, Percentage.withPercentage(1));
 
-    // Node(partitions): n1(3), n2(4), n3(1)
     assertThat(stats.partitionsSkew(n1))
-        .isCloseTo(BigDecimal.valueOf(125), Percentage.withPercentage(1));
+        .isCloseTo(BigDecimal.valueOf(12.5), Percentage.withPercentage(1));
     assertThat(stats.partitionsSkew(n2))
-        .isCloseTo(BigDecimal.valueOf(200), Percentage.withPercentage(1));
+        .isCloseTo(BigDecimal.valueOf(50), Percentage.withPercentage(1));
     assertThat(stats.partitionsSkew(n3))
-        .isCloseTo(BigDecimal.valueOf(-25), Percentage.withPercentage(1));
+        .isCloseTo(BigDecimal.valueOf(-62.5), Percentage.withPercentage(1));
+    assertThat(stats.partitionsSkew(n4))
+        .isCloseTo(BigDecimal.valueOf(-100), Percentage.withPercentage(1));
 
-    //  Node(leaders): n1(2), n2(1), n3(0)
+    //  Node(leaders): n1(2), n2(1), n3(0), n4(0)
+    //  average leaders cnt = (2+1) / 2 = 1.5 (counting only nodes with leaders!)
     assertThat(stats.leadersSkew(n1))
-        .isCloseTo(BigDecimal.valueOf(50), Percentage.withPercentage(1));
+        .isCloseTo(BigDecimal.valueOf(33.33), Percentage.withPercentage(1));
     assertThat(stats.leadersSkew(n2))
-        .isCloseTo(BigDecimal.valueOf(-25), Percentage.withPercentage(1));
+        .isCloseTo(BigDecimal.valueOf(-33.33), Percentage.withPercentage(1));
     assertThat(stats.leadersSkew(n3))
         .isCloseTo(BigDecimal.valueOf(-100), Percentage.withPercentage(1));
+    assertThat(stats.leadersSkew(n4))
+        .isCloseTo(BigDecimal.valueOf(-100), Percentage.withPercentage(1));
   }
 
 }