KAFKA-17367: Share coordinator impl. New merge batches algorithm. [3/N]

checkstyle/import-control-share-coordinator.xml

@@ -62,6 +62,8 @@
             <allow pkg="org.apache.kafka.server.util.timer" />
             <allow pkg="org.apache.kafka.timeline" />
             <allow pkg="org.junit.jupiter.api" />
+            <allow pkg="org.junit.jupiter.params" />
+            <allow pkg="org.junit.jupiter.provider" />
             <allow pkg="org.mockito" />
             <allow pkg="org.slf4j" />
             <subpackage name="generated">

checkstyle/suppressions.xml

@@ -340,6 +340,12 @@
     <suppress checks="NPathComplexity"
               files="CoordinatorRuntime.java"/>
 
+    <!-- share coordinator -->
+    <suppress checks="NPathComplexity"
+              files="ShareCoordinatorShard.java"/>
+    <suppress checks="CyclomaticComplexity"
+              files="ShareCoordinatorShard.java"/>
+
     <!-- storage -->
     <suppress checks="CyclomaticComplexity"
               files="(LogValidator|RemoteLogManagerConfig|RemoteLogManager).java"/>

server-common/src/main/java/org/apache/kafka/server/group/share/PersisterStateBatch.java

@@ -25,11 +25,11 @@
 /**
  * This class contains the information for a single batch of state information for use by the {@link Persister}.
  */
-public class PersisterStateBatch {
+public class PersisterStateBatch implements Comparable {
     private final long firstOffset;
     private final long lastOffset;
-    private final byte deliveryState;
     private final short deliveryCount;
+    private final byte deliveryState;
 
     public PersisterStateBatch(long firstOffset, long lastOffset, byte deliveryState, short deliveryCount) {
         this.firstOffset = firstOffset;
@@ -77,22 +77,40 @@ public boolean equals(Object o) {
         PersisterStateBatch that = (PersisterStateBatch) o;
         return firstOffset == that.firstOffset &&
             lastOffset == that.lastOffset &&
-            deliveryState == that.deliveryState &&
-            deliveryCount == that.deliveryCount;
+            deliveryCount == that.deliveryCount &&
+            deliveryState == that.deliveryState;
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(firstOffset, lastOffset, deliveryState, deliveryCount);
+        return Objects.hash(firstOffset, lastOffset, deliveryCount, deliveryState);
     }
 
     @Override
     public String toString() {
         return "PersisterStateBatch(" +
             "firstOffset=" + firstOffset + "," +
             "lastOffset=" + lastOffset + "," +
-            "deliveryState=" + deliveryState + "," +
-            "deliveryCount=" + deliveryCount +
+            "deliveryCount=" + deliveryCount + "," +
+            "deliveryState=" + deliveryState +
             ")";
     }
+
+    @Override
+    public int compareTo(Object o) {
+        PersisterStateBatch that = (PersisterStateBatch) o;
+        int deltaFirst = Long.compare(this.firstOffset(), that.firstOffset());
+        if (deltaFirst == 0) {
+            int deltaLast = Long.compare(this.lastOffset(), that.lastOffset());
+            if (deltaLast == 0) {
+                int deltaCount = this.deliveryCount() - that.deliveryCount();
+                if (deltaCount == 0) {
+                    return Byte.compare(this.deliveryState(), that.deliveryState());
+                }
+                return deltaCount;
+            }
+            return deltaLast;
+        }
+        return deltaFirst;
+    }
 }