improve logs in PastureCoordinator

PlaytikaOSS · Mar 13, 2024 · a72f7bf · a72f7bf
1 parent 0fda531
commit a72f7bf
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 2 deletions.
diff --git a/kafka/src/main/java/com/playtika/shepherd/inernal/CheckedHerd.java b/kafka/src/main/java/com/playtika/shepherd/inernal/CheckedHerd.java
@@ -4,6 +4,7 @@ public class CheckedHerd implements Herd {
 
     private final Herd herd;
     private boolean requested = false;
+    private Exception stackTrace;
 
     public static Herd checked(Herd herd){
         return new CheckedHerd(herd);
@@ -16,12 +17,13 @@ private CheckedHerd(Herd herd) {
     @Override
     public Population getPopulation(){
         if(requested){
-            throw new IllegalStateException("Should be called only once on rebalance");
+            throw new IllegalStateException("Should be called only once on rebalance", stackTrace);
         }
         try {
             return herd.getPopulation();
         } finally {
             requested = true;
+            stackTrace = new RuntimeException();
         }
     }
 

diff --git a/kafka/src/main/java/com/playtika/shepherd/inernal/PastureCoordinator.java b/kafka/src/main/java/com/playtika/shepherd/inernal/PastureCoordinator.java
@@ -33,6 +33,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 import static com.playtika.shepherd.inernal.ProtocolHelper.decompress;
 import static com.playtika.shepherd.inernal.ProtocolHelper.deserializeAssignment;
@@ -179,7 +180,21 @@ protected Map<String, ByteBuffer> onLeaderElected(String leaderId,
         Population population = herd.getPopulation();
         leaderElected = leaderId;
 
-        logger.info("Will rebalance population: [{}]", toBytes(population.getSheep()));
+        if(logger.isDebugEnabled()) {
+            logger.debug("""
+                            Will rebalance population of size=[{}] among members count=[{}],
+                            members=[{}],
+                            population=[{}]""",
+                    population.getSheep().size(), allMemberMetadata.size(),
+                    allMemberMetadata.stream().map(JoinGroupResponseMember::memberId).collect(Collectors.joining(", ")),
+                    toBytes(population.getSheep()));
+        } else {
+            logger.info("""
+                            Will rebalance population of size=[{}] among members count=[{}],
+                            members=[{}]""",
+                    population.getSheep().size(), allMemberMetadata.size(),
+                    allMemberMetadata.stream().map(JoinGroupResponseMember::memberId).collect(Collectors.joining(", ")));
+        }
 
         return assignor.performAssignment(leaderId, protocol,
                 population.getSheep(), population.getVersion(),