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,7 +25,7 @@
 /**
  * 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;
@@ -95,4 +95,22 @@ public String toString() {
             "deliveryCount=" + deliveryCount +
             ")";
     }
+
+    @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 countDelta = this.deliveryCount() - that.deliveryCount();
+                if (countDelta == 0) {
+                    return Byte.compare(this.deliveryState(), that.deliveryState());
+                }
+                return countDelta;
+            }
+            return deltaLast;
+        }
+        return deltaFirst;
+    }
 }