Merge pull request #42 from PlaytikaOSS/feature/pasture-logs

Removed CheckedHerd and improve logs in PastureCoordinator

Author: kptfh

kafka-functional-tests/src/test/java/com/playtika/shepherd/inernal/ShepherdTest.java

@@ -15,7 +15,6 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Stream;
 
-import static com.playtika.shepherd.inernal.CheckedHerd.checked;
 import static com.playtika.shepherd.inernal.utils.BytesUtils.toBytes;
 import static java.time.Duration.ofSeconds;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -35,7 +34,7 @@ public void shouldBalanceStaticHerd() {
         ByteBuffer cow2 = ByteBuffer.wrap(new byte[]{0});
         AtomicInteger pasturesCountParameter = new AtomicInteger();
 
-        Herd herd = checked(new Herd() {
+        Herd herd = new Herd() {
             @Override
             public Population getPopulation(List<JoinGroupResponseData.JoinGroupResponseMember> allMemberMetadata) {
                 pasturesCountParameter.set(allMemberMetadata.size());
@@ -45,7 +44,7 @@ public Population getPopulation(List<JoinGroupResponseData.JoinGroupResponseMemb
             @Override
             public void reset() {
             }
-        });
+        };
 
 
         LinkedBlockingQueue<ByteBuffer> cows1 = new LinkedBlockingQueue<>();
@@ -128,7 +127,7 @@ public void shouldBalanceDynamicHerd() {
         AtomicInteger version = new AtomicInteger(1);
         AtomicInteger pasturesCountParameter = new AtomicInteger();
 
-        Herd herd = checked(new Herd() {
+        Herd herd = new Herd() {
             @Override
             public Population getPopulation(List<JoinGroupResponseData.JoinGroupResponseMember> allMemberMetadata) {
                 pasturesCountParameter.set(allMemberMetadata.size());
@@ -138,7 +137,7 @@ public Population getPopulation(List<JoinGroupResponseData.JoinGroupResponseMemb
             @Override
             public void reset() {
             }
-        });
+        };
 
         LinkedBlockingQueue<ByteBuffer> cows1 = new LinkedBlockingQueue<>();
         PastureListener<ByteBuffer> rebalanceListener1 = new PastureListener<>() {

kafka/src/main/java/com/playtika/shepherd/KafkaPullFarm.java

@@ -16,7 +16,6 @@
 import java.util.List;
 import java.util.Map;
 
-import static com.playtika.shepherd.inernal.CheckedHerd.checked;
 import static com.playtika.shepherd.serde.SerDeUtils.BYTE_BUFFER_DE_SER;
 import static com.playtika.shepherd.serde.SerDeUtils.getSerDe;
 
@@ -52,7 +51,7 @@ private  <Breed> Pasture addBreedingPasture(Herd<Breed> herd, SerDe<Breed> serDe
                 .setGroupId(herd.getName())
                 .setProperties(properties)
                 .setRebalanceListener(pullHerd)
-                .setHerd(checked(pullHerd))
+                .setHerd(pullHerd)
                 .build();
 
         pullHerd.setPastureShepherd(pastureShepherd);

kafka/src/main/java/com/playtika/shepherd/KafkaPushFarm.java

@@ -18,7 +18,6 @@
 import java.util.Map;
 import java.util.function.Supplier;
 
-import static com.playtika.shepherd.inernal.CheckedHerd.checked;
 import static com.playtika.shepherd.inernal.utils.CacheUtils.memoize;
 import static com.playtika.shepherd.serde.SerDeUtils.BYTE_BUFFER_DE_SER;
 import static com.playtika.shepherd.serde.SerDeUtils.getSerDe;
@@ -56,7 +55,7 @@ private  <Breed> Pasture<Breed> addBreedingPasture(String herdName, SerDe<Breed>
                 .setGroupId(herdName)
                 .setProperties(properties)
                 .setRebalanceListener(pushHerd)
-                .setHerd(checked(pushHerd))
+                .setHerd(pushHerd)
                 .build();
 
         pushHerd.setPastureShepherd(pastureShepherd);

kafka/src/main/java/com/playtika/shepherd/inernal/CheckedHerd.java

@@ -1,37 +0,0 @@
-package com.playtika.shepherd.inernal;
-
-import org.apache.kafka.common.message.JoinGroupResponseData;
-
-import java.util.List;
-
-public class CheckedHerd implements Herd {
-
-    private final Herd herd;
-    private boolean requested = false;
-
-    public static Herd checked(Herd herd){
-        return new CheckedHerd(herd);
-    }
-
-    private CheckedHerd(Herd herd) {
-        this.herd = herd;
-    }
-
-    @Override
-    public Population getPopulation(List<JoinGroupResponseData.JoinGroupResponseMember> allMemberMetadata){
-        if(requested){
-            throw new IllegalStateException("Should be called only once on rebalance");
-        }
-        try {
-            return herd.getPopulation(allMemberMetadata);
-        } finally {
-            requested = true;
-        }
-    }
-
-    @Override
-    public void reset() {
-        herd.reset();
-        requested = false;
-    }
-}

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;
@@ -176,14 +177,39 @@ protected Map<String, ByteBuffer> onLeaderElected(String leaderId,
         if (skipAssignment)
             throw new IllegalStateException("Can't skip assignment because static membership is not supported.");
 
-        Population population = herd.getPopulation(allMemberMetadata);
-        leaderElected = leaderId;
+        try {
+            Population population = herd.getPopulation(allMemberMetadata);
+            leaderElected = leaderId;
+
+            if (logger.isDebugEnabled()) {
+                logger.debug("""
+                                Will assign population of size=[{}] among members count=[{}],
+                                members=[{}],
+                                population=[{}]""",
+                        population.getSheep().size(), allMemberMetadata.size(),
+                        collectMemberIds(allMemberMetadata),
+                        toBytes(population.getSheep()));
+            } else {
+                logger.info("""
+                                Will assign population of size=[{}] among members count=[{}],
+                                members=[{}]""",
+                        population.getSheep().size(), allMemberMetadata.size(),
+                        collectMemberIds(allMemberMetadata));
+            }
 
-        logger.info("Will rebalance population: [{}]", toBytes(population.getSheep()));
+            return assignor.performAssignment(leaderId, protocol,
+                    population.getSheep(), population.getVersion(),
+                    allMemberMetadata);
+        } catch (Throwable t) {
+            logger.error("Failed to assign population. Will return empty assignment.\n" +
+                            "members count=[{}], members=[{}]",
+                    allMemberMetadata.size(), collectMemberIds(allMemberMetadata));
+            return Map.of();
+        }
+    }
 
-        return assignor.performAssignment(leaderId, protocol,
-                population.getSheep(), population.getVersion(),
-                allMemberMetadata);
+    private static String collectMemberIds(List<JoinGroupResponseMember> allMemberMetadata) {
+        return allMemberMetadata.stream().map(JoinGroupResponseMember::memberId).collect(Collectors.joining(", "));
     }
 
     @Override