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,32 @@ 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.";
@@ -537,16 +552,15 @@ private static AssignmentsChain updateAssignmentsChain(AssignmentsChain assignme
               else
                 ms.chain = ms.chain + pending/stable // [A,B,C,D,E,F,G] => [A,B,C,D,E,F,G] -> [E]
         */
-        AssignmentsChain newAssignmentsChain;
         if (!pendingAssignments.force() && !pendingAssignments.fromReset()) {
-            newAssignmentsChain = AssignmentsChain.of(newStable);
+            return AssignmentsChain.of(configurationTerm, configurationIndex, newStable);
         } else if (!pendingAssignments.force() && pendingAssignments.fromReset()) {
-            newAssignmentsChain = assignmentsChain.replaceLast(newStable);
+            assignmentsChain.replaceLast(newStable, configurationTerm, configurationIndex);
         } else {
-            newAssignmentsChain = assignmentsChain.addLast(newStable);
+            assignmentsChain.addLast(newStable, configurationTerm, configurationIndex);
         }
 
-        return newAssignmentsChain;
+        return assignmentsChain;
     }
 
     /**

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

@@ -17,9 +17,11 @@
 
 package org.apache.ignite.internal.partitiondistribution;
 
-import java.util.ArrayList;
+import static java.util.stream.Collectors.toList;
+
 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,65 +32,114 @@
  * Contains the chain of changed assignments.
  */
 public class AssignmentsChain {
+    // TODO https://issues.apache.org/jira/browse/IGNITE-24177 Either remove default values or add proper javadoc.
+    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) {
-        this.chain = chain;
+    private AssignmentsChain(List<AssignmentsLink> chain) {
+        this.chain = chain.stream()
+                .map(assignmentsLink -> new AssignmentsLink(assignmentsLink, this::nextLink))
+                .collect(toList());
     }
 
-    public List<Assignments> chain() {
+    public List<AssignmentsLink> chain() {
         return chain;
     }
 
     /**
-     * Create a new {@link AssignmentsChain} with the last link in the chain replaced with the provided one.
+     * Replace the last link in the chain with the provided one.
      *
      * @param newLast New last link.
-     * @return new AssignmentsChain.
+     * @return the created {@link AssignmentsLink}
      */
-    public AssignmentsChain replaceLast(Assignments newLast) {
+    public AssignmentsLink replaceLast(Assignments newLast, long configurationTerm, long configurationIndex) {
         assert !chain.isEmpty() : "Assignments chain is empty.";
 
-        List<Assignments> newChain = new ArrayList<>(chain);
+        AssignmentsLink link = new AssignmentsLink(newLast, configurationTerm, configurationIndex, this::nextLink);
 
-        newChain.set(newChain.size() - 1, newLast);
+        chain.set(chain.size() - 1, link);
 
-        return new AssignmentsChain(newChain);
+        return link;
     }
 
     /**
-     * Create a new {@link AssignmentsChain} with a new link added to the chain.
+     * Add a new link to the end of the chain.
      *
      * @param newLast New last link.
-     * @return new AssignmentsChain.
+     * @return the created {@link AssignmentsLink}
      */
-    public AssignmentsChain addLast(Assignments newLast) {
+    public AssignmentsLink addLast(Assignments newLast, long configurationTerm, long configurationIndex) {
         assert !chain.isEmpty() : "Assignments chain is empty.";
 
-        List<Assignments> newChain = new ArrayList<>(chain);
+        AssignmentsLink link = new AssignmentsLink(newLast, configurationTerm, configurationIndex, this::nextLink);
+
+        chain.add(link);
+
+        return link;
+    }
+
+    /**
+     * 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.
+     */
+    private @Nullable AssignmentsLink nextLink(AssignmentsLink link) {
+        int i = chain.indexOf(link);
 
-        newChain.add(newLast);
+        return i < 0 || i == chain().size() - 1 ? null : chain.get(i + 1);
+    }
+
+    public AssignmentsLink firstLink() {
+        return chain.get(0);
+    }
+
+    /**
+     * Returns the last {@link AssignmentsLink} in the chain that contains the specified node. {@code}
+     * <pre>
+     *   on input link1([A,B,C,D,E,F,G]) > link2([E,F,G]) > link3([G])
+     *   chain.lastLink(F) should return link2(E,F,G).
+     * </pre>
+     *
+     * @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;
+    }
 
-        return new AssignmentsChain(newChain);
+    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) {
+        return of(Stream.of(assignments)
+                .map(assignment -> new AssignmentsLink(assignment, configurationTerm, configurationIndex, null))
+                .collect(toList()));
     }
 
     /**
      * Creates a new instance.
      *
      * @param assignmentsChain Chain of partition assignments.
      */
-    public static AssignmentsChain of(List<Assignments> assignmentsChain) {
+    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,126 @@
+/*
+ * 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 static java.util.stream.Collectors.toSet;
+
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.Function;
+import org.apache.ignite.internal.tostring.S;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * 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;
+    private final Function<AssignmentsLink, AssignmentsLink> nextLinkFunc;
+
+    AssignmentsLink(
+            Assignments assignments,
+            long configurationTerm,
+            long configurationIndex,
+            @Nullable Function<AssignmentsLink, AssignmentsLink> nextLinkFunc
+    ) {
+        this.assignments = assignments;
+        this.configurationIndex = configurationIndex;
+        this.configurationTerm = configurationTerm;
+        this.nextLinkFunc = nextLinkFunc;
+    }
+
+    AssignmentsLink(
+            AssignmentsLink other,
+            @Nullable Function<AssignmentsLink, AssignmentsLink> nextLinkFunc
+    ) {
+        this.assignments = other.assignments;
+        this.configurationIndex = other.configurationIndex;
+        this.configurationTerm = other.configurationTerm;
+        this.nextLinkFunc = nextLinkFunc;
+    }
+
+    public Assignments assignments() {
+        return assignments;
+    }
+
+    /**
+     * Gets the next link in the chain after the given link.
+     *
+     * @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() {
+        return nextLinkFunc == null ? null : nextLinkFunc.apply(this);
+    }
+
+    /**
+     * 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(consistentId -> consistentId.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(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);
+    }
+}