IGNITE-28013 Fix stale leaseholder ID handling in LeaseUpdater and lease batch serialization

modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/LeaseUpdater.java

@@ -486,7 +486,6 @@ private void updateLeaseBatchInternal() {
                         : lease.proposedCandidate();
 
                 InternalClusterNode candidate = nextLeaseHolder(stableAssignments, pendingAssignments, grpId, proposedLeaseholder);
-
                 boolean canBeProlonged = lease.isAccepted()
                         && lease.isProlongable()
                         && candidate != null && candidate.id().equals(lease.getLeaseholderId());

modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/leases/LeaseBatchSerializer.java

@@ -141,6 +141,15 @@ public class LeaseBatchSerializer extends VersionedSerializer<LeaseBatch> {
     /** Mask to extract lease holder index from compact representation. */
     private static final int COMPACT_HOLDER_INDEX_MASK = (1 << BIT_WIDTH_TO_FIT_IN_HALF_BYTE) - 1;
 
+    private static final byte PROTOCOL_V1 = 1;
+
+    private static final byte PROTOCOL_V2 = 2;
+
+    @Override
+    protected byte getProtocolVersion() {
+        return PROTOCOL_V2;
+    }
+
     @Override
     protected void writeExternalData(LeaseBatch batch, IgniteDataOutput out) throws IOException {
         long minExpirationTimePhysical = minExpirationTimePhysicalPart(batch);
@@ -356,7 +365,9 @@ private static int packNodesInfo(int holderNodeIndex, int proposedCandidateNameI
     private static boolean holderIdAndProposedCandidateFitIn1Byte(NodesDictionary dictionary) {
         // Up to 8 names means that for name index it's enough to have 3 bits, same for node index, so, in sum, they
         // require up to 6 bits, and we have 7 bits in a varint byte.
-        return dictionary.nameCount() <= MAX_NODES_FOR_COMPACT_MODE;
+        // We need to check both: name count (for proposed candidate index) and node count (for holder node index),
+        // as these can diverge when nodes restart with new UUIDs but the same name.
+        return dictionary.nameCount() <= MAX_NODES_FOR_COMPACT_MODE && dictionary.nodeCount() <= MAX_NODES_FOR_COMPACT_MODE;
     }
 
     private static int flags(
@@ -378,13 +389,15 @@ protected LeaseBatch readExternalData(byte protoVer, IgniteDataInput in) throws
         long minExpirationTimePhysical = in.readVarInt();
         HybridTimestamp commonExpirationTime = new HybridTimestamp(minExpirationTimePhysical + in.readVarInt(), in.readVarIntAsInt());
         NodesDictionary nodesDictionary = NodesDictionary.readFrom(in);
+        boolean canReadNodesInfoCompactly = holderIdAndProposedCandidateFitIn1ByteForRead(protoVer, nodesDictionary);
 
         List<Lease> leases = new ArrayList<>();
 
         readPartitionedGroupLeases(
                 minExpirationTimePhysical,
                 commonExpirationTime,
                 nodesDictionary,
+                canReadNodesInfoCompactly,
                 leases,
                 in,
                 TablePartitionId::new
@@ -395,6 +408,7 @@ protected LeaseBatch readExternalData(byte protoVer, IgniteDataInput in) throws
                     minExpirationTimePhysical,
                     commonExpirationTime,
                     nodesDictionary,
+                    canReadNodesInfoCompactly,
                     leases,
                     in,
                     ZonePartitionId::new
@@ -408,6 +422,7 @@ private static void readPartitionedGroupLeases(
             long minExpirationTimePhysical,
             HybridTimestamp commonExpirationTime,
             NodesDictionary nodesDictionary,
+            boolean canReadNodesInfoCompactly,
             List<Lease> leases,
             IgniteDataInput in,
             GroupIdFactory groupIdFactory
@@ -420,6 +435,7 @@ private static void readPartitionedGroupLeases(
                     minExpirationTimePhysical,
                     commonExpirationTime,
                     nodesDictionary,
+                    canReadNodesInfoCompactly,
                     leases,
                     in,
                     groupIdFactory,
@@ -432,6 +448,7 @@ private static int readLeasesForObject(
             long minExpirationTimePhysical,
             HybridTimestamp commonExpirationTime,
             NodesDictionary nodesDictionary,
+            boolean canReadNodesInfoCompactly,
             List<Lease> leases,
             IgniteDataInput in,
             GroupIdFactory groupIdFactory,
@@ -447,6 +464,7 @@ private static int readLeasesForObject(
                     minExpirationTimePhysical,
                     commonExpirationTime,
                     nodesDictionary,
+                    canReadNodesInfoCompactly,
                     in,
                     groupIdFactory
             );
@@ -464,6 +482,7 @@ private static int readLeasesForObject(
             long minExpirationTimePhysical,
             HybridTimestamp commonExpirationTime,
             NodesDictionary nodesDictionary,
+            boolean canReadNodesInfoCompactly,
             IgniteDataInput in,
             GroupIdFactory groupIdFactory
     ) throws IOException {
@@ -477,7 +496,7 @@ private static int readLeasesForObject(
 
         int holderNodeIndex;
         int proposedCandidateNodeIndex = -1;
-        if (holderIdAndProposedCandidateFitIn1Byte(nodesDictionary)) {
+        if (canReadNodesInfoCompactly) {
             int nodesInfo = in.readVarIntAsInt();
 
             holderNodeIndex = unpackHolderNodeIndex(nodesInfo);
@@ -538,6 +557,16 @@ private static boolean flagSet(int flags, int mask) {
         return (flags & mask) != 0;
     }
 
+    private static boolean holderIdAndProposedCandidateFitIn1ByteForRead(byte protoVer, NodesDictionary dictionary) {
+        if (protoVer == PROTOCOL_V1) {
+            // In V1 format, we assumed that name and node tables have the same size,
+            // so compact-mode eligibility was determined only by the name table size.
+            return dictionary.nameCount() <= MAX_NODES_FOR_COMPACT_MODE;
+        }
+
+        return holderIdAndProposedCandidateFitIn1Byte(dictionary);
+    }
+
     @FunctionalInterface
     private interface GroupIdFactory {
         PartitionGroupId create(int objectId, int partitionId);

modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/leases/NodesDictionary.java

@@ -132,6 +132,10 @@ int nameCount() {
         return nameIndexToName.size();
     }
 
+    int nodeCount() {
+        return nodeIndexToId.size();
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {