IGNITE-24089 Introduce chain and link abstractions

modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceRaftGroupEventsListener.java

@@ -178,7 +178,7 @@ public void onNewPeersConfigurationApplied(PeersAndLearners configuration, long
                 try {
                     Set<Assignment> stable = createAssignments(configuration);
 
-                    doStableKeySwitch(stable, tablePartitionId, metaStorageMgr, calculateAssignmentsFn);
+                    doStableKeySwitch(stable, tablePartitionId, metaStorageMgr, term, index, calculateAssignmentsFn);
                 } finally {
                     busyLock.leaveBusy();
                 }
@@ -266,6 +266,8 @@ private static void doStableKeySwitch(
             Set<Assignment> stableFromRaft,
             TablePartitionId tablePartitionId,
             MetaStorageManager metaStorageMgr,
+            long configurationTerm,
+            long configurationIndex,
             BiFunction<TablePartitionId, Long, CompletableFuture<Set<Assignment>>> calculateAssignmentsFn
     ) {
         try {
@@ -356,7 +358,9 @@ private static void doStableKeySwitch(
                     assignmentsChainKey,
                     assignmentsChainEntry,
                     pendingAssignments,
-                    newStableAssignments
+                    newStableAssignments,
+                    configurationTerm,
+                    configurationIndex
             );
 
             Update successCase;
@@ -449,6 +453,8 @@ private static void doStableKeySwitch(
                         stableFromRaft,
                         tablePartitionId,
                         metaStorageMgr,
+                        configurationTerm,
+                        configurationIndex,
                         calculateAssignmentsFn
                 );
 
@@ -493,23 +499,31 @@ private static Operation handleAssignmentsChainChange(
             ByteArray assignmentsChainKey,
             Entry assignmentsChainEntry,
             Assignments pendingAssignments,
-            Assignments stableAssignments
+            Assignments stableAssignments,
+            long configurationTerm,
+            long configurationIndex
     ) {
         // We write this key only in HA mode. See TableManager.writeTableAssignmentsToMetastore.
         if (assignmentsChainEntry.value() != null) {
             AssignmentsChain updatedAssignmentsChain = updateAssignmentsChain(
                     AssignmentsChain.fromBytes(assignmentsChainEntry.value()),
                     stableAssignments,
-                    pendingAssignments
+                    pendingAssignments,
+                    configurationTerm,
+                    configurationIndex
             );
             return put(assignmentsChainKey, updatedAssignmentsChain.toBytes());
         } else {
             return Operations.noop();
         }
     }
 
-    private static AssignmentsChain updateAssignmentsChain(AssignmentsChain assignmentsChain, Assignments newStable,
-            Assignments pendingAssignments) {
+    private static AssignmentsChain updateAssignmentsChain(
+            AssignmentsChain assignmentsChain,
+            Assignments newStable,
+            Assignments pendingAssignments,
+            long configurationTerm,
+            long configurationIndex) {
         assert assignmentsChain != null : "Assignments chain cannot be null in HA mode.";
 
         assert !assignmentsChain.chain().isEmpty() : "Assignments chain cannot be empty on stable switch.";
@@ -539,11 +553,11 @@ private static AssignmentsChain updateAssignmentsChain(AssignmentsChain assignme
         */
         AssignmentsChain newAssignmentsChain;
         if (!pendingAssignments.force() && !pendingAssignments.fromReset()) {
-            newAssignmentsChain = AssignmentsChain.of(newStable);
+            newAssignmentsChain = AssignmentsChain.of(configurationTerm, configurationIndex, newStable);
         } else if (!pendingAssignments.force() && pendingAssignments.fromReset()) {
-            newAssignmentsChain = assignmentsChain.replaceLast(newStable);
+            newAssignmentsChain = assignmentsChain.replaceLast(newStable, configurationTerm, configurationIndex);
         } else {
-            newAssignmentsChain = assignmentsChain.addLast(newStable);
+            newAssignmentsChain = assignmentsChain.addLast(newStable, configurationTerm, configurationIndex);
         }
 
         return newAssignmentsChain;

modules/partition-distribution/src/main/java/org/apache/ignite/internal/partitiondistribution/AssignmentsChain.java

@@ -17,9 +17,12 @@
 
 package org.apache.ignite.internal.partitiondistribution;
 
+import static java.util.stream.Collectors.toList;
+
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
+import java.util.stream.Stream;
 import org.apache.ignite.internal.tostring.IgniteToStringInclude;
 import org.apache.ignite.internal.tostring.S;
 import org.apache.ignite.internal.versioned.VersionedSerialization;
@@ -30,15 +33,19 @@
  * Contains the chain of changed assignments.
  */
 public class AssignmentsChain {
+
+    private static final long DEFAULT_CONF_TERM = -1;
+    private static final long DEFAULT_CONF_IDX = -1;
+
     /** Chain of assignments. */
     @IgniteToStringInclude
-    private final List<Assignments> chain;
+    private final List<AssignmentsLink> chain;
 
-    private AssignmentsChain(List<Assignments> chain) {
+    private AssignmentsChain(List<AssignmentsLink> chain) {
         this.chain = chain;
     }
 
-    public List<Assignments> chain() {
+    public List<AssignmentsLink> chain() {
         return chain;
     }
 
@@ -48,12 +55,12 @@ public List<Assignments> chain() {
      * @param newLast New last link.
      * @return new AssignmentsChain.
      */
-    public AssignmentsChain replaceLast(Assignments newLast) {
+    public AssignmentsChain replaceLast(Assignments newLast, long configurationTerm, long configurationIndex) {
         assert !chain.isEmpty() : "Assignments chain is empty.";
 
-        List<Assignments> newChain = new ArrayList<>(chain);
+        List<AssignmentsLink> newChain = new ArrayList<>(chain);
 
-        newChain.set(newChain.size() - 1, newLast);
+        newChain.set(newChain.size() - 1, new AssignmentsLink(newLast, configurationTerm, configurationIndex));
 
         return new AssignmentsChain(newChain);
     }
@@ -64,31 +71,70 @@ public AssignmentsChain replaceLast(Assignments newLast) {
      * @param newLast New last link.
      * @return new AssignmentsChain.
      */
-    public AssignmentsChain addLast(Assignments newLast) {
+    public AssignmentsChain addLast(Assignments newLast, long configurationTerm, long configurationIndex) {
         assert !chain.isEmpty() : "Assignments chain is empty.";
 
-        List<Assignments> newChain = new ArrayList<>(chain);
+        List<AssignmentsLink> newChain = new ArrayList<>(chain);
 
-        newChain.add(newLast);
+        newChain.add(new AssignmentsLink(newLast, configurationTerm, configurationIndex));
 
         return new AssignmentsChain(newChain);
     }
 
+
+    /**
+     * Gets the next link in the chain after the given link.
+     *
+     * @param link The link to get the next one from.
+     * @return The next link in the chain, or {@code null} if the given link is the last one in the chain.
+     */
+    public @Nullable AssignmentsLink nextLink(AssignmentsLink link) {
+        int i = chain.indexOf(link);
+
+        return i < 0 || i == chain().size() - 1 ? null : chain.get(i + 1);
+    }
+
+    /**
+     * Returns the last {@link AssignmentsLink} in the chain that contains the specified node.
+     *
+     * @param nodeConsistentId The consistent identifier of the node to search for.
+     * @return The last {@link AssignmentsLink} that contains the node, or {@code null} if no such link exists.
+     */
+    public @Nullable AssignmentsLink lastLink(String nodeConsistentId) {
+        for (int i = chain.size() - 1; i >= 0; i--) {
+            AssignmentsLink link = chain.get(i);
+            if (link.hasNode(nodeConsistentId)) {
+                return link;
+            }
+        }
+
+        return null;
+    }
+
+    public static AssignmentsChain of(Assignments... assignments) {
+        return of(DEFAULT_CONF_TERM, DEFAULT_CONF_IDX, assignments);
+    }
+
     /**
      * Creates a new instance.
      *
      * @param assignments Partition assignments.
      */
-    public static AssignmentsChain of(Assignments assignments) {
-        return new AssignmentsChain(List.of(assignments));
+    public static AssignmentsChain of(long configurationTerm, long configurationIndex, Assignments... assignments) {
+        List<AssignmentsLink> links =
+                Stream.of(assignments)
+                        .map(assignment -> new AssignmentsLink(assignment, configurationTerm, configurationIndex))
+                        .collect(toList());
+
+        return new AssignmentsChain(links);
     }
 
     /**
      * Creates a new instance.
      *
      * @param assignmentsChain Chain of partition assignments.
      */
-    public static AssignmentsChain of(List<Assignments> assignmentsChain) {
+    public static AssignmentsChain of(List<AssignmentsLink> assignmentsChain) {
         return new AssignmentsChain(assignmentsChain);
     }
 

modules/partition-distribution/src/main/java/org/apache/ignite/internal/partitiondistribution/AssignmentsChainSerializer.java

@@ -37,19 +37,20 @@ public class AssignmentsChainSerializer extends VersionedSerializer<AssignmentsC
     protected void writeExternalData(AssignmentsChain chain, IgniteDataOutput out) throws IOException {
         out.writeVarInt(chain.chain().size());
 
-        for (Assignments assignment : chain.chain()) {
-            AssignmentsSerializer.INSTANCE.writeExternal(assignment, out);
+        for (AssignmentsLink assignment : chain.chain()) {
+            AssignmentsLinkSerializer.INSTANCE.writeExternal(assignment, out);
         }
     }
 
     @Override
     protected AssignmentsChain readExternalData(byte protoVer, IgniteDataInput in) throws IOException {
         int length = in.readVarIntAsInt();
-        List<Assignments> assignmentsChain = new ArrayList<>(length);
+        List<AssignmentsLink> links = new ArrayList<>(length);
 
         for (int i = 0; i < length; i++) {
-            assignmentsChain.add(AssignmentsSerializer.INSTANCE.readExternal(in));
+            links.add(AssignmentsLinkSerializer.INSTANCE.readExternal(in));
         }
-        return AssignmentsChain.of(assignmentsChain);
+
+        return AssignmentsChain.of(links);
     }
 }

modules/partition-distribution/src/main/java/org/apache/ignite/internal/partitiondistribution/AssignmentsLink.java

@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.partitiondistribution;
+
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.ignite.internal.tostring.S;
+
+/**
+ * Represents a link in the chain of assignments.
+ *
+ * <p>An AssignmentsLink instance encapsulates a set of node assignments along with the associated
+ * configuration term and index. This is used to keep track of changes in the node assignments for a partition over time.
+ */
+public class AssignmentsLink {
+    private final Assignments assignments;
+    private final long configurationIndex;
+    private final long configurationTerm;
+
+    AssignmentsLink(
+            Assignments assignments,
+            long configurationTerm,
+            long configurationIndex
+    ) {
+        this.assignments = assignments;
+        this.configurationIndex = configurationIndex;
+        this.configurationTerm = configurationTerm;
+    }
+
+    public Assignments assignments() {
+        return assignments;
+    }
+
+    /**
+     * Checks if the specified node is part of the current assignments.
+     *
+     * @param nodeConsistentId The consistent identifier of the node to check.
+     * @return {@code true} if the node is present in the assignments, otherwise {@code false}.
+     */
+
+    public boolean hasNode(String nodeConsistentId) {
+        return assignments.nodes().stream().map(Assignment::consistentId).anyMatch(nodeId -> nodeId.equals(nodeConsistentId));
+    }
+
+    /**
+     * Returns a set of consistent nodes present in the current assignments.
+     *
+     * @return Set of consistent node identifiers.
+     */
+    public Set<String> nodeNames() {
+        return assignments.nodes().stream().map(Assignment::consistentId).collect(Collectors.toSet());
+    }
+
+    public long configurationIndex() {
+        return configurationIndex;
+    }
+
+    public long configurationTerm() {
+        return configurationTerm;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+
+        AssignmentsLink link = (AssignmentsLink) o;
+        return configurationIndex == link.configurationIndex && configurationTerm == link.configurationTerm && Objects.equals(
+                assignments, link.assignments);
+    }
+
+    @Override
+    public int hashCode() {
+        int result = Objects.hashCode(assignments);
+        result = 31 * result + Long.hashCode(configurationIndex);
+        result = 31 * result + Long.hashCode(configurationTerm);
+        return result;
+    }
+
+    @Override
+    public String toString() {
+        return S.toString(this);
+    }
+}

modules/partition-distribution/src/main/java/org/apache/ignite/internal/partitiondistribution/AssignmentsLinkSerializer.java

@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.partitiondistribution;
+
+import java.io.IOException;
+import org.apache.ignite.internal.util.io.IgniteDataInput;
+import org.apache.ignite.internal.util.io.IgniteDataOutput;
+import org.apache.ignite.internal.versioned.VersionedSerializer;
+
+/**
+ * {@link VersionedSerializer} for {@link AssignmentsLink} instances.
+ */
+public class AssignmentsLinkSerializer extends VersionedSerializer<AssignmentsLink> {
+
+    /** Serializer instance. */
+    public static final AssignmentsLinkSerializer INSTANCE = new AssignmentsLinkSerializer();
+
+    @Override
+    protected void writeExternalData(AssignmentsLink link, IgniteDataOutput out) throws IOException {
+        AssignmentsSerializer.INSTANCE.writeExternal(link.assignments(), out);
+
+        out.writeVarInt(link.configurationTerm());
+        out.writeVarInt(link.configurationIndex());
+    }
+
+    @Override
+    protected AssignmentsLink readExternalData(byte protoVer, IgniteDataInput in) throws IOException {
+        Assignments assignments = AssignmentsSerializer.INSTANCE.readExternal(in);
+
+        long term = in.readVarInt();
+        long index = in.readVarInt();
+
+        return new AssignmentsLink(assignments, term, index);
+    }
+}