You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by mp...@apache.org on 2025/03/03 11:40:00 UTC
(ignite-3) branch main updated: IGNITE-24069 Turn the pending assignments into a queue (#5244)
This is an automated email from the ASF dual-hosted git repository.
mpochatkin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new b1a3da3228a IGNITE-24069 Turn the pending assignments into a queue (#5244)
b1a3da3228a is described below
commit b1a3da3228ac7a40ebe0f1bff10228c3dee661cd
Author: Vadim Kolodin <vk...@gridgain.com>
AuthorDate: Mon Mar 3 14:39:55 2025 +0300
IGNITE-24069 Turn the pending assignments into a queue (#5244)
---
check-rules/spotbugs-excludes.xml | 2 +-
.../ItDistributionZonesFiltersTest.java | 4 +-
.../rebalance/ItRebalanceDistributedTest.java | 21 ++--
.../rebalance/ItRebalanceTriggersRecoveryTest.java | 16 +--
.../RebalanceMinimumRequiredTimeProviderImpl.java | 22 +++-
.../RebalanceRaftGroupEventsListener.java | 15 ++-
.../distributionzones/rebalance/RebalanceUtil.java | 61 +++++----
.../ZoneRebalanceRaftGroupEventsListener.java | 23 ++--
.../rebalance/ZoneRebalanceUtil.java | 60 +++++----
.../DistributionZoneRebalanceEngineTest.java | 21 +++-
...balanceMinimumRequiredTimeProviderImplTest.java | 22 +++-
.../RebalanceUtilUpdateAssignmentsTest.java | 9 +-
.../ZoneRebalanceUtilUpdateAssignmentsTest.java | 9 +-
.../partitiondistribution/AssignmentsQueue.java | 140 +++++++++++++++++++++
.../AssignmentsQueueSerializer.java | 52 ++++++++
.../PendingAssignmentsCalculator.java | 63 ++++++++++
.../AssignmentsQueueSerializerTest.java | 93 ++++++++++++++
.../partition/replicator/fixtures/Node.java | 4 +-
.../PartitionReplicaLifecycleManager.java | 19 +--
.../placementdriver/AssignmentsTracker.java | 41 +++---
.../internal/placementdriver/LeaseUpdaterTest.java | 11 +-
.../placementdriver/PlacementDriverTest.java | 17 ++-
.../ignite/internal/replicator/ReplicaImpl.java | 4 +-
.../runner/app/ItIgniteNodeRestartTest.java | 4 +-
.../org/apache/ignite/internal/app/IgniteImpl.java | 4 +-
...bstractHighAvailablePartitionsRecoveryTest.java | 16 ++-
.../internal/table/distributed/TableManager.java | 22 ++--
.../distributed/disaster/GroupUpdateRequest.java | 16 ++-
.../ignite/internal/utils/RebalanceUtilEx.java | 7 +-
.../disaster/DisasterRecoveryMsInvokeTest.java | 9 +-
modules/table/tech-notes/rebalance.md | 10 +-
.../ItDisasterRecoveryReconfigurationTest.java | 7 +-
32 files changed, 643 insertions(+), 181 deletions(-)
diff --git a/check-rules/spotbugs-excludes.xml b/check-rules/spotbugs-excludes.xml
index 7c22a0c17d7..b5e5d527774 100644
--- a/check-rules/spotbugs-excludes.xml
+++ b/check-rules/spotbugs-excludes.xml
@@ -250,7 +250,7 @@
<Class name="org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil"/>
<Or>
<Field name="STABLE_ASSIGNMENTS_PREFIX_BYTES"/>
- <Field name="PENDING_ASSIGNMENTS_PREFIX_BYTES"/>
+ <Field name="PENDING_ASSIGNMENTS_QUEUE_PREFIX_BYTES"/>
<Field name="PENDING_CHANGE_TRIGGER_PREFIX_BYTES"/>
</Or>
</Match>
diff --git a/modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/distributionzones/ItDistributionZonesFiltersTest.java b/modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/distributionzones/ItDistributionZonesFiltersTest.java
index 03d48634881..45efcf85cce 100644
--- a/modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/distributionzones/ItDistributionZonesFiltersTest.java
+++ b/modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/distributionzones/ItDistributionZonesFiltersTest.java
@@ -26,7 +26,7 @@ import static org.apache.ignite.internal.catalog.commands.CatalogUtils.IMMEDIATE
import static org.apache.ignite.internal.distributionzones.DistributionZonesTestUtil.assertValueInStorage;
import static org.apache.ignite.internal.distributionzones.DistributionZonesTestUtil.deserializeLatestDataNodesHistoryEntry;
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zoneDataNodesHistoryKey;
-import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsQueueKey;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
import static org.apache.ignite.internal.util.ByteUtils.toBytes;
@@ -465,7 +465,7 @@ public class ItDistributionZonesFiltersTest extends ClusterPerTestIntegrationTes
MetaStorageManager metaStorageManager,
TablePartitionId partId
) throws InterruptedException, ExecutionException {
- assertTrue(metaStorageManager.get(pendingPartAssignmentsKey(partId)).get().empty());
+ assertTrue(metaStorageManager.get(pendingPartAssignmentsQueueKey(partId)).get().empty());
}
private static String createZoneSql(
diff --git a/modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/rebalance/ItRebalanceDistributedTest.java b/modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/rebalance/ItRebalanceDistributedTest.java
index eb1e8f12f85..288add55c07 100644
--- a/modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/rebalance/ItRebalanceDistributedTest.java
+++ b/modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/rebalance/ItRebalanceDistributedTest.java
@@ -32,7 +32,7 @@ import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.REBALANCE_SCHEDULER_POOL_SIZE;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.STABLE_ASSIGNMENTS_PREFIX_BYTES;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.extractTablePartitionId;
-import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsQueueKey;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.plannedPartAssignmentsKey;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartitionAssignments;
import static org.apache.ignite.internal.partitiondistribution.PartitionDistributionUtils.calculateAssignmentForPartition;
@@ -180,6 +180,7 @@ import org.apache.ignite.internal.partition.replicator.network.PartitionReplicat
import org.apache.ignite.internal.partition.replicator.raft.snapshot.outgoing.OutgoingSnapshotsManager;
import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.partitiondistribution.Assignments;
+import org.apache.ignite.internal.partitiondistribution.AssignmentsQueue;
import org.apache.ignite.internal.placementdriver.TestPlacementDriver;
import org.apache.ignite.internal.placementdriver.TestReplicaMetaImpl;
import org.apache.ignite.internal.raft.JraftGroupEventsListener;
@@ -760,12 +761,12 @@ public class ItRebalanceDistributedTest extends BaseIgniteAbstractTest {
{
TablePartitionId partId = new TablePartitionId(getTableId(node, TABLE_NAME), 0);
- ByteArray partAssignmentsPendingKey = pendingPartAssignmentsKey(partId);
+ ByteArray partAssignmentsPendingKey = pendingPartAssignmentsQueueKey(partId);
int catalogVersion = node.catalogManager.latestCatalogVersion();
long timestamp = node.catalogManager.catalog(catalogVersion).time();
- byte[] bytesPendingAssignments = Assignments.toBytes(newAssignment, timestamp);
+ byte[] bytesPendingAssignments = AssignmentsQueue.toBytes(Assignments.of(newAssignment, timestamp));
node.metaStorageManager
.put(partAssignmentsPendingKey, bytesPendingAssignments)
@@ -833,12 +834,12 @@ public class ItRebalanceDistributedTest extends BaseIgniteAbstractTest {
// Write the new assignments to metastore as a pending assignments.
TablePartitionId partId = new TablePartitionId(getTableId(node, TABLE_NAME), 0);
- ByteArray partAssignmentsPendingKey = pendingPartAssignmentsKey(partId);
+ ByteArray partAssignmentsPendingKey = pendingPartAssignmentsQueueKey(partId);
int catalogVersion = node.catalogManager.latestCatalogVersion();
long timestamp = node.catalogManager.catalog(catalogVersion).time();
- byte[] bytesPendingAssignments = Assignments.toBytes(newAssignment, timestamp);
+ byte[] bytesPendingAssignments = AssignmentsQueue.toBytes(Assignments.of(newAssignment, timestamp));
AtomicBoolean dropMessages = new AtomicBoolean(true);
@@ -953,12 +954,12 @@ public class ItRebalanceDistributedTest extends BaseIgniteAbstractTest {
int catalogVersion = node0.catalogManager.latestCatalogVersion();
long timestamp = node0.catalogManager.catalog(catalogVersion).time();
- byte[] bytesPendingAssignments = Assignments.toBytes(pendingAssignments, timestamp);
+ byte[] bytesPendingAssignments = AssignmentsQueue.toBytes(Assignments.of(pendingAssignments, timestamp));
byte[] bytesPlannedAssignments = Assignments.toBytes(plannedAssignments, timestamp);
TablePartitionId partId = new TablePartitionId(getTableId(node0, TABLE_NAME), 0);
- ByteArray partAssignmentsPendingKey = pendingPartAssignmentsKey(partId);
+ ByteArray partAssignmentsPendingKey = pendingPartAssignmentsQueueKey(partId);
ByteArray partAssignmentsPlannedKey = plannedPartAssignmentsKey(partId);
Map<ByteArray, byte[]> msEntries = new HashMap<>();
@@ -1079,8 +1080,8 @@ public class ItRebalanceDistributedTest extends BaseIgniteAbstractTest {
int partitionNumber
) {
return metaStorageManager
- .get(pendingPartAssignmentsKey(new TablePartitionId(tableId, partitionNumber)))
- .thenApply(e -> (e.value() == null) ? null : Assignments.fromBytes(e.value()).nodes());
+ .get(pendingPartAssignmentsQueueKey(new TablePartitionId(tableId, partitionNumber)))
+ .thenApply(e -> (e.value() == null) ? null : AssignmentsQueue.fromBytes(e.value()).poll().nodes());
}
private static CompletableFuture<Set<Assignment>> partitionPlannedAssignments(
@@ -1439,7 +1440,7 @@ public class ItRebalanceDistributedTest extends BaseIgniteAbstractTest {
partitionRaftConfigurer,
view -> new LocalLogStorageFactory(),
ForkJoinPool.commonPool(),
- replicaGrpId -> metaStorageManager.get(pendingPartAssignmentsKey((TablePartitionId) replicaGrpId))
+ replicaGrpId -> metaStorageManager.get(pendingPartAssignmentsQueueKey((TablePartitionId) replicaGrpId))
.thenApply(Entry::value)
));
diff --git a/modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/rebalance/ItRebalanceTriggersRecoveryTest.java b/modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/rebalance/ItRebalanceTriggersRecoveryTest.java
index 53441236cef..09c69c8390c 100644
--- a/modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/rebalance/ItRebalanceTriggersRecoveryTest.java
+++ b/modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/rebalance/ItRebalanceTriggersRecoveryTest.java
@@ -20,7 +20,7 @@ package org.apache.ignite.internal.rebalance;
import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
import static org.apache.ignite.internal.TestWrappers.unwrapTableManager;
import static org.apache.ignite.internal.catalog.CatalogService.DEFAULT_STORAGE_PROFILE;
-import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsQueueKey;
import static org.apache.ignite.internal.table.TableTestUtils.getTableId;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.bypassingThreadAssertions;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
@@ -38,7 +38,7 @@ import org.apache.ignite.internal.app.IgniteImpl;
import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.partitiondistribution.Assignment;
-import org.apache.ignite.internal.partitiondistribution.Assignments;
+import org.apache.ignite.internal.partitiondistribution.AssignmentsQueue;
import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.storage.MvPartitionStorage;
import org.apache.ignite.internal.storage.RowId;
@@ -126,7 +126,7 @@ public class ItRebalanceTriggersRecoveryTest extends ClusterPerTestIntegrationTe
Integer tableId = getTableId(unwrapIgniteImpl(node(0)).catalogManager(), "TEST", new HybridClockImpl().nowLong());
unwrapIgniteImpl(node(0))
.metaStorageManager()
- .remove(pendingPartAssignmentsKey(new TablePartitionId(tableId, PARTITION_ID))).join();
+ .remove(pendingPartAssignmentsQueueKey(new TablePartitionId(tableId, PARTITION_ID))).join();
restartNode(1);
restartNode(2);
@@ -172,7 +172,7 @@ public class ItRebalanceTriggersRecoveryTest extends ClusterPerTestIntegrationTe
Integer tableId = getTableId(unwrapIgniteImpl(node(0)).catalogManager(), "TEST", new HybridClockImpl().nowLong());
unwrapIgniteImpl(node(0))
.metaStorageManager()
- .remove(pendingPartAssignmentsKey(new TablePartitionId(tableId, PARTITION_ID))).join();
+ .remove(pendingPartAssignmentsQueueKey(new TablePartitionId(tableId, PARTITION_ID))).join();
restartNode(1);
restartNode(2);
@@ -219,14 +219,14 @@ public class ItRebalanceTriggersRecoveryTest extends ClusterPerTestIntegrationTe
PARTITION_ID
);
long pendingsKeysRevisionBeforeRecovery = unwrapIgniteImpl(node(0)).metaStorageManager()
- .get(pendingPartAssignmentsKey(tablePartitionId))
+ .get(pendingPartAssignmentsQueueKey(tablePartitionId))
.get(10, TimeUnit.SECONDS).revision();
startNode(3, GLOBAL_NODE_BOOTSTRAP_CFG_TEMPLATE);
long pendingsKeysRevisionAfterRecovery = unwrapIgniteImpl(node(0)).metaStorageManager()
- .get(pendingPartAssignmentsKey(tablePartitionId))
+ .get(pendingPartAssignmentsQueueKey(tablePartitionId))
.get(10, TimeUnit.SECONDS).revision();
// Check that recovered node doesn't produce new rebalances for already processed triggers.
@@ -245,8 +245,8 @@ public class ItRebalanceTriggersRecoveryTest extends ClusterPerTestIntegrationTe
int partitionNumber
) {
return metaStorageManager
- .get(pendingPartAssignmentsKey(new TablePartitionId(tableId, partitionNumber)))
- .thenApply(e -> (e.value() == null) ? null : Assignments.fromBytes(e.value()).nodes());
+ .get(pendingPartAssignmentsQueueKey(new TablePartitionId(tableId, partitionNumber)))
+ .thenApply(e -> (e.value() == null) ? null : AssignmentsQueue.fromBytes(e.value()).poll().nodes());
}
private static boolean containsPartition(Ignite node) {
diff --git a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceMinimumRequiredTimeProviderImpl.java b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceMinimumRequiredTimeProviderImpl.java
index 7cfb58931c1..724a9c42eff 100644
--- a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceMinimumRequiredTimeProviderImpl.java
+++ b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceMinimumRequiredTimeProviderImpl.java
@@ -21,7 +21,7 @@ import static java.lang.Math.min;
import static java.util.Collections.emptyMap;
import static java.util.stream.Collectors.toMap;
import static org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor.updateRequiresAssignmentsRecalculation;
-import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.PENDING_ASSIGNMENTS_PREFIX_BYTES;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.PENDING_ASSIGNMENTS_QUEUE_PREFIX_BYTES;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.PENDING_CHANGE_TRIGGER_PREFIX_BYTES;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.STABLE_ASSIGNMENTS_PREFIX_BYTES;
@@ -32,6 +32,7 @@ import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;
import java.util.TreeMap;
+import java.util.function.Function;
import org.apache.ignite.internal.catalog.Catalog;
import org.apache.ignite.internal.catalog.CatalogService;
import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
@@ -40,6 +41,7 @@ import org.apache.ignite.internal.lang.ByteArray;
import org.apache.ignite.internal.metastorage.Entry;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.partitiondistribution.Assignments;
+import org.apache.ignite.internal.partitiondistribution.AssignmentsQueue;
import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.util.ByteUtils;
import org.apache.ignite.internal.util.Cursor;
@@ -72,7 +74,7 @@ public class RebalanceMinimumRequiredTimeProviderImpl implements RebalanceMinimu
long minTimestamp = metaStorageSafeTime;
Map<Integer, Map<Integer, Assignments>> stableAssignments = readAssignments(STABLE_ASSIGNMENTS_PREFIX_BYTES, appliedRevision);
- Map<Integer, Map<Integer, Assignments>> pendingAssignments = readAssignments(PENDING_ASSIGNMENTS_PREFIX_BYTES, appliedRevision);
+ Map<Integer, Map<Integer, Assignments>> pendingAssignments = readPendingAssignments(appliedRevision);
Map<Integer, Long> pendingChangeTriggerRevisions = readPendingChangeTriggerRevisions(
PENDING_CHANGE_TRIGGER_PREFIX_BYTES,
@@ -250,10 +252,22 @@ public class RebalanceMinimumRequiredTimeProviderImpl implements RebalanceMinimu
return tableIdToZoneIdMap;
}
+ private Map<Integer, Map<Integer, Assignments>> readPendingAssignments(long appliedRevision) {
+ return readAssignments(PENDING_ASSIGNMENTS_QUEUE_PREFIX_BYTES, appliedRevision, bytes -> AssignmentsQueue.fromBytes(bytes).poll());
+ }
+
+ private Map<Integer, Map<Integer, Assignments>> readAssignments(byte[] prefix, long appliedRevision) {
+ return readAssignments(prefix, appliedRevision, Assignments::fromBytes);
+ }
+
/**
* Reads assignments from the metastorage locally. The resulting map is a {@code tableId -> {partitionId -> assignments}} mapping.
*/
- Map<Integer, Map<Integer, Assignments>> readAssignments(byte[] prefix, long appliedRevision) {
+ Map<Integer, Map<Integer, Assignments>> readAssignments(
+ byte[] prefix,
+ long appliedRevision,
+ Function<byte[], Assignments> deserializer
+ ) {
Map<Integer, Map<Integer, Assignments>> assignments = new HashMap<>();
try (Cursor<Entry> entries = readLocallyByPrefix(prefix, appliedRevision)) {
@@ -267,7 +281,7 @@ public class RebalanceMinimumRequiredTimeProviderImpl implements RebalanceMinimu
int partitionId = tablePartitionId.partitionId();
assignments.computeIfAbsent(tableId, id -> new HashMap<>())
- .put(partitionId, Assignments.fromBytes(entry.value()));
+ .put(partitionId, deserializer.apply(entry.value()));
}
}
diff --git a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceRaftGroupEventsListener.java b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceRaftGroupEventsListener.java
index 5440841bf3a..c5fb1dded2d 100644
--- a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceRaftGroupEventsListener.java
+++ b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceRaftGroupEventsListener.java
@@ -18,7 +18,7 @@
package org.apache.ignite.internal.distributionzones.rebalance;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.assignmentsChainKey;
-import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsQueueKey;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.plannedPartAssignmentsKey;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.switchAppendKey;
@@ -58,6 +58,7 @@ import org.apache.ignite.internal.metastorage.dsl.Update;
import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.partitiondistribution.Assignments;
import org.apache.ignite.internal.partitiondistribution.AssignmentsChain;
+import org.apache.ignite.internal.partitiondistribution.AssignmentsQueue;
import org.apache.ignite.internal.raft.PeersAndLearners;
import org.apache.ignite.internal.raft.RaftError;
import org.apache.ignite.internal.raft.RaftGroupEventsListener;
@@ -271,7 +272,7 @@ public class RebalanceRaftGroupEventsListener implements RaftGroupEventsListener
BiFunction<TablePartitionId, Long, CompletableFuture<Set<Assignment>>> calculateAssignmentsFn
) {
try {
- ByteArray pendingPartAssignmentsKey = pendingPartAssignmentsKey(tablePartitionId);
+ ByteArray pendingPartAssignmentsKey = pendingPartAssignmentsQueueKey(tablePartitionId);
ByteArray stablePartAssignmentsKey = stablePartAssignmentsKey(tablePartitionId);
ByteArray plannedPartAssignmentsKey = plannedPartAssignmentsKey(tablePartitionId);
ByteArray switchReduceKey = switchReduceKey(tablePartitionId);
@@ -301,7 +302,9 @@ public class RebalanceRaftGroupEventsListener implements RaftGroupEventsListener
Set<Assignment> retrievedSwitchReduce = readAssignments(switchReduceEntry).nodes();
Set<Assignment> retrievedSwitchAppend = readAssignments(switchAppendEntry).nodes();
- Assignments pendingAssignments = readAssignments(pendingEntry);
+ Assignments pendingAssignments = pendingEntry.value() == null
+ ? Assignments.EMPTY
+ : AssignmentsQueue.fromBytes(pendingEntry.value()).poll();
Set<Assignment> retrievedPending = pendingAssignments.nodes();
if (!retrievedPending.equals(stableFromRaft)) {
@@ -367,8 +370,8 @@ public class RebalanceRaftGroupEventsListener implements RaftGroupEventsListener
Update failCase;
byte[] stableFromRaftByteArray = newStableAssignments.toBytes();
- byte[] additionByteArray = Assignments.toBytes(calculatedPendingAddition, catalogTimestamp);
- byte[] reductionByteArray = Assignments.toBytes(calculatedPendingReduction, catalogTimestamp);
+ byte[] additionByteArray = AssignmentsQueue.toBytes(Assignments.of(calculatedPendingAddition, catalogTimestamp));
+ byte[] reductionByteArray = AssignmentsQueue.toBytes(Assignments.of(calculatedPendingReduction, catalogTimestamp));
byte[] switchReduceByteArray = Assignments.toBytes(calculatedSwitchReduce, catalogTimestamp);
byte[] switchAppendByteArray = Assignments.toBytes(calculatedSwitchAppend, catalogTimestamp);
@@ -398,7 +401,7 @@ public class RebalanceRaftGroupEventsListener implements RaftGroupEventsListener
successCase = ops(
put(stablePartAssignmentsKey, stableFromRaftByteArray),
- put(pendingPartAssignmentsKey, plannedEntry.value()),
+ put(pendingPartAssignmentsKey, AssignmentsQueue.toBytes(Assignments.fromBytes(plannedEntry.value()))),
remove(plannedPartAssignmentsKey),
assignmentChainChangeOp
).yield(SCHEDULE_PENDING_REBALANCE_SUCCESS);
diff --git a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceUtil.java b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceUtil.java
index 9c81eb1e650..0e2b0e8f27e 100644
--- a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceUtil.java
+++ b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceUtil.java
@@ -39,6 +39,7 @@ import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
import static org.apache.ignite.internal.metastorage.dsl.Operations.remove;
import static org.apache.ignite.internal.metastorage.dsl.Statements.iif;
import static org.apache.ignite.internal.partitiondistribution.PartitionDistributionUtils.calculateAssignmentForPartition;
+import static org.apache.ignite.internal.partitiondistribution.PendingAssignmentsCalculator.pendingAssignmentsCalculator;
import static org.apache.ignite.internal.util.ByteUtils.longToBytesKeepingOrder;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.StringUtils.toStringWithoutPrefix;
@@ -64,6 +65,7 @@ import org.apache.ignite.internal.metastorage.dsl.Iif;
import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.partitiondistribution.Assignments;
import org.apache.ignite.internal.partitiondistribution.AssignmentsChain;
+import org.apache.ignite.internal.partitiondistribution.AssignmentsQueue;
import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.util.ExceptionUtils;
import org.jetbrains.annotations.Nullable;
@@ -158,7 +160,7 @@ public class RebalanceUtil {
) {
ByteArray partChangeTriggerKey = pendingChangeTriggerKey(partId);
- ByteArray partAssignmentsPendingKey = pendingPartAssignmentsKey(partId);
+ ByteArray partAssignmentsPendingKey = pendingPartAssignmentsQueueKey(partId);
ByteArray partAssignmentsPlannedKey = plannedPartAssignmentsKey(partId);
@@ -166,7 +168,7 @@ public class RebalanceUtil {
Set<Assignment> calculatedAssignments = calculateAssignmentForPartition(dataNodes, partNum, partitions, replicas);
- Set<Assignment> partAssignments;
+ Set<Assignment> targetAssignmentSet;
if (consistencyMode == ConsistencyMode.HIGH_AVAILABILITY) {
// All complicated logic here is needed because we want to return back to stable nodes
@@ -194,26 +196,34 @@ public class RebalanceUtil {
}
}
- partAssignments = resultingAssignments;
+ targetAssignmentSet = resultingAssignments;
} else {
- partAssignments = calculatedAssignments;
+ targetAssignmentSet = calculatedAssignments;
}
- boolean isNewAssignments = !tableCfgPartAssignments.equals(partAssignments);
+ boolean isNewAssignments = !tableCfgPartAssignments.equals(targetAssignmentSet);
+
+ Assignments targetAssignments = Assignments.of(targetAssignmentSet, assignmentsTimestamp);
+ AssignmentsQueue partAssignmentsPendingQueue = pendingAssignmentsCalculator()
+ .stable(Assignments.of(tableCfgPartAssignments, assignmentsTimestamp))
+ .target(targetAssignments)
+ .toQueue();
+
+ byte[] partAssignmentsPlannedBytes = targetAssignments.toBytes();
+ byte[] partAssignmentsPendingQueueBytes = partAssignmentsPendingQueue.toBytes();
- byte[] partAssignmentsBytes = Assignments.toBytes(partAssignments, assignmentsTimestamp);
// if empty(partition.change.trigger.revision) || partition.change.trigger.revision < event.revision:
// if empty(partition.assignments.pending)
// && ((isNewAssignments && empty(partition.assignments.stable))
// || (partition.assignments.stable != calcPartAssignments() && !empty(partition.assignments.stable))):
- // partition.assignments.pending = calcPartAssignments()
+ // partition.assignments.pending = partAssignmentsPendingQueue
// partition.change.trigger.revision = event.revision
// else:
- // if partition.assignments.pending != calcPartAssignments && !empty(partition.assignments.pending)
+ // if partition.assignments.pending != partAssignmentsPendingQueue && !empty(partition.assignments.pending)
// partition.assignments.planned = calcPartAssignments()
// partition.change.trigger.revision = event.revision
- // else if partition.assignments.pending == calcPartAssignments
+ // else if partition.assignments.pending == partAssignmentsPendingQueue
// remove(partition.assignments.planned)
// partition.change.trigger.revision = event.revision
// message after the metastorage invoke:
@@ -226,7 +236,8 @@ public class RebalanceUtil {
// else:
// skip
- Condition newAssignmentsCondition = exists(partAssignmentsStableKey).and(value(partAssignmentsStableKey).ne(partAssignmentsBytes));
+ Condition newAssignmentsCondition = exists(partAssignmentsStableKey)
+ .and(value(partAssignmentsStableKey).ne(partAssignmentsPlannedBytes));
if (isNewAssignments) {
newAssignmentsCondition = notExists(partAssignmentsStableKey).or(newAssignmentsCondition);
@@ -237,15 +248,15 @@ public class RebalanceUtil {
Iif iif = iif(or(notExists(partChangeTriggerKey), value(partChangeTriggerKey).lt(revisionBytes)),
iif(and(notExists(partAssignmentsPendingKey), newAssignmentsCondition),
ops(
- put(partAssignmentsPendingKey, partAssignmentsBytes),
+ put(partAssignmentsPendingKey, partAssignmentsPendingQueueBytes),
put(partChangeTriggerKey, revisionBytes)
).yield(PENDING_KEY_UPDATED.ordinal()),
- iif(and(value(partAssignmentsPendingKey).ne(partAssignmentsBytes), exists(partAssignmentsPendingKey)),
+ iif(and(value(partAssignmentsPendingKey).ne(partAssignmentsPendingQueueBytes), exists(partAssignmentsPendingKey)),
ops(
- put(partAssignmentsPlannedKey, partAssignmentsBytes),
+ put(partAssignmentsPlannedKey, partAssignmentsPlannedBytes),
put(partChangeTriggerKey, revisionBytes)
).yield(PLANNED_KEY_UPDATED.ordinal()),
- iif(value(partAssignmentsPendingKey).eq(partAssignmentsBytes),
+ iif(value(partAssignmentsPendingKey).eq(partAssignmentsPendingQueueBytes),
ops(
remove(partAssignmentsPlannedKey),
put(partChangeTriggerKey, revisionBytes)
@@ -265,14 +276,14 @@ public class RebalanceUtil {
LOG.info(
"Update metastore pending partitions key [key={}, partition={}, table={}/{}, newVal={}]",
partAssignmentsPendingKey.toString(), partNum, tableDescriptor.id(), tableDescriptor.name(),
- partAssignments);
+ partAssignmentsPendingQueue);
break;
case PLANNED_KEY_UPDATED:
LOG.info(
"Update metastore planned partitions key [key={}, partition={}, table={}/{}, newVal={}]",
partAssignmentsPlannedKey, partNum, tableDescriptor.id(), tableDescriptor.name(),
- partAssignments
+ targetAssignmentSet
);
break;
@@ -280,7 +291,7 @@ public class RebalanceUtil {
LOG.info(
"Remove planned key because current pending key has the same value [key={}, partition={}, table={}/{}, val={}]",
partAssignmentsPlannedKey.toString(), partNum, tableDescriptor.id(), tableDescriptor.name(),
- partAssignments
+ targetAssignmentSet
);
break;
@@ -289,7 +300,7 @@ public class RebalanceUtil {
"Remove planned key because pending is empty and calculated assignments are equal to current assignments "
+ "[key={}, partition={}, table={}/{}, val={}]",
partAssignmentsPlannedKey.toString(), partNum, tableDescriptor.id(), tableDescriptor.name(),
- partAssignments
+ targetAssignmentSet
);
break;
@@ -297,7 +308,7 @@ public class RebalanceUtil {
LOG.debug(
"Assignments are not updated [key={}, partition={}, table={}/{}, val={}]",
partAssignmentsPlannedKey.toString(), partNum, tableDescriptor.id(), tableDescriptor.name(),
- partAssignments
+ targetAssignmentSet
);
break;
@@ -429,9 +440,9 @@ public class RebalanceUtil {
public static final String PLANNED_ASSIGNMENTS_PREFIX = "assignments.planned.";
/** Key prefix for pending assignments. */
- public static final String PENDING_ASSIGNMENTS_PREFIX = "assignments.pending.";
+ public static final String PENDING_ASSIGNMENTS_QUEUE_PREFIX = "assignments.pending.";
- public static final byte[] PENDING_ASSIGNMENTS_PREFIX_BYTES = "assignments.pending.".getBytes(UTF_8);
+ public static final byte[] PENDING_ASSIGNMENTS_QUEUE_PREFIX_BYTES = "assignments.pending.".getBytes(UTF_8);
/** Key prefix for stable assignments. */
public static final String STABLE_ASSIGNMENTS_PREFIX = "assignments.stable.";
@@ -473,8 +484,8 @@ public class RebalanceUtil {
* @return Key for a partition.
* @see <a href="https://github.com/apache/ignite-3/blob/main/modules/table/tech-notes/rebalance.md">Rebalance documentation</a>
*/
- public static ByteArray pendingPartAssignmentsKey(TablePartitionId partId) {
- return new ByteArray(PENDING_ASSIGNMENTS_PREFIX + partId);
+ public static ByteArray pendingPartAssignmentsQueueKey(TablePartitionId partId) {
+ return new ByteArray(PENDING_ASSIGNMENTS_QUEUE_PREFIX + partId);
}
/**
@@ -732,9 +743,9 @@ public class RebalanceUtil {
) {
return IntStream.range(0, numberOfPartitions)
.mapToObj(p -> {
- Entry e = metaStorageManager.getLocally(pendingPartAssignmentsKey(new TablePartitionId(tableId, p)), revision);
+ Entry e = metaStorageManager.getLocally(pendingPartAssignmentsQueueKey(new TablePartitionId(tableId, p)), revision);
- return e != null ? Assignments.fromBytes(e.value()) : null;
+ return e != null && e.value() != null ? AssignmentsQueue.fromBytes(e.value()).poll() : null;
})
.collect(toList());
}
diff --git a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/ZoneRebalanceRaftGroupEventsListener.java b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/ZoneRebalanceRaftGroupEventsListener.java
index 03b353622b9..f7ea964cc51 100644
--- a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/ZoneRebalanceRaftGroupEventsListener.java
+++ b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/ZoneRebalanceRaftGroupEventsListener.java
@@ -17,7 +17,7 @@
package org.apache.ignite.internal.distributionzones.rebalance;
-import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.pendingPartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.pendingPartAssignmentsQueueKey;
import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.plannedPartAssignmentsKey;
import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.stablePartAssignmentsKey;
import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.switchAppendKey;
@@ -62,6 +62,7 @@ import org.apache.ignite.internal.metastorage.dsl.SimpleCondition;
import org.apache.ignite.internal.metastorage.dsl.Update;
import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.partitiondistribution.Assignments;
+import org.apache.ignite.internal.partitiondistribution.AssignmentsQueue;
import org.apache.ignite.internal.raft.PeersAndLearners;
import org.apache.ignite.internal.raft.RaftError;
import org.apache.ignite.internal.raft.RaftGroupEventsListener;
@@ -175,10 +176,10 @@ public class ZoneRebalanceRaftGroupEventsListener implements RaftGroupEventsList
try {
rebalanceAttempts.set(0);
- byte[] pendingAssignmentsBytes = metaStorageMgr.get(pendingPartAssignmentsKey(zonePartitionId)).get().value();
+ byte[] pendingAssignmentsBytes = metaStorageMgr.get(pendingPartAssignmentsQueueKey(zonePartitionId)).get().value();
if (pendingAssignmentsBytes != null) {
- Set<Assignment> pendingAssignments = Assignments.fromBytes(pendingAssignmentsBytes).nodes();
+ Set<Assignment> pendingAssignments = AssignmentsQueue.fromBytes(pendingAssignmentsBytes).poll().nodes();
var peers = new HashSet<String>();
var learners = new HashSet<String>();
@@ -324,7 +325,7 @@ public class ZoneRebalanceRaftGroupEventsListener implements RaftGroupEventsList
BiFunction<ZonePartitionId, Long, CompletableFuture<Set<Assignment>>> calculateAssignmentsFn
) {
try {
- ByteArray pendingPartAssignmentsKey = pendingPartAssignmentsKey(zonePartitionId);
+ ByteArray pendingPartAssignmentsKey = pendingPartAssignmentsQueueKey(zonePartitionId);
ByteArray stablePartAssignmentsKey = stablePartAssignmentsKey(zonePartitionId);
ByteArray plannedPartAssignmentsKey = plannedPartAssignmentsKey(zonePartitionId);
ByteArray switchReduceKey = switchReduceKey(zonePartitionId);
@@ -349,7 +350,9 @@ public class ZoneRebalanceRaftGroupEventsListener implements RaftGroupEventsList
Set<Assignment> retrievedStable = readAssignments(stableEntry).nodes();
Set<Assignment> retrievedSwitchReduce = readAssignments(switchReduceEntry).nodes();
Set<Assignment> retrievedSwitchAppend = readAssignments(switchAppendEntry).nodes();
- Assignments pendingAssignments = readAssignments(pendingEntry);
+ Assignments pendingAssignments = pendingEntry.value() == null
+ ? Assignments.EMPTY
+ : AssignmentsQueue.fromBytes(pendingEntry.value()).poll();
Set<Assignment> retrievedPending = pendingAssignments.nodes();
if (!retrievedPending.equals(stableFromRaft)) {
@@ -404,8 +407,8 @@ public class ZoneRebalanceRaftGroupEventsListener implements RaftGroupEventsList
long catalogTimestamp = pendingAssignments.timestamp();
byte[] stableFromRaftByteArray = Assignments.toBytes(stableFromRaft, catalogTimestamp);
- byte[] additionByteArray = Assignments.toBytes(calculatedPendingAddition, catalogTimestamp);
- byte[] reductionByteArray = Assignments.toBytes(calculatedPendingReduction, catalogTimestamp);
+ byte[] additionByteArray = AssignmentsQueue.toBytes(Assignments.of(calculatedPendingAddition, catalogTimestamp));
+ byte[] reductionByteArray = AssignmentsQueue.toBytes(Assignments.of(calculatedPendingReduction, catalogTimestamp));
byte[] switchReduceByteArray = Assignments.toBytes(calculatedSwitchReduce, catalogTimestamp);
byte[] switchAppendByteArray = Assignments.toBytes(calculatedSwitchAppend, catalogTimestamp);
@@ -433,7 +436,7 @@ public class ZoneRebalanceRaftGroupEventsListener implements RaftGroupEventsList
successCase = ops(
put(stablePartAssignmentsKey, stableFromRaftByteArray),
- put(pendingPartAssignmentsKey, plannedEntry.value()),
+ put(pendingPartAssignmentsKey, AssignmentsQueue.toBytes(Assignments.fromBytes(plannedEntry.value()))),
remove(plannedPartAssignmentsKey)
).yield(SCHEDULE_PENDING_REBALANCE_SUCCESS);
@@ -584,11 +587,11 @@ public class ZoneRebalanceRaftGroupEventsListener implements RaftGroupEventsList
Set<Assignment> assignments = calculateAssignmentForPartition(dataNodes, partId.partitionId(), partitions, replicas);
- ByteArray pendingKey = pendingPartAssignmentsKey(partId);
+ ByteArray pendingKey = pendingPartAssignmentsQueueKey(partId);
Set<Assignment> pendingAssignments = difference(assignments, switchReduce.nodes());
- byte[] pendingByteArray = Assignments.toBytes(pendingAssignments, assignmentsTimestamp);
+ byte[] pendingByteArray = AssignmentsQueue.toBytes(Assignments.of(pendingAssignments, assignmentsTimestamp));
byte[] assignmentsByteArray = Assignments.toBytes(assignments, assignmentsTimestamp);
ByteArray changeTriggerKey = ZoneRebalanceUtil.pendingChangeTriggerKey(partId);
diff --git a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/ZoneRebalanceUtil.java b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/ZoneRebalanceUtil.java
index 54be01f8f7c..f30a5d42aba 100644
--- a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/ZoneRebalanceUtil.java
+++ b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/ZoneRebalanceUtil.java
@@ -37,6 +37,7 @@ import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
import static org.apache.ignite.internal.metastorage.dsl.Operations.remove;
import static org.apache.ignite.internal.metastorage.dsl.Statements.iif;
import static org.apache.ignite.internal.partitiondistribution.PartitionDistributionUtils.calculateAssignmentForPartition;
+import static org.apache.ignite.internal.partitiondistribution.PendingAssignmentsCalculator.pendingAssignmentsCalculator;
import static org.apache.ignite.internal.util.ByteUtils.longToBytesKeepingOrder;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
@@ -62,6 +63,7 @@ import org.apache.ignite.internal.metastorage.dsl.Condition;
import org.apache.ignite.internal.metastorage.dsl.Iif;
import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.partitiondistribution.Assignments;
+import org.apache.ignite.internal.partitiondistribution.AssignmentsQueue;
import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.util.ExceptionUtils;
import org.apache.ignite.internal.util.IgniteSpinBusyLock;
@@ -152,7 +154,7 @@ public class ZoneRebalanceUtil {
) {
ByteArray partChangeTriggerKey = pendingChangeTriggerKey(zonePartitionId);
- ByteArray partAssignmentsPendingKey = pendingPartAssignmentsKey(zonePartitionId);
+ ByteArray partAssignmentsPendingKey = pendingPartAssignmentsQueueKey(zonePartitionId);
ByteArray partAssignmentsPlannedKey = plannedPartAssignmentsKey(zonePartitionId);
@@ -160,7 +162,7 @@ public class ZoneRebalanceUtil {
Set<Assignment> calculatedAssignments = calculateAssignmentForPartition(dataNodes, partNum, partitions, replicas);
- Set<Assignment> partAssignments;
+ Set<Assignment> targetAssignmentSet;
if (consistencyMode == ConsistencyMode.HIGH_AVAILABILITY) {
// All complicated logic here is needed because we want to return back to stable nodes
@@ -187,26 +189,33 @@ public class ZoneRebalanceUtil {
}
}
- partAssignments = resultingAssignments;
+ targetAssignmentSet = resultingAssignments;
} else {
- partAssignments = calculatedAssignments;
+ targetAssignmentSet = calculatedAssignments;
}
- boolean isNewAssignments = !zoneCfgPartAssignments.equals(partAssignments);
+ boolean isNewAssignments = !zoneCfgPartAssignments.equals(targetAssignmentSet);
- byte[] partAssignmentsBytes = Assignments.toBytes(partAssignments, assignmentsTimestamp);
+ Assignments targetAssignments = Assignments.of(targetAssignmentSet, assignmentsTimestamp);
+ AssignmentsQueue partAssignmentsPendingQueue = pendingAssignmentsCalculator()
+ .stable(Assignments.of(zoneCfgPartAssignments, assignmentsTimestamp))
+ .target(targetAssignments)
+ .toQueue();
+
+ byte[] partAssignmentsPlannedBytes = targetAssignments.toBytes();
+ byte[] partAssignmentsPendingBytes = partAssignmentsPendingQueue.toBytes();
// if empty(partition.change.trigger.revision) || partition.change.trigger.revision < event.revision:
// if empty(partition.assignments.pending)
// && ((isNewAssignments && empty(partition.assignments.stable))
// || (partition.assignments.stable != calcPartAssignments() && !empty(partition.assignments.stable))):
- // partition.assignments.pending = calcPartAssignments()
+ // partition.assignments.pending = partAssignmentsPendingQueue
// partition.change.trigger.revision = event.revision
// else:
- // if partition.assignments.pending != calcPartAssignments && !empty(partition.assignments.pending)
+ // if partition.assignments.pending != partAssignmentsPendingQueue && !empty(partition.assignments.pending)
// partition.assignments.planned = calcPartAssignments()
// partition.change.trigger.revision = event.revision
- // else if partition.assignments.pending == calcPartAssignments
+ // else if partition.assignments.pending == partAssignmentsPendingQueue
// remove(partition.assignments.planned)
// partition.change.trigger.revision = event.revision
// message after the metastorage invoke:
@@ -219,7 +228,8 @@ public class ZoneRebalanceUtil {
// else:
// skip
- Condition newAssignmentsCondition = exists(partAssignmentsStableKey).and(value(partAssignmentsStableKey).ne(partAssignmentsBytes));
+ Condition newAssignmentsCondition = exists(partAssignmentsStableKey)
+ .and(value(partAssignmentsStableKey).ne(partAssignmentsPlannedBytes));
if (isNewAssignments) {
newAssignmentsCondition = notExists(partAssignmentsStableKey).or(newAssignmentsCondition);
@@ -230,15 +240,15 @@ public class ZoneRebalanceUtil {
Iif iif = iif(or(notExists(partChangeTriggerKey), value(partChangeTriggerKey).lt(revisionBytes)),
iif(and(notExists(partAssignmentsPendingKey), newAssignmentsCondition),
ops(
- put(partAssignmentsPendingKey, partAssignmentsBytes),
+ put(partAssignmentsPendingKey, partAssignmentsPendingBytes),
put(partChangeTriggerKey, revisionBytes)
).yield(PENDING_KEY_UPDATED.ordinal()),
- iif(and(value(partAssignmentsPendingKey).ne(partAssignmentsBytes), exists(partAssignmentsPendingKey)),
+ iif(and(value(partAssignmentsPendingKey).ne(partAssignmentsPendingBytes), exists(partAssignmentsPendingKey)),
ops(
- put(partAssignmentsPlannedKey, partAssignmentsBytes),
+ put(partAssignmentsPlannedKey, partAssignmentsPlannedBytes),
put(partChangeTriggerKey, revisionBytes)
).yield(PLANNED_KEY_UPDATED.ordinal()),
- iif(value(partAssignmentsPendingKey).eq(partAssignmentsBytes),
+ iif(value(partAssignmentsPendingKey).eq(partAssignmentsPendingBytes),
ops(
remove(partAssignmentsPlannedKey),
put(partChangeTriggerKey, revisionBytes)
@@ -258,14 +268,14 @@ public class ZoneRebalanceUtil {
LOG.info(
"Update metastore pending partitions key [key={}, partition={}, zone={}/{}, newVal={}]",
partAssignmentsPendingKey.toString(), partNum, zoneDescriptor.id(), zoneDescriptor.name(),
- partAssignments);
+ partAssignmentsPendingQueue);
break;
case PLANNED_KEY_UPDATED:
LOG.info(
"Update metastore planned partitions key [key={}, partition={}, zone={}/{}, newVal={}]",
partAssignmentsPlannedKey, partNum, zoneDescriptor.id(), zoneDescriptor.name(),
- partAssignments
+ targetAssignmentSet
);
break;
@@ -273,7 +283,7 @@ public class ZoneRebalanceUtil {
LOG.info(
"Remove planned key because current pending key has the same value [key={}, partition={}, zone={}/{}, val={}]",
partAssignmentsPlannedKey.toString(), partNum, zoneDescriptor.id(), zoneDescriptor.name(),
- partAssignments
+ targetAssignmentSet
);
break;
@@ -282,7 +292,7 @@ public class ZoneRebalanceUtil {
"Remove planned key because pending is empty and calculated assignments are equal to current assignments "
+ "[key={}, partition={}, zone={}/{}, val={}]",
partAssignmentsPlannedKey.toString(), partNum, zoneDescriptor.id(), zoneDescriptor.name(),
- partAssignments
+ targetAssignmentSet
);
break;
@@ -290,7 +300,7 @@ public class ZoneRebalanceUtil {
LOG.debug(
"Assignments are not updated [key={}, partition={}, zone={}/{}, val={}]",
partAssignmentsPlannedKey.toString(), partNum, zoneDescriptor.id(), zoneDescriptor.name(),
- partAssignments
+ targetAssignmentSet
);
break;
@@ -403,9 +413,9 @@ public class ZoneRebalanceUtil {
}
/** Key prefix for pending assignments. */
- public static final String PENDING_ASSIGNMENTS_PREFIX = "zone.assignments.pending.";
+ public static final String PENDING_ASSIGNMENTS_QUEUE_PREFIX = "zone.assignments.pending.";
- public static final byte[] PENDING_ASSIGNMENTS_PREFIX_BYTES = PENDING_ASSIGNMENTS_PREFIX.getBytes(UTF_8);
+ public static final byte[] PENDING_ASSIGNMENTS_QUEUE_PREFIX_BYTES = PENDING_ASSIGNMENTS_QUEUE_PREFIX.getBytes(UTF_8);
/** Key prefix for stable assignments. */
public static final String STABLE_ASSIGNMENTS_PREFIX = "zone.assignments.stable.";
@@ -441,8 +451,8 @@ public class ZoneRebalanceUtil {
* @return Key for a partition.
* @see <a href="https://github.com/apache/ignite-3/blob/main/modules/table/tech-notes/rebalance.md">Rebalance documentation</a>
*/
- public static ByteArray pendingPartAssignmentsKey(ZonePartitionId zonePartitionId) {
- return new ByteArray(PENDING_ASSIGNMENTS_PREFIX + zonePartitionId);
+ public static ByteArray pendingPartAssignmentsQueueKey(ZonePartitionId zonePartitionId) {
+ return new ByteArray(PENDING_ASSIGNMENTS_QUEUE_PREFIX + zonePartitionId);
}
/**
@@ -663,8 +673,8 @@ public class ZoneRebalanceUtil {
int partitionId,
long revision
) {
- Entry entry = metaStorageManager.getLocally(pendingPartAssignmentsKey(new ZonePartitionId(zoneId, partitionId)), revision);
+ Entry entry = metaStorageManager.getLocally(pendingPartAssignmentsQueueKey(new ZonePartitionId(zoneId, partitionId)), revision);
- return entry != null ? Assignments.fromBytes(entry.value()) : null;
+ return entry != null ? AssignmentsQueue.fromBytes(entry.value()).poll() : null;
}
}
diff --git a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/DistributionZoneRebalanceEngineTest.java b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/DistributionZoneRebalanceEngineTest.java
index c4c41fe7333..2bfaa113cc9 100644
--- a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/DistributionZoneRebalanceEngineTest.java
+++ b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/DistributionZoneRebalanceEngineTest.java
@@ -97,6 +97,7 @@ import org.apache.ignite.internal.network.ClusterService;
import org.apache.ignite.internal.network.MessagingService;
import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.partitiondistribution.Assignments;
+import org.apache.ignite.internal.partitiondistribution.AssignmentsQueue;
import org.apache.ignite.internal.raft.Command;
import org.apache.ignite.internal.raft.WriteCommand;
import org.apache.ignite.internal.raft.service.CommandClosure;
@@ -305,7 +306,7 @@ public class DistributionZoneRebalanceEngineTest extends IgniteAbstractTest {
zoneNodes.put(zoneId, nodes);
- checkAssignments(zoneNodes, RebalanceUtil::pendingPartAssignmentsKey);
+ checkAssignments(zoneNodes, RebalanceUtil::pendingPartAssignmentsQueueKey, bytes -> AssignmentsQueue.fromBytes(bytes).poll());
verify(keyValueStorage, timeout(1000).times(8)).invoke(any(), any(), any());
}
@@ -328,7 +329,7 @@ public class DistributionZoneRebalanceEngineTest extends IgniteAbstractTest {
zoneNodes.put(zoneId, nodes);
- checkAssignments(zoneNodes, RebalanceUtil::pendingPartAssignmentsKey);
+ checkAssignments(zoneNodes, RebalanceUtil::pendingPartAssignmentsQueueKey, bytes -> AssignmentsQueue.fromBytes(bytes).poll());
verify(keyValueStorage, timeout(1000).times(1)).invoke(any(), any(), any());
@@ -364,7 +365,7 @@ public class DistributionZoneRebalanceEngineTest extends IgniteAbstractTest {
zoneNodes.put(zoneId, nodes);
- checkAssignments(zoneNodes, RebalanceUtil::pendingPartAssignmentsKey);
+ checkAssignments(zoneNodes, RebalanceUtil::pendingPartAssignmentsQueueKey, bytes -> AssignmentsQueue.fromBytes(bytes).poll());
verify(keyValueStorage, timeout(1000).times(1)).invoke(any(), any(), any());
@@ -398,7 +399,7 @@ public class DistributionZoneRebalanceEngineTest extends IgniteAbstractTest {
zoneNodes.put(zoneId, nodes);
- checkAssignments(zoneNodes, RebalanceUtil::pendingPartAssignmentsKey);
+ checkAssignments(zoneNodes, RebalanceUtil::pendingPartAssignmentsQueueKey, bytes -> AssignmentsQueue.fromBytes(bytes).poll());
verify(keyValueStorage, timeout(1000).times(1)).invoke(any(), any(), any());
@@ -406,7 +407,7 @@ public class DistributionZoneRebalanceEngineTest extends IgniteAbstractTest {
watchListenerOnUpdate(zoneId, nodes2, 1);
- checkAssignments(zoneNodes, RebalanceUtil::pendingPartAssignmentsKey);
+ checkAssignments(zoneNodes, RebalanceUtil::pendingPartAssignmentsQueueKey, bytes -> AssignmentsQueue.fromBytes(bytes).poll());
TablePartitionId partId = new TablePartitionId(getTableId(TABLE_NAME), 0);
@@ -497,6 +498,14 @@ public class DistributionZoneRebalanceEngineTest extends IgniteAbstractTest {
}
private void checkAssignments(Map<Integer, Set<String>> zoneNodes, Function<TablePartitionId, ByteArray> assignmentFunction) {
+ checkAssignments(zoneNodes, assignmentFunction, Assignments::fromBytes);
+ }
+
+ private void checkAssignments(
+ Map<Integer, Set<String>> zoneNodes,
+ Function<TablePartitionId, ByteArray> assignmentFunction,
+ Function<byte[], Assignments> assignmentsFromBytesFunction
+ ) {
Catalog catalog = catalogManager.catalog(catalogManager.latestCatalogVersion());
catalog.tables().forEach(tableDescriptor -> {
@@ -520,7 +529,7 @@ public class DistributionZoneRebalanceEngineTest extends IgniteAbstractTest {
assertNotNull(actualAssignmentsBytes);
- Set<String> actualAssignments = Assignments.fromBytes(actualAssignmentsBytes).nodes()
+ Set<String> actualAssignments = assignmentsFromBytesFunction.apply(actualAssignmentsBytes).nodes()
.stream().map(Assignment::consistentId).collect(toSet());
assertTrue(expectedAssignments.containsAll(actualAssignments));
diff --git a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceMinimumRequiredTimeProviderImplTest.java b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceMinimumRequiredTimeProviderImplTest.java
index 8f1f5d475f3..821af11b469 100644
--- a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceMinimumRequiredTimeProviderImplTest.java
+++ b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceMinimumRequiredTimeProviderImplTest.java
@@ -46,6 +46,7 @@ import org.apache.ignite.internal.distributionzones.BaseDistributionZoneManagerT
import org.apache.ignite.internal.lang.ByteArray;
import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.partitiondistribution.Assignments;
+import org.apache.ignite.internal.partitiondistribution.AssignmentsQueue;
import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.sql.SqlCommon;
import org.apache.ignite.internal.util.ByteUtils;
@@ -401,7 +402,11 @@ class RebalanceMinimumRequiredTimeProviderImplTest extends BaseDistributionZoneM
Catalog catalog,
boolean allPartitions
) throws Exception {
- saveAssignments(false, zoneName, tableId, catalog, RebalanceUtil::pendingPartAssignmentsKey, allPartitions);
+ Function<Long, byte[]> valueFunction = timestamp -> new AssignmentsQueue(
+ Assignments.of(timestamp, Assignment.forPeer("nodeName"))
+ ).toBytes();
+
+ saveAssignments(false, zoneName, tableId, catalog, RebalanceUtil::pendingPartAssignmentsQueueKey, valueFunction, allPartitions);
}
private void savePlannedAssignments(String zoneName,
@@ -419,6 +424,19 @@ class RebalanceMinimumRequiredTimeProviderImplTest extends BaseDistributionZoneM
Catalog catalog,
Function<TablePartitionId, ByteArray> keyFunction,
boolean allPartitions
+ ) throws Exception {
+ Function<Long, byte[]> valueFunction = timestamp -> Assignments.of(timestamp, Assignment.forPeer("nodeName")).toBytes();
+ saveAssignments(stable, zoneName, tableId, catalog, keyFunction, valueFunction, allPartitions);
+ }
+
+ private void saveAssignments(
+ boolean stable,
+ String zoneName,
+ int tableId,
+ Catalog catalog,
+ Function<TablePartitionId, ByteArray> keyFunction,
+ Function<Long, byte[]> valueFunction,
+ boolean allPartitions
) throws Exception {
long timestamp = catalog.time();
@@ -445,7 +463,7 @@ class RebalanceMinimumRequiredTimeProviderImplTest extends BaseDistributionZoneM
metaStorageData.put(
keyFunction.apply(tablePartitionId),
- Assignments.of(timestamp, Assignment.forPeer("nodeName")).toBytes()
+ valueFunction.apply(timestamp)
);
}
}
diff --git a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceUtilUpdateAssignmentsTest.java b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceUtilUpdateAssignmentsTest.java
index f6df8cb16a3..44fc2f530ca 100644
--- a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceUtilUpdateAssignmentsTest.java
+++ b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceUtilUpdateAssignmentsTest.java
@@ -68,6 +68,7 @@ import org.apache.ignite.internal.network.ClusterService;
import org.apache.ignite.internal.network.MessagingService;
import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.partitiondistribution.Assignments;
+import org.apache.ignite.internal.partitiondistribution.AssignmentsQueue;
import org.apache.ignite.internal.raft.Command;
import org.apache.ignite.internal.raft.WriteCommand;
import org.apache.ignite.internal.raft.service.CommandClosure;
@@ -286,8 +287,8 @@ public class RebalanceUtilUpdateAssignmentsTest extends IgniteAbstractTest {
if (currentPendingAssignments != null) {
keyValueStorage.put(
- RebalanceUtil.pendingPartAssignmentsKey(tablePartitionId).bytes(),
- toBytes(currentPendingAssignments, assignmentsTimestamp),
+ RebalanceUtil.pendingPartAssignmentsQueueKey(tablePartitionId).bytes(),
+ AssignmentsQueue.toBytes(Assignments.of(currentPendingAssignments, assignmentsTimestamp)),
KV_UPDATE_CONTEXT
);
}
@@ -328,11 +329,11 @@ public class RebalanceUtilUpdateAssignmentsTest extends IgniteAbstractTest {
actualStableAssignments = Assignments.fromBytes(actualStableBytes).nodes();
}
- byte[] actualPendingBytes = keyValueStorage.get(RebalanceUtil.pendingPartAssignmentsKey(tablePartitionId).bytes()).value();
+ byte[] actualPendingBytes = keyValueStorage.get(RebalanceUtil.pendingPartAssignmentsQueueKey(tablePartitionId).bytes()).value();
Set<Assignment> actualPendingAssignments = null;
if (actualPendingBytes != null) {
- actualPendingAssignments = Assignments.fromBytes(actualPendingBytes).nodes();
+ actualPendingAssignments = AssignmentsQueue.fromBytes(actualPendingBytes).poll().nodes();
}
byte[] actualPlannedBytes = keyValueStorage.get(RebalanceUtil.plannedPartAssignmentsKey(tablePartitionId).bytes()).value();
diff --git a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/ZoneRebalanceUtilUpdateAssignmentsTest.java b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/ZoneRebalanceUtilUpdateAssignmentsTest.java
index 7f7d6330c36..1db28fcf010 100644
--- a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/ZoneRebalanceUtilUpdateAssignmentsTest.java
+++ b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/ZoneRebalanceUtilUpdateAssignmentsTest.java
@@ -68,6 +68,7 @@ import org.apache.ignite.internal.network.ClusterService;
import org.apache.ignite.internal.network.MessagingService;
import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.partitiondistribution.Assignments;
+import org.apache.ignite.internal.partitiondistribution.AssignmentsQueue;
import org.apache.ignite.internal.raft.Command;
import org.apache.ignite.internal.raft.WriteCommand;
import org.apache.ignite.internal.raft.service.CommandClosure;
@@ -292,8 +293,8 @@ public class ZoneRebalanceUtilUpdateAssignmentsTest extends IgniteAbstractTest {
if (currentPendingAssignments != null) {
keyValueStorage.put(
- ZoneRebalanceUtil.pendingPartAssignmentsKey(zonePartitionId).bytes(),
- toBytes(currentPendingAssignments, assignmentsTimestamp),
+ ZoneRebalanceUtil.pendingPartAssignmentsQueueKey(zonePartitionId).bytes(),
+ AssignmentsQueue.toBytes(Assignments.of(currentPendingAssignments, assignmentsTimestamp)),
KV_UPDATE_CONTEXT
);
}
@@ -334,11 +335,11 @@ public class ZoneRebalanceUtilUpdateAssignmentsTest extends IgniteAbstractTest {
actualStableAssignments = Assignments.fromBytes(actualStableBytes).nodes();
}
- byte[] actualPendingBytes = keyValueStorage.get(ZoneRebalanceUtil.pendingPartAssignmentsKey(zonePartitionId).bytes()).value();
+ byte[] actualPendingBytes = keyValueStorage.get(ZoneRebalanceUtil.pendingPartAssignmentsQueueKey(zonePartitionId).bytes()).value();
Set<Assignment> actualPendingAssignments = null;
if (actualPendingBytes != null) {
- actualPendingAssignments = Assignments.fromBytes(actualPendingBytes).nodes();
+ actualPendingAssignments = AssignmentsQueue.fromBytes(actualPendingBytes).poll().nodes();
}
byte[] actualPlannedBytes = keyValueStorage.get(ZoneRebalanceUtil.plannedPartAssignmentsKey(zonePartitionId).bytes()).value();
diff --git a/modules/partition-distribution/src/main/java/org/apache/ignite/internal/partitiondistribution/AssignmentsQueue.java b/modules/partition-distribution/src/main/java/org/apache/ignite/internal/partitiondistribution/AssignmentsQueue.java
new file mode 100644
index 00000000000..46403d9abaf
--- /dev/null
+++ b/modules/partition-distribution/src/main/java/org/apache/ignite/internal/partitiondistribution/AssignmentsQueue.java
@@ -0,0 +1,140 @@
+/*
+ * 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.Arrays;
+import java.util.Collection;
+import java.util.Deque;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.Objects;
+import org.apache.ignite.internal.tostring.IgniteToStringInclude;
+import org.apache.ignite.internal.tostring.S;
+import org.apache.ignite.internal.versioned.VersionedSerialization;
+import org.jetbrains.annotations.Contract;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Class that encapsulates a queue with consequent configuration switches
+ * e.g. promoting a learner to a peer or demoting a peer to a learner in following steps:
+ * <ul>
+ * <li> node removed from configuration
+ * <li> node added to configuration with different type (peer or learner)
+ * </ul>
+ */
+public class AssignmentsQueue implements Iterable<Assignments> {
+
+ @IgniteToStringInclude
+ private final Deque<Assignments> queue;
+
+ /** Constructor. */
+ public AssignmentsQueue(Assignments... assignments) {
+ this(Arrays.asList(assignments));
+ }
+
+ /** Constructor. */
+ AssignmentsQueue(Collection<Assignments> assignments) {
+ this.queue = new LinkedList<>(assignments);
+ }
+
+ /**
+ * Retrieves and removes the head of this queue.
+ *
+ * @return the head of this queue
+ */
+ public @Nullable Assignments poll() {
+ assert !queue.isEmpty() : "Assignments queue must contain at least one element.";
+ return queue.poll();
+ }
+
+ /**
+ * Retrieves, but does not remove, the last element of this queue.
+ *
+ * @return the tail of this queue
+ */
+ public @Nullable Assignments peekLast() {
+ assert !queue.isEmpty() : "Assignments queue must contain at least one element.";
+ return queue.peekLast();
+ }
+
+ public byte[] toBytes() {
+ return VersionedSerialization.toBytes(this, AssignmentsQueueSerializer.INSTANCE);
+ }
+
+ public static byte[] toBytes(Assignments... assignments) {
+ return toBytes(Arrays.asList(assignments));
+ }
+
+ public static byte[] toBytes(Collection<Assignments> assignments) {
+ return new AssignmentsQueue(assignments).toBytes();
+ }
+
+ @Nullable
+ @Contract("null -> null; !null -> !null")
+ public static AssignmentsQueue fromBytes(byte @Nullable [] bytes) {
+ return bytes == null ? null : VersionedSerialization.fromBytes(bytes, AssignmentsQueueSerializer.INSTANCE);
+ }
+
+ /**
+ * Returns the number of elements in this queue.
+ *
+ * @return the number of elements in this queue
+ */
+ public int size() {
+ return queue.size();
+ }
+
+ public boolean isEmpty() {
+ return queue.isEmpty();
+ }
+
+ /**
+ * Returns an iterator over the elements in this queue in proper sequence.
+ * The elements will be returned in order from first (head) to last (tail).
+ *
+ * @return an iterator over the elements in this queue in proper sequence
+ */
+ @Override
+ public Iterator<Assignments> iterator() {
+ return queue.iterator();
+ }
+
+ @Override
+ public String toString() {
+ return S.toString(this);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ AssignmentsQueue that = (AssignmentsQueue) o;
+ return Objects.equals(queue, that.queue);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(queue);
+ }
+}
diff --git a/modules/partition-distribution/src/main/java/org/apache/ignite/internal/partitiondistribution/AssignmentsQueueSerializer.java b/modules/partition-distribution/src/main/java/org/apache/ignite/internal/partitiondistribution/AssignmentsQueueSerializer.java
new file mode 100644
index 00000000000..53bbf047ac0
--- /dev/null
+++ b/modules/partition-distribution/src/main/java/org/apache/ignite/internal/partitiondistribution/AssignmentsQueueSerializer.java
@@ -0,0 +1,52 @@
+/*
+ * 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 java.util.LinkedList;
+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 AssignmentsQueue} instances.
+ */
+public class AssignmentsQueueSerializer extends VersionedSerializer<AssignmentsQueue> {
+ /** Serializer instance. */
+ public static final AssignmentsQueueSerializer INSTANCE = new AssignmentsQueueSerializer();
+
+ @Override
+ protected void writeExternalData(AssignmentsQueue queue, IgniteDataOutput out) throws IOException {
+ out.writeVarInt(queue.size());
+ for (Assignments assignments : queue) {
+ AssignmentsSerializer.INSTANCE.writeExternalData(assignments, out);
+ }
+ }
+
+ @Override
+ protected AssignmentsQueue readExternalData(byte protoVer, IgniteDataInput in) throws IOException {
+ int length = in.readVarIntAsInt();
+
+ LinkedList<Assignments> queue = new LinkedList<>();
+ for (int i = 0; i < length; i++) {
+ queue.add(AssignmentsSerializer.INSTANCE.readExternalData(protoVer, in));
+ }
+
+ return new AssignmentsQueue(queue);
+ }
+}
diff --git a/modules/partition-distribution/src/main/java/org/apache/ignite/internal/partitiondistribution/PendingAssignmentsCalculator.java b/modules/partition-distribution/src/main/java/org/apache/ignite/internal/partitiondistribution/PendingAssignmentsCalculator.java
new file mode 100644
index 00000000000..52a4831c197
--- /dev/null
+++ b/modules/partition-distribution/src/main/java/org/apache/ignite/internal/partitiondistribution/PendingAssignmentsCalculator.java
@@ -0,0 +1,63 @@
+/*
+ * 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;
+
+/**
+ * Calculates the pending assignments queue between current stable and target assignments.
+ */
+public class PendingAssignmentsCalculator {
+ private Assignments stable;
+ private Assignments target;
+
+ private PendingAssignmentsCalculator() {}
+
+ public static PendingAssignmentsCalculator pendingAssignmentsCalculator() {
+ return new PendingAssignmentsCalculator();
+ }
+
+ /**
+ * Current stable assignments.
+ */
+ public PendingAssignmentsCalculator stable(Assignments stable) {
+ this.stable = stable;
+ return this;
+ }
+
+ /**
+ * Target assignments.
+ */
+ public PendingAssignmentsCalculator target(Assignments target) {
+ this.target = target;
+ return this;
+ }
+
+ /**
+ * Calculates the pending assignments queue between current stable and target assignments.
+ */
+ public AssignmentsQueue toQueue() {
+ assert stable != null;
+ assert target != null;
+
+ // naive version with only one configuration switch, calculation will be changed during IGNITE-23790 epic
+ AssignmentsQueue queue = new AssignmentsQueue(target);
+ assert Objects.equals(queue.peekLast(), target) : "Target assignments should be equal to the last element in the queue";
+ return queue;
+ }
+}
diff --git a/modules/partition-distribution/src/test/java/org/apache/ignite/internal/partitiondistribution/AssignmentsQueueSerializerTest.java b/modules/partition-distribution/src/test/java/org/apache/ignite/internal/partitiondistribution/AssignmentsQueueSerializerTest.java
new file mode 100644
index 00000000000..023ecea6b4e
--- /dev/null
+++ b/modules/partition-distribution/src/test/java/org/apache/ignite/internal/partitiondistribution/AssignmentsQueueSerializerTest.java
@@ -0,0 +1,93 @@
+/*
+ * 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 org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.time.LocalDateTime;
+import java.time.Month;
+import java.time.ZoneOffset;
+import java.util.Base64;
+import java.util.Set;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.versioned.VersionedSerialization;
+import org.apache.ignite.internal.versioned.VersionedSerializer;
+import org.junit.jupiter.api.Test;
+import org.junitpioneer.jupiter.cartesian.CartesianTest;
+import org.junitpioneer.jupiter.cartesian.CartesianTest.Values;
+
+class AssignmentsQueueSerializerTest {
+ private static final String ASSIGNMENTS_QUEUE_V1 =
+ "Ae++QwQDBGFiYwEEZGVmAAFRwowBAPQGAAMEYWJjAQRkZWYAAFHCjAEA9AYBAwRhYmMBBGRlZgAAUcKMAQD0BgA=";
+
+ private static final long BASE_PHYSICAL_TIME = LocalDateTime.of(2024, Month.JANUARY, 1, 0, 0)
+ .atOffset(ZoneOffset.UTC)
+ .toInstant()
+ .toEpochMilli();
+
+ private final AssignmentsQueueSerializer serializer = new AssignmentsQueueSerializer();
+
+ @CartesianTest
+ void serializationAndDeserialization(
+ @Values(booleans = {true, false}) boolean force,
+ @Values(booleans = {true, false}) boolean fromReset
+ ) {
+ AssignmentsQueue originalAssignmentsQueue = new AssignmentsQueue(testAssignments(force, fromReset));
+
+ byte[] bytes = VersionedSerialization.toBytes(originalAssignmentsQueue, serializer);
+ AssignmentsQueue restoredAssignmentsQueue = VersionedSerialization.fromBytes(bytes, serializer);
+
+ assertThat(restoredAssignmentsQueue, equalTo(originalAssignmentsQueue));
+ }
+
+ @Test
+ void v1CanBeDeserialized() {
+ byte[] bytes = Base64.getDecoder().decode(ASSIGNMENTS_QUEUE_V1);
+ AssignmentsQueue restoredAssignmentsQueue = VersionedSerialization.fromBytes(bytes, serializer);
+
+ assertFalse(restoredAssignmentsQueue.isEmpty());
+ assertThat(restoredAssignmentsQueue.poll(), equalTo(testAssignments(true, false)));
+ assertThat(restoredAssignmentsQueue.poll(), equalTo(testAssignments(false, true)));
+ assertThat(restoredAssignmentsQueue.poll(), equalTo(testAssignments(false, false)));
+
+ assertTrue(restoredAssignmentsQueue.isEmpty());
+ assertThrows(AssertionError.class, restoredAssignmentsQueue::poll);
+ }
+
+ private static Assignments testAssignments(boolean force, boolean fromReset) {
+ Set<Assignment> nodes = Set.of(Assignment.forPeer("abc"), Assignment.forLearner("def"));
+
+ return force
+ ? Assignments.forced(nodes, baseTimestamp(5))
+ : Assignments.of(nodes, baseTimestamp(5), fromReset);
+ }
+
+ private static long baseTimestamp(int logical) {
+ return new HybridTimestamp(BASE_PHYSICAL_TIME, logical).longValue();
+ }
+
+ @SuppressWarnings("unused")
+ private static <T> String v1Base64(T object, VersionedSerializer<T> serializer) {
+ byte[] v1Bytes = VersionedSerialization.toBytes(object, serializer);
+ return Base64.getEncoder().encodeToString(v1Bytes);
+ }
+}
diff --git a/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/fixtures/Node.java b/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/fixtures/Node.java
index b3232715e47..efc33e8f91d 100644
--- a/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/fixtures/Node.java
+++ b/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/fixtures/Node.java
@@ -23,7 +23,7 @@ import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.apache.ignite.internal.BaseIgniteRestartTest.createVault;
import static org.apache.ignite.internal.configuration.IgnitePaths.partitionsPath;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.REBALANCE_SCHEDULER_POOL_SIZE;
-import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.pendingPartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.pendingPartAssignmentsQueueKey;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.testNodeName;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willSucceedIn;
@@ -591,7 +591,7 @@ public class Node {
partitionRaftConfigurer,
view -> new LocalLogStorageFactory(),
threadPoolsManager.tableIoExecutor(),
- replicaGrpId -> metaStorageManager.get(pendingPartAssignmentsKey((ZonePartitionId) replicaGrpId))
+ replicaGrpId -> metaStorageManager.get(pendingPartAssignmentsQueueKey((ZonePartitionId) replicaGrpId))
.thenApply(Entry::value)
);
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java
index 466c9d82987..2b1d8c9397b 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java
@@ -33,10 +33,10 @@ import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUt
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.union;
import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceRaftGroupEventsListener.handleReduceChanged;
import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.ASSIGNMENTS_SWITCH_REDUCE_PREFIX_BYTES;
-import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.PENDING_ASSIGNMENTS_PREFIX_BYTES;
+import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.PENDING_ASSIGNMENTS_QUEUE_PREFIX_BYTES;
import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.STABLE_ASSIGNMENTS_PREFIX_BYTES;
import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.extractZonePartitionId;
-import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.pendingPartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.pendingPartAssignmentsQueueKey;
import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.stablePartAssignmentsKey;
import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.zoneAssignmentsGetLocally;
import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.zonePartitionAssignmentsGetLocally;
@@ -119,6 +119,7 @@ import org.apache.ignite.internal.partition.replicator.schema.CatalogValidationS
import org.apache.ignite.internal.partition.replicator.schema.ExecutorInclinedSchemaSyncService;
import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.partitiondistribution.Assignments;
+import org.apache.ignite.internal.partitiondistribution.AssignmentsQueue;
import org.apache.ignite.internal.placementdriver.PlacementDriver;
import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEvent;
import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEventParameters;
@@ -357,7 +358,7 @@ public class PartitionReplicaLifecycleManager extends
CompletableFuture<Void> processZonesAndAssignmentsOnStart = processZonesOnStart(recoveryRevision, lowWatermark.getLowWatermark())
.thenCompose(ignored -> processAssignmentsOnRecovery(recoveryRevision));
- metaStorageMgr.registerPrefixWatch(new ByteArray(PENDING_ASSIGNMENTS_PREFIX_BYTES), pendingAssignmentsRebalanceListener);
+ metaStorageMgr.registerPrefixWatch(new ByteArray(PENDING_ASSIGNMENTS_QUEUE_PREFIX_BYTES), pendingAssignmentsRebalanceListener);
metaStorageMgr.registerPrefixWatch(new ByteArray(STABLE_ASSIGNMENTS_PREFIX_BYTES), stableAssignmentsRebalanceListener);
metaStorageMgr.registerPrefixWatch(new ByteArray(ASSIGNMENTS_SWITCH_REDUCE_PREFIX_BYTES), assignmentsSwitchRebalanceListener);
@@ -414,7 +415,7 @@ public class PartitionReplicaLifecycleManager extends
private CompletableFuture<Void> processAssignmentsOnRecovery(long recoveryRevision) {
var stableAssignmentsPrefix = new ByteArray(STABLE_ASSIGNMENTS_PREFIX_BYTES);
- var pendingAssignmentsPrefix = new ByteArray(PENDING_ASSIGNMENTS_PREFIX_BYTES);
+ var pendingAssignmentsPrefix = new ByteArray(PENDING_ASSIGNMENTS_QUEUE_PREFIX_BYTES);
// It's required to handle stable assignments changes on recovery in order to cleanup obsolete resources.
CompletableFuture<Void> stableFuture = handleAssignmentsOnRecovery(
@@ -941,13 +942,13 @@ public class PartitionReplicaLifecycleManager extends
: Assignments.fromBytes(stableAssignmentsWatchEvent.value()).nodes();
return supplyAsync(() -> {
- Entry pendingAssignmentsEntry = metaStorageMgr.getLocally(pendingPartAssignmentsKey(zonePartitionId), revision);
+ Entry pendingAssignmentsEntry = metaStorageMgr.getLocally(pendingPartAssignmentsQueueKey(zonePartitionId), revision);
byte[] pendingAssignmentsFromMetaStorage = pendingAssignmentsEntry.value();
Assignments pendingAssignments = pendingAssignmentsFromMetaStorage == null
? Assignments.EMPTY
- : Assignments.fromBytes(pendingAssignmentsFromMetaStorage);
+ : AssignmentsQueue.fromBytes(pendingAssignmentsFromMetaStorage).poll();
return stopAndDestroyPartitionAndUpdateClients(
zonePartitionId,
@@ -1011,12 +1012,12 @@ public class PartitionReplicaLifecycleManager extends
return nullCompletedFuture();
}
- ZonePartitionId zonePartitionId = extractZonePartitionId(pendingAssignmentsEntry.key(), PENDING_ASSIGNMENTS_PREFIX_BYTES);
+ ZonePartitionId zonePartitionId = extractZonePartitionId(pendingAssignmentsEntry.key(), PENDING_ASSIGNMENTS_QUEUE_PREFIX_BYTES);
// Stable assignments from the meta store, which revision is bounded by the current pending event.
Assignments stableAssignments = stableAssignments(zonePartitionId, revision);
- Assignments pendingAssignments = Assignments.fromBytes(pendingAssignmentsEntry.value());
+ Assignments pendingAssignments = AssignmentsQueue.fromBytes(pendingAssignmentsEntry.value()).poll();
if (!busyLock.enterBusy()) {
return failedFuture(new NodeStoppingException());
@@ -1201,7 +1202,7 @@ public class PartitionReplicaLifecycleManager extends
+ "Initiate rebalance process for partition={}, zoneId={}",
leaderWithTerm.leader(), replicaGrpId, replicaGrpId.partitionId(), replicaGrpId.zoneId());
- return metaStorageMgr.get(pendingPartAssignmentsKey(replicaGrpId))
+ return metaStorageMgr.get(pendingPartAssignmentsQueueKey(replicaGrpId))
.thenCompose(latestPendingAssignmentsEntry -> {
// Do not change peers of the raft group if this is a stale event.
// Note that we start raft node before for the sake of the consistency in a
diff --git a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/AssignmentsTracker.java b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/AssignmentsTracker.java
index 50eef8417cc..cd78b3a69a2 100644
--- a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/AssignmentsTracker.java
+++ b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/AssignmentsTracker.java
@@ -17,7 +17,7 @@
package org.apache.ignite.internal.placementdriver;
-import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.PENDING_ASSIGNMENTS_PREFIX_BYTES;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.PENDING_ASSIGNMENTS_QUEUE_PREFIX_BYTES;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.STABLE_ASSIGNMENTS_PREFIX_BYTES;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.extractTablePartitionId;
import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.extractZonePartitionId;
@@ -32,6 +32,7 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil;
import org.apache.ignite.internal.hlc.HybridTimestamp;
@@ -46,6 +47,7 @@ import org.apache.ignite.internal.metastorage.WatchEvent;
import org.apache.ignite.internal.metastorage.WatchListener;
import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.partitiondistribution.Assignments;
+import org.apache.ignite.internal.partitiondistribution.AssignmentsQueue;
import org.apache.ignite.internal.partitiondistribution.TokenizedAssignments;
import org.apache.ignite.internal.partitiondistribution.TokenizedAssignmentsImpl;
import org.apache.ignite.internal.replicator.ReplicationGroupId;
@@ -98,12 +100,16 @@ public class AssignmentsTracker implements AssignmentsPlacementDriver {
* Restores assignments form Vault and subscribers on further updates.
*/
public void startTrack() {
- msManager.registerPrefixWatch(new ByteArray(pendingAssignmentsPrefixBytes()), pendingAssignmentsListener);
+ msManager.registerPrefixWatch(new ByteArray(pendingAssignmentsQueuePrefixBytes()), pendingAssignmentsListener);
msManager.registerPrefixWatch(new ByteArray(stableAssignmentsPrefixBytes()), stableAssignmentsListener);
msManager.recoveryFinishedFuture().thenAccept(recoveryRevisions -> {
- handleRecoveryAssignments(recoveryRevisions, pendingAssignmentsPrefixBytes(), groupPendingAssignments);
- handleRecoveryAssignments(recoveryRevisions, stableAssignmentsPrefixBytes(), groupStableAssignments);
+ handleRecoveryAssignments(recoveryRevisions, pendingAssignmentsQueuePrefixBytes(), groupPendingAssignments,
+ bytes -> AssignmentsQueue.fromBytes(bytes).poll().nodes()
+ );
+ handleRecoveryAssignments(recoveryRevisions, stableAssignmentsPrefixBytes(), groupStableAssignments,
+ bytes -> Assignments.fromBytes(bytes).nodes()
+ );
}).whenComplete((res, ex) -> {
if (ex != null) {
LOG.error("Failed to start assignment tracker due to recovery error", ex);
@@ -165,7 +171,9 @@ public class AssignmentsTracker implements AssignmentsPlacementDriver {
LOG.debug("Stable assignments update [revision={}, keys={}]", event.revision(), collectKeysFromEventAsString(event));
}
- handleReceivedAssignments(event, stableAssignmentsPrefixBytes(), groupStableAssignments);
+ handleReceivedAssignments(event, stableAssignmentsPrefixBytes(), groupStableAssignments,
+ bytes -> Assignments.fromBytes(bytes).nodes()
+ );
return nullCompletedFuture();
};
@@ -177,7 +185,9 @@ public class AssignmentsTracker implements AssignmentsPlacementDriver {
LOG.debug("Pending assignments update [revision={}, keys={}]", event.revision(), collectKeysFromEventAsString(event));
}
- handleReceivedAssignments(event, pendingAssignmentsPrefixBytes(), groupPendingAssignments);
+ handleReceivedAssignments(event, pendingAssignmentsQueuePrefixBytes(), groupPendingAssignments,
+ bytes -> AssignmentsQueue.fromBytes(bytes).poll().nodes()
+ );
return nullCompletedFuture();
};
@@ -186,7 +196,8 @@ public class AssignmentsTracker implements AssignmentsPlacementDriver {
private static void handleReceivedAssignments(
WatchEvent event,
byte[] assignmentsMetastoreKeyPrefix,
- Map<ReplicationGroupId, TokenizedAssignments> groupIdToAssignmentsMap
+ Map<ReplicationGroupId, TokenizedAssignments> groupIdToAssignmentsMap,
+ Function<byte[], Set<Assignment>> deserializer
) {
for (EntryEvent evt : event.entryEvents()) {
Entry entry = evt.newEntry();
@@ -196,7 +207,7 @@ public class AssignmentsTracker implements AssignmentsPlacementDriver {
if (entry.tombstone()) {
groupIdToAssignmentsMap.remove(grpId);
} else {
- updateGroupAssignments(groupIdToAssignmentsMap, grpId, entry);
+ updateGroupAssignments(groupIdToAssignmentsMap, grpId, entry, deserializer);
}
}
}
@@ -204,7 +215,8 @@ public class AssignmentsTracker implements AssignmentsPlacementDriver {
private void handleRecoveryAssignments(
Revisions recoveryRevisions,
byte[] assignmentsMetastoreKeyPrefix,
- Map<ReplicationGroupId, TokenizedAssignments> groupIdToAssignmentsMap
+ Map<ReplicationGroupId, TokenizedAssignments> groupIdToAssignmentsMap,
+ Function<byte[], Set<Assignment>> deserializer
) {
var prefix = new ByteArray(assignmentsMetastoreKeyPrefix);
@@ -218,7 +230,7 @@ public class AssignmentsTracker implements AssignmentsPlacementDriver {
ReplicationGroupId grpId = extractReplicationGroupPartitionId(entry.key(), assignmentsMetastoreKeyPrefix);
- updateGroupAssignments(groupIdToAssignmentsMap, grpId, entry);
+ updateGroupAssignments(groupIdToAssignmentsMap, grpId, entry, deserializer);
}
}
}
@@ -226,14 +238,15 @@ public class AssignmentsTracker implements AssignmentsPlacementDriver {
private static void updateGroupAssignments(
Map<ReplicationGroupId, TokenizedAssignments> groupIdToAssignmentsMap,
ReplicationGroupId grpId,
- Entry entry
+ Entry entry,
+ Function<byte[], Set<Assignment>> deserFunction
) {
byte[] value = entry.value();
// MetaStorage iterator should not return nulls as values.
assert value != null;
- Set<Assignment> assignmentNodes = Assignments.fromBytes(value).nodes();
+ Set<Assignment> assignmentNodes = deserFunction.apply(value);
groupIdToAssignmentsMap.put(grpId, new TokenizedAssignmentsImpl(assignmentNodes, entry.revision()));
}
@@ -328,8 +341,8 @@ public class AssignmentsTracker implements AssignmentsPlacementDriver {
return sb.toString();
}
- private static byte[] pendingAssignmentsPrefixBytes() {
- return enabledColocation() ? ZoneRebalanceUtil.PENDING_ASSIGNMENTS_PREFIX_BYTES : PENDING_ASSIGNMENTS_PREFIX_BYTES;
+ private static byte[] pendingAssignmentsQueuePrefixBytes() {
+ return enabledColocation() ? ZoneRebalanceUtil.PENDING_ASSIGNMENTS_QUEUE_PREFIX_BYTES : PENDING_ASSIGNMENTS_QUEUE_PREFIX_BYTES;
}
private static byte[] stableAssignmentsPrefixBytes() {
diff --git a/modules/placement-driver/src/test/java/org/apache/ignite/internal/placementdriver/LeaseUpdaterTest.java b/modules/placement-driver/src/test/java/org/apache/ignite/internal/placementdriver/LeaseUpdaterTest.java
index d75c8931115..47cd7af2f2c 100644
--- a/modules/placement-driver/src/test/java/org/apache/ignite/internal/placementdriver/LeaseUpdaterTest.java
+++ b/modules/placement-driver/src/test/java/org/apache/ignite/internal/placementdriver/LeaseUpdaterTest.java
@@ -21,9 +21,9 @@ import static java.util.Collections.emptyMap;
import static java.util.UUID.randomUUID;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.stream.Collectors.toSet;
-import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.PENDING_ASSIGNMENTS_PREFIX_BYTES;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.PENDING_ASSIGNMENTS_QUEUE_PREFIX_BYTES;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.STABLE_ASSIGNMENTS_PREFIX_BYTES;
-import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsQueueKey;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
import static org.apache.ignite.internal.util.ArrayUtils.BYTE_EMPTY_ARRAY;
import static org.apache.ignite.internal.util.CompletableFutures.trueCompletedFuture;
@@ -69,6 +69,7 @@ import org.apache.ignite.internal.network.ClusterService;
import org.apache.ignite.internal.network.MessagingService;
import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.partitiondistribution.Assignments;
+import org.apache.ignite.internal.partitiondistribution.AssignmentsQueue;
import org.apache.ignite.internal.placementdriver.leases.Lease;
import org.apache.ignite.internal.placementdriver.leases.LeaseBatch;
import org.apache.ignite.internal.placementdriver.leases.LeaseTracker;
@@ -132,8 +133,8 @@ public class LeaseUpdaterTest extends BaseIgniteAbstractTest {
);
Entry pendingEntry = new EntryImpl(
- pendingPartAssignmentsKey(new TablePartitionId(1, 0)).bytes(),
- Assignments.of(HybridTimestamp.MIN_VALUE.longValue(), Assignment.forPeer(pendingNode.name())).toBytes(),
+ pendingPartAssignmentsQueueKey(new TablePartitionId(1, 0)).bytes(),
+ AssignmentsQueue.toBytes(Assignments.of(HybridTimestamp.MIN_VALUE.longValue(), Assignment.forPeer(pendingNode.name()))),
1,
clock.now()
);
@@ -149,7 +150,7 @@ public class LeaseUpdaterTest extends BaseIgniteAbstractTest {
when(metaStorageManager.recoveryFinishedFuture()).thenReturn(completedFuture(new Revisions(1, -1)));
when(metaStorageManager.prefixLocally(eq(new ByteArray(STABLE_ASSIGNMENTS_PREFIX_BYTES)), anyLong()))
.thenReturn(Cursor.fromIterable(List.of(stableEntry)));
- when(metaStorageManager.prefixLocally(eq(new ByteArray(PENDING_ASSIGNMENTS_PREFIX_BYTES)), anyLong()))
+ when(metaStorageManager.prefixLocally(eq(new ByteArray(PENDING_ASSIGNMENTS_QUEUE_PREFIX_BYTES)), anyLong()))
.thenReturn(Cursor.fromIterable(List.of(pendingEntry)));
when(topologyService.logicalTopologyOnLeader()).thenReturn(completedFuture(new LogicalTopologySnapshot(1, List.of(stableNode))));
diff --git a/modules/placement-driver/src/test/java/org/apache/ignite/internal/placementdriver/PlacementDriverTest.java b/modules/placement-driver/src/test/java/org/apache/ignite/internal/placementdriver/PlacementDriverTest.java
index 1ef2659a21e..2c614a73a32 100644
--- a/modules/placement-driver/src/test/java/org/apache/ignite/internal/placementdriver/PlacementDriverTest.java
+++ b/modules/placement-driver/src/test/java/org/apache/ignite/internal/placementdriver/PlacementDriverTest.java
@@ -69,6 +69,7 @@ import org.apache.ignite.internal.network.ClusterNodeImpl;
import org.apache.ignite.internal.network.ClusterNodeResolver;
import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.partitiondistribution.Assignments;
+import org.apache.ignite.internal.partitiondistribution.AssignmentsQueue;
import org.apache.ignite.internal.partitiondistribution.TokenizedAssignments;
import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEventParameters;
import org.apache.ignite.internal.placementdriver.leases.Lease;
@@ -848,12 +849,12 @@ public class PlacementDriverTest extends BaseIgniteAbstractTest {
return expRev;
}
- private void publishAssignments(ByteArray assignmentsKey, Set<Assignment> assignments) {
+ private void publishAssignments(ByteArray assignmentsKey, byte[] assignmentsValue) {
long timestampBeforeUpdate = clockService.nowLong();
metastore.invoke(
Conditions.notExists(FAKE_KEY),
- put(assignmentsKey, Assignments.toBytes(assignments, assignmentsTimestamp)),
+ put(assignmentsKey, assignmentsValue),
noop()
);
@@ -864,15 +865,21 @@ public class PlacementDriverTest extends BaseIgniteAbstractTest {
}
private void publishStableAssignments(Set<Assignment> assignments) {
- publishAssignments(RebalanceUtil.stablePartAssignmentsKey(GROUP_1), assignments);
+ publishAssignments(RebalanceUtil.stablePartAssignmentsKey(GROUP_1),
+ Assignments.toBytes(assignments, assignmentsTimestamp)
+ );
}
private void publishPendingAssignments(Set<Assignment> assignments) {
- publishAssignments(RebalanceUtil.pendingPartAssignmentsKey(GROUP_1), assignments);
+ publishAssignments(RebalanceUtil.pendingPartAssignmentsQueueKey(GROUP_1),
+ AssignmentsQueue.toBytes(Assignments.of(assignments, assignmentsTimestamp))
+ );
}
private void publishPlannedAssignments(Set<Assignment> assignments) {
- publishAssignments(RebalanceUtil.plannedPartAssignmentsKey(GROUP_1), assignments);
+ publishAssignments(RebalanceUtil.plannedPartAssignmentsKey(GROUP_1),
+ Assignments.toBytes(assignments, assignmentsTimestamp)
+ );
}
private CompletableFuture<PrimaryReplicaEventParameters> listenAnyReplicaBecomePrimaryEvent() {
diff --git a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaImpl.java b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaImpl.java
index dacac31987c..69dcf02ab1f 100644
--- a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaImpl.java
+++ b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaImpl.java
@@ -32,7 +32,7 @@ import org.apache.ignite.internal.failure.FailureManager;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.network.NetworkMessage;
-import org.apache.ignite.internal.partitiondistribution.Assignments;
+import org.apache.ignite.internal.partitiondistribution.AssignmentsQueue;
import org.apache.ignite.internal.placementdriver.PlacementDriver;
import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEvent;
import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEventParameters;
@@ -207,7 +207,7 @@ public class ReplicaImpl implements Replica {
return nullCompletedFuture();
}
- PeersAndLearners newConfiguration = fromAssignments(Assignments.fromBytes(pendingsBytes).nodes());
+ PeersAndLearners newConfiguration = fromAssignments(AssignmentsQueue.fromBytes(pendingsBytes).poll().nodes());
LOG.info(
"New leader elected. Going to apply new configuration [tablePartitionId={}, peers={}, learners={}]",
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
index 5ddd0ef6ad6..606796ea4a9 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
@@ -28,7 +28,7 @@ import static org.apache.ignite.internal.configuration.IgnitePaths.partitionsPat
import static org.apache.ignite.internal.distributionzones.DistributionZonesTestUtil.alterZone;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.REBALANCE_SCHEDULER_POOL_SIZE;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.STABLE_ASSIGNMENTS_PREFIX;
-import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsQueueKey;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
import static org.apache.ignite.internal.network.utils.ClusterServiceTestUtils.defaultChannelTypeRegistry;
import static org.apache.ignite.internal.network.utils.ClusterServiceTestUtils.defaultSerializationRegistry;
@@ -597,7 +597,7 @@ public class ItIgniteNodeRestartTest extends BaseIgniteRestartTest {
partitionRaftConfigurer,
view -> new LocalLogStorageFactory(),
threadPoolsManager.tableIoExecutor(),
- replicaGrpId -> metaStorageMgr.get(pendingPartAssignmentsKey((TablePartitionId) replicaGrpId))
+ replicaGrpId -> metaStorageMgr.get(pendingPartAssignmentsQueueKey((TablePartitionId) replicaGrpId))
.thenApply(Entry::value)
);
diff --git a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
index 7a31c6eefbf..5997b2fc201 100644
--- a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
+++ b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
@@ -25,7 +25,7 @@ import static org.apache.ignite.internal.configuration.IgnitePaths.metastoragePa
import static org.apache.ignite.internal.configuration.IgnitePaths.partitionsPath;
import static org.apache.ignite.internal.configuration.IgnitePaths.vaultPath;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.REBALANCE_SCHEDULER_POOL_SIZE;
-import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsQueueKey;
import static org.apache.ignite.internal.thread.ThreadOperation.STORAGE_READ;
import static org.apache.ignite.internal.thread.ThreadOperation.STORAGE_WRITE;
import static org.apache.ignite.internal.util.CompletableFutures.copyStateTo;
@@ -867,7 +867,7 @@ public class IgniteImpl implements Ignite {
partitionRaftConfigurer,
volatileLogStorageFactoryCreator,
threadPoolsManager.tableIoExecutor(),
- replicaGrpId -> metaStorageMgr.get(pendingPartAssignmentsKey((TablePartitionId) replicaGrpId))
+ replicaGrpId -> metaStorageMgr.get(pendingPartAssignmentsQueueKey((TablePartitionId) replicaGrpId))
.thenApply(org.apache.ignite.internal.metastorage.Entry::value)
);
diff --git a/modules/table/src/integrationTest/java/org/apache/ignite/internal/table/distributed/disaster/AbstractHighAvailablePartitionsRecoveryTest.java b/modules/table/src/integrationTest/java/org/apache/ignite/internal/table/distributed/disaster/AbstractHighAvailablePartitionsRecoveryTest.java
index 6711b9396ea..73496f33608 100644
--- a/modules/table/src/integrationTest/java/org/apache/ignite/internal/table/distributed/disaster/AbstractHighAvailablePartitionsRecoveryTest.java
+++ b/modules/table/src/integrationTest/java/org/apache/ignite/internal/table/distributed/disaster/AbstractHighAvailablePartitionsRecoveryTest.java
@@ -26,7 +26,7 @@ import static org.apache.ignite.internal.catalog.commands.CatalogUtils.DEFAULT_F
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.PARTITION_DISTRIBUTION_RESET_TIMEOUT;
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zoneDataNodesHistoryKey;
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zoneScaleDownTimerKey;
-import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsQueueKey;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.plannedPartAssignmentsKey;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartitionAssignments;
@@ -78,6 +78,7 @@ import org.apache.ignite.internal.lang.NodeStoppingException;
import org.apache.ignite.internal.metastorage.Entry;
import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.partitiondistribution.Assignments;
+import org.apache.ignite.internal.partitiondistribution.AssignmentsQueue;
import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.schema.SchemaRegistry;
@@ -227,7 +228,7 @@ public abstract class AbstractHighAvailablePartitionsRecoveryTest extends Cluste
TablePartitionId tablePartitionId = new TablePartitionId(tableId, partNum);
ByteArray stableKey = stablePartAssignmentsKey(tablePartitionId);
- ByteArray pendingKey = pendingPartAssignmentsKey(tablePartitionId);
+ ByteArray pendingKey = pendingPartAssignmentsQueueKey(tablePartitionId);
ByteArray plannedKey = plannedPartAssignmentsKey(tablePartitionId);
Map<ByteArray, Entry> results = await(gatewayNode.metaStorageManager()
@@ -238,7 +239,7 @@ public abstract class AbstractHighAvailablePartitionsRecoveryTest extends Cluste
boolean isPlannedEmpty = results.get(plannedKey).value() == null;
stableNodes.set(assignmentsFromEntry(results.get(stableKey)));
- pendingNodes.set(assignmentsFromEntry(results.get(pendingKey)));
+ pendingNodes.set(assignmentsFromPendingEntry(results.get(pendingKey)));
plannedNodes.set(assignmentsFromEntry(results.get(plannedKey)));
return isStableAsExpected && isPendingEmpty && isPlannedEmpty;
@@ -265,6 +266,15 @@ public abstract class AbstractHighAvailablePartitionsRecoveryTest extends Cluste
: emptySet();
}
+ private static Set<String> assignmentsFromPendingEntry(Entry entry) {
+ return (entry.value() != null)
+ ? AssignmentsQueue.fromBytes(entry.value()).poll().nodes()
+ .stream()
+ .map(Assignment::consistentId)
+ .collect(Collectors.toUnmodifiableSet())
+ : emptySet();
+ }
+
private Set<Assignment> getPartitionClusterNodes(IgniteImpl node, String tableName, int partNum) {
return Optional.ofNullable(getTableId(node.catalogManager(), tableName, clock.nowLong()))
.map(tableId -> stablePartitionAssignments(node.metaStorageManager(), tableId, partNum).join())
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
index d5aca724b5c..fcaeed0d363 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
@@ -31,13 +31,13 @@ import static java.util.stream.Collectors.toList;
import static java.util.stream.Collectors.toSet;
import static org.apache.ignite.internal.causality.IncrementalVersionedValue.dependingOn;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.ASSIGNMENTS_SWITCH_REDUCE_PREFIX_BYTES;
-import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.PENDING_ASSIGNMENTS_PREFIX_BYTES;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.PENDING_ASSIGNMENTS_QUEUE_PREFIX_BYTES;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.STABLE_ASSIGNMENTS_PREFIX_BYTES;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.assignmentsChainGetLocally;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.assignmentsChainKey;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.extractTablePartitionId;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.partitionAssignmentsGetLocally;
-import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsQueueKey;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stableAssignmentsGetLocally;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.subtract;
@@ -163,6 +163,7 @@ import org.apache.ignite.internal.partition.replicator.schema.ExecutorInclinedSc
import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.partitiondistribution.Assignments;
import org.apache.ignite.internal.partitiondistribution.AssignmentsChain;
+import org.apache.ignite.internal.partitiondistribution.AssignmentsQueue;
import org.apache.ignite.internal.partitiondistribution.PartitionDistributionUtils;
import org.apache.ignite.internal.placementdriver.PlacementDriver;
import org.apache.ignite.internal.placementdriver.ReplicaMeta;
@@ -641,7 +642,10 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
processAssignmentsOnRecovery(recoveryRevision);
if (!enabledColocation()) {
- metaStorageMgr.registerPrefixWatch(new ByteArray(PENDING_ASSIGNMENTS_PREFIX_BYTES), pendingAssignmentsRebalanceListener);
+ metaStorageMgr.registerPrefixWatch(
+ new ByteArray(PENDING_ASSIGNMENTS_QUEUE_PREFIX_BYTES),
+ pendingAssignmentsRebalanceListener
+ );
metaStorageMgr.registerPrefixWatch(new ByteArray(STABLE_ASSIGNMENTS_PREFIX_BYTES), stableAssignmentsRebalanceListener);
metaStorageMgr.registerPrefixWatch(
new ByteArray(ASSIGNMENTS_SWITCH_REDUCE_PREFIX_BYTES),
@@ -960,7 +964,7 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
private void processAssignmentsOnRecovery(long recoveryRevision) {
var stableAssignmentsPrefix = new ByteArray(STABLE_ASSIGNMENTS_PREFIX_BYTES);
- var pendingAssignmentsPrefix = new ByteArray(PENDING_ASSIGNMENTS_PREFIX_BYTES);
+ var pendingAssignmentsPrefix = new ByteArray(PENDING_ASSIGNMENTS_QUEUE_PREFIX_BYTES);
startVv.update(recoveryRevision, (v, e) -> handleAssignmentsOnRecovery(
stableAssignmentsPrefix,
@@ -2221,14 +2225,14 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
return nullCompletedFuture();
}
- TablePartitionId replicaGrpId = extractTablePartitionId(pendingAssignmentsEntry.key(), PENDING_ASSIGNMENTS_PREFIX_BYTES);
+ TablePartitionId replicaGrpId = extractTablePartitionId(pendingAssignmentsEntry.key(), PENDING_ASSIGNMENTS_QUEUE_PREFIX_BYTES);
// Stable assignments from the meta store, which revision is bounded by the current pending event.
Assignments stableAssignments = stableAssignmentsGetLocally(metaStorageMgr, replicaGrpId, revision);
AssignmentsChain assignmentsChain = assignmentsChainGetLocally(metaStorageMgr, replicaGrpId, revision);
- Assignments pendingAssignments = Assignments.fromBytes(pendingAssignmentsEntry.value());
+ Assignments pendingAssignments = AssignmentsQueue.fromBytes(pendingAssignmentsEntry.value()).poll();
return tablesVv.get(revision)
.thenApply(ignore -> {
@@ -2470,7 +2474,7 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
Assignments pendingAssignments,
long currentRevision
) {
- metaStorageMgr.get(pendingPartAssignmentsKey(replicationGroupId)).thenAccept(latestPendingAssignmentsEntry -> {
+ metaStorageMgr.get(pendingPartAssignmentsQueueKey(replicationGroupId)).thenAccept(latestPendingAssignmentsEntry -> {
// Do not change peers of the raft group if this is a stale event.
// Note that we start raft node before for the sake of the consistency in a
// starting and stopping raft nodes.
@@ -2703,13 +2707,13 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
: Assignments.fromBytes(stableAssignmentsWatchEvent.value()).nodes();
return supplyAsync(() -> {
- Entry pendingAssignmentsEntry = metaStorageMgr.getLocally(pendingPartAssignmentsKey(tablePartitionId), revision);
+ Entry pendingAssignmentsEntry = metaStorageMgr.getLocally(pendingPartAssignmentsQueueKey(tablePartitionId), revision);
byte[] pendingAssignmentsFromMetaStorage = pendingAssignmentsEntry.value();
Assignments pendingAssignments = pendingAssignmentsFromMetaStorage == null
? Assignments.EMPTY
- : Assignments.fromBytes(pendingAssignmentsFromMetaStorage);
+ : AssignmentsQueue.fromBytes(pendingAssignmentsFromMetaStorage).poll();
if (LOG.isInfoEnabled()) {
var stringKey = new String(stableAssignmentsWatchEvent.key(), UTF_8);
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/GroupUpdateRequest.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/GroupUpdateRequest.java
index b3afe4904aa..c730ce8ca01 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/GroupUpdateRequest.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/GroupUpdateRequest.java
@@ -26,7 +26,7 @@ import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUt
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.UpdateStatus.OUTDATED_UPDATE_RECEIVED;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.UpdateStatus.PENDING_KEY_UPDATED;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingChangeTriggerKey;
-import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsQueueKey;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.plannedPartAssignmentsKey;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.tableStableAssignments;
import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
@@ -38,6 +38,7 @@ import static org.apache.ignite.internal.metastorage.dsl.Statements.iif;
import static org.apache.ignite.internal.partition.replicator.network.disaster.LocalPartitionStateEnum.CATCHING_UP;
import static org.apache.ignite.internal.partition.replicator.network.disaster.LocalPartitionStateEnum.HEALTHY;
import static org.apache.ignite.internal.partitiondistribution.PartitionDistributionUtils.calculateAssignmentForPartition;
+import static org.apache.ignite.internal.partitiondistribution.PendingAssignmentsCalculator.pendingAssignmentsCalculator;
import static org.apache.ignite.internal.table.distributed.disaster.DisasterRecoveryRequestType.SINGLE_NODE;
import static org.apache.ignite.internal.util.ByteUtils.longToBytesKeepingOrder;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
@@ -338,7 +339,10 @@ class GroupUpdateRequest implements DisasterRecoveryRequest {
Iif invokeClosure = prepareMsInvokeClosure(
partId,
longToBytesKeepingOrder(revision),
- Assignments.forced(Set.of(nextAssignment), assignmentsTimestamp).toBytes(),
+ pendingAssignmentsCalculator()
+ .stable(Assignments.of(currentAssignments, assignmentsTimestamp))
+ .target(Assignments.forced(Set.of(nextAssignment), assignmentsTimestamp))
+ .toQueue().toBytes(),
// If planned nodes set consists of reset node assignment only then we shouldn't schedule the same planned rebalance.
isProposedPendingEqualsProposedPlanned
? null
@@ -403,16 +407,16 @@ class GroupUpdateRequest implements DisasterRecoveryRequest {
* <ul>
* <li>Guards the condition with a standard {@link RebalanceUtil#pendingChangeTriggerKey(TablePartitionId)} check.</li>
* <li>Adds additional guard with comparison of real and proposed values of
- * {@link RebalanceUtil#pendingPartAssignmentsKey(TablePartitionId)}, just in case.</li>
+ * {@link RebalanceUtil#pendingPartAssignmentsQueueKey(TablePartitionId)}, just in case.</li>
* <li>Updates the value of {@link RebalanceUtil#pendingChangeTriggerKey(TablePartitionId)}.</li>
- * <li>Updates the value of {@link RebalanceUtil#pendingPartAssignmentsKey(TablePartitionId)}.</li>
+ * <li>Updates the value of {@link RebalanceUtil#pendingPartAssignmentsQueueKey(TablePartitionId)}.</li>
* <li>Updates the value of {@link RebalanceUtil#plannedPartAssignmentsKey(TablePartitionId)} or removes it, if
* {@code plannedAssignmentsBytes} is {@code null}.</li>
* </ul>
*
* @param partId Partition ID.
* @param revisionBytes Properly serialized current meta-storage revision.
- * @param pendingAssignmentsBytes Value for {@link RebalanceUtil#pendingPartAssignmentsKey(TablePartitionId)}.
+ * @param pendingAssignmentsBytes Value for {@link RebalanceUtil#pendingPartAssignmentsQueueKey(TablePartitionId)}.
* @param plannedAssignmentsBytes Value for {@link RebalanceUtil#plannedPartAssignmentsKey(TablePartitionId)} or {@code null}.
* @return {@link Iif} instance.
*/
@@ -423,7 +427,7 @@ class GroupUpdateRequest implements DisasterRecoveryRequest {
byte @Nullable [] plannedAssignmentsBytes
) {
ByteArray pendingChangeTriggerKey = pendingChangeTriggerKey(partId);
- ByteArray partAssignmentsPendingKey = pendingPartAssignmentsKey(partId);
+ ByteArray partAssignmentsPendingKey = pendingPartAssignmentsQueueKey(partId);
ByteArray partAssignmentsPlannedKey = plannedPartAssignmentsKey(partId);
return iif(notExists(pendingChangeTriggerKey).or(value(pendingChangeTriggerKey).lt(revisionBytes)),
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/utils/RebalanceUtilEx.java b/modules/table/src/main/java/org/apache/ignite/internal/utils/RebalanceUtilEx.java
index 1a3339f7703..aa3c321407e 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/utils/RebalanceUtilEx.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/utils/RebalanceUtilEx.java
@@ -18,7 +18,7 @@
package org.apache.ignite.internal.utils;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingChangeTriggerKey;
-import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsQueueKey;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.switchReduceKey;
import static org.apache.ignite.internal.metastorage.dsl.Conditions.and;
@@ -46,6 +46,7 @@ import org.apache.ignite.internal.metastorage.dsl.Iif;
import org.apache.ignite.internal.metastorage.dsl.Operations;
import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.partitiondistribution.Assignments;
+import org.apache.ignite.internal.partitiondistribution.AssignmentsQueue;
import org.apache.ignite.internal.replicator.TablePartitionId;
/**
@@ -138,11 +139,11 @@ public class RebalanceUtilEx {
Set<Assignment> assignments = calculateAssignmentForPartition(dataNodes, partId.partitionId(), partitions, replicas);
- ByteArray pendingKey = pendingPartAssignmentsKey(partId);
+ ByteArray pendingKey = pendingPartAssignmentsQueueKey(partId);
Set<Assignment> pendingAssignments = difference(assignments, switchReduce.nodes());
- byte[] pendingByteArray = Assignments.toBytes(pendingAssignments, assignmentsTimestamp);
+ byte[] pendingByteArray = AssignmentsQueue.toBytes(Assignments.of(pendingAssignments, assignmentsTimestamp));
byte[] assignmentsByteArray = Assignments.toBytes(assignments, assignmentsTimestamp);
ByteArray changeTriggerKey = pendingChangeTriggerKey(partId);
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryMsInvokeTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryMsInvokeTest.java
index 56649edce2b..4f93a4c81e1 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryMsInvokeTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryMsInvokeTest.java
@@ -19,7 +19,7 @@ package org.apache.ignite.internal.table.distributed.disaster;
import static java.util.stream.Collectors.toSet;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingChangeTriggerKey;
-import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsQueueKey;
import static org.apache.ignite.internal.partitiondistribution.PartitionDistributionUtils.calculateAssignmentForPartition;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.apache.ignite.internal.util.ByteUtils.bytesToLongKeepingOrder;
@@ -41,6 +41,7 @@ import org.apache.ignite.internal.metastorage.impl.MetaStorageManagerImpl;
import org.apache.ignite.internal.metastorage.impl.StandaloneMetaStorageManager;
import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.partitiondistribution.Assignments;
+import org.apache.ignite.internal.partitiondistribution.AssignmentsQueue;
import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
import org.junit.jupiter.api.BeforeEach;
@@ -96,8 +97,8 @@ public class DisasterRecoveryMsInvokeTest extends BaseIgniteAbstractTest {
if (currentPending != null) {
assertThat(
metaStorageManager.put(
- pendingPartAssignmentsKey(tablePartitionId),
- Assignments.toBytes(currentPending, assignmentsTimestamp)
+ pendingPartAssignmentsQueueKey(tablePartitionId),
+ AssignmentsQueue.toBytes(Assignments.of(currentPending, assignmentsTimestamp))
),
willCompleteSuccessfully()
);
@@ -108,7 +109,7 @@ public class DisasterRecoveryMsInvokeTest extends BaseIgniteAbstractTest {
GroupUpdateRequest.prepareMsInvokeClosure(
tablePartitionId,
longToBytesKeepingOrder(expectedPendingChangeTriggerKey),
- Assignments.toBytes(pending, assignmentsTimestamp),
+ AssignmentsQueue.toBytes(Assignments.of(pending, assignmentsTimestamp)),
null
)
),
diff --git a/modules/table/tech-notes/rebalance.md b/modules/table/tech-notes/rebalance.md
index 5474c131d62..450ea83c724 100644
--- a/modules/table/tech-notes/rebalance.md
+++ b/modules/table/tech-notes/rebalance.md
@@ -20,7 +20,7 @@ Every algorithm phase has the following main sections:
## New metastore keys
For further steps, we should introduce some new metastore keys:
- `partition.assignments.stable` - the list of peers, which process operations for a partition at the current moment.
-- `partition.assignments.pending` - the list of peers, where current rebalance move the partition.
+- `partition.assignments.pending` - the queue of lists of peers, where current rebalance move the partition. Queue is needed in cases when multiple configuration switches during one rebalance are required.
- `partition.assignments.planned` - the list of peers, which will be used for new rebalance, when current will be finished.
Also, we will need the utility key:
@@ -50,10 +50,10 @@ onReplicaNumberChange:
metastoreInvoke: // atomic metastore call through multi-invoke api
if empty(partition.change.trigger.revision) || partition.change.trigger.revision < event.revision:
if empty(partition.assignments.pending) && partition.assignments.stable != calcPartAssighments():
- partition.assignments.pending = calcPartAssignments()
+ partition.assignments.pending = partAssignmentsPendingQueue
partition.change.trigger.revision = event.revision
else:
- if partition.assignments.pending != calcPartAssignments
+ if partition.assignments.pending != partAssignmentsPendingQueue
partition.assignments.planned = calcPartAssignments()
partition.change.trigger.revision = event.revision
else
@@ -146,13 +146,13 @@ metastoreInvoke: // atomic metastore call through multi-invoke api
if empty(partition.change.trigger.revision) || partition.change.trigger.revision < event.revision:
if empty(partition.assignments.pending):
if partition.assignments.stable != calcPartAssignments():
- partition.assignments.pending = calcPartAssignments()
+ partition.assignments.pending = partAssignmentsPendingQueue
partition.change.trigger.revision = event.revision
else:
skip
else:
if empty(partition.assignments.pending.lock):
- partition.assignments.pending = calcPartAssignments()
+ partition.assignments.pending = partAssignmentsPendingQueue
partition.change.trigger.revision = event.revision
else:
partition.assignments.planned = calcPartAssignments()
diff --git a/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryReconfigurationTest.java b/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryReconfigurationTest.java
index 3235b2343f9..6ffc00bc2b6 100644
--- a/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryReconfigurationTest.java
+++ b/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryReconfigurationTest.java
@@ -30,7 +30,7 @@ import static org.apache.ignite.internal.catalog.CatalogService.DEFAULT_STORAGE_
import static org.apache.ignite.internal.catalog.commands.CatalogUtils.INFINITE_TIMER_VALUE;
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.PARTITION_DISTRIBUTION_RESET_TIMEOUT;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.assignmentsChainKey;
-import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsQueueKey;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.plannedPartAssignmentsKey;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
import static org.apache.ignite.internal.replicator.configuration.ReplicationConfigurationSchema.DEFAULT_IDLE_SAFE_TIME_PROP_DURATION;
@@ -95,6 +95,7 @@ import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.partitiondistribution.Assignments;
import org.apache.ignite.internal.partitiondistribution.AssignmentsChain;
import org.apache.ignite.internal.partitiondistribution.AssignmentsLink;
+import org.apache.ignite.internal.partitiondistribution.AssignmentsQueue;
import org.apache.ignite.internal.partitiondistribution.RendezvousDistributionFunction;
import org.apache.ignite.internal.placementdriver.ReplicaMeta;
import org.apache.ignite.internal.raft.Peer;
@@ -2004,13 +2005,13 @@ public class ItDisasterRecoveryReconfigurationTest extends ClusterPerTestIntegra
private @Nullable Assignments getPendingAssignments(IgniteImpl node, int partId) {
CompletableFuture<Entry> pendingFut = node.metaStorageManager()
- .get(pendingPartAssignmentsKey(new TablePartitionId(tableId, partId)));
+ .get(pendingPartAssignmentsQueueKey(new TablePartitionId(tableId, partId)));
assertThat(pendingFut, willCompleteSuccessfully());
Entry pending = pendingFut.join();
- return pending.empty() ? null : Assignments.fromBytes(pending.value());
+ return pending.empty() ? null : AssignmentsQueue.fromBytes(pending.value()).poll();
}
private @Nullable Assignments getStableAssignments(IgniteImpl node, int partId) {