You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ib...@apache.org on 2024/06/21 09:23:44 UTC
(ignite-3) branch main updated: IGNITE-21661 Add test that reset partition can use data from a node outside of stable assignments (#3945)
This is an automated email from the ASF dual-hosted git repository.
ibessonov 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 d34486b099 IGNITE-21661 Add test that reset partition can use data from a node outside of stable assignments (#3945)
d34486b099 is described below
commit d34486b099d8df0d17416e153ec71cdb087e3aed
Author: Ivan Bessonov <be...@gmail.com>
AuthorDate: Fri Jun 21 12:23:39 2024 +0300
IGNITE-21661 Add test that reset partition can use data from a node outside of stable assignments (#3945)
---
.../ignite/internal/affinity/Assignments.java | 20 ++
.../java/org/apache/ignite/lang/ErrorGroups.java | 3 +
.../distributionzones/rebalance/RebalanceUtil.java | 145 +-------------
...istributionZoneManagerScaleUpScaleDownTest.java | 5 +-
.../DistributionZonesTestUtil.java | 2 +-
.../impl/ItIdempotentCommandCacheTest.java | 2 +-
.../internal/network/DefaultMessagingService.java | 2 +-
modules/platforms/cpp/ignite/common/error_codes.h | 3 +-
modules/platforms/cpp/ignite/odbc/common_types.cpp | 1 +
.../platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs | 3 +
.../org/apache/ignite/internal/app/IgniteImpl.java | 2 +-
.../ItDisasterRecoveryReconfigurationTest.java | 211 +++++++++++++++++++-
.../internal/table/distributed/TableManager.java | 1 -
.../disaster/DisasterRecoveryManager.java | 29 +--
.../distributed/disaster/GlobalPartitionState.java | 12 ++
.../distributed/disaster/LocalPartitionState.java | 11 +
.../disaster/LocalPartitionStateByNode.java | 9 +-
.../disaster/LocalPartitionStateMessageByNode.java | 5 +
.../disaster/ManualGroupUpdateRequest.java | 221 ++++++++++++++++++++-
.../exceptions/DisasterRecoveryException.java | 6 +-
20 files changed, 521 insertions(+), 172 deletions(-)
diff --git a/modules/affinity/src/main/java/org/apache/ignite/internal/affinity/Assignments.java b/modules/affinity/src/main/java/org/apache/ignite/internal/affinity/Assignments.java
index 2f0d7fcbee..6115122561 100644
--- a/modules/affinity/src/main/java/org/apache/ignite/internal/affinity/Assignments.java
+++ b/modules/affinity/src/main/java/org/apache/ignite/internal/affinity/Assignments.java
@@ -153,6 +153,26 @@ public class Assignments implements Serializable {
return S.toString(this);
}
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ Assignments that = (Assignments) o;
+ return force == that.force && nodes.equals(that.nodes);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = nodes.hashCode();
+ result = 31 * result + Boolean.hashCode(force);
+ return result;
+ }
+
/**
* Creates a string representation of the given assignments list for logging usage purpose mostly.
*
diff --git a/modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java b/modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java
index 960f6e7821..232c8121db 100755
--- a/modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java
+++ b/modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java
@@ -623,5 +623,8 @@ public class ErrorGroups {
/** Error while returning partition states. */
public static final int PARTITION_STATE_ERR = RECOVERY_ERR_GROUP.registerErrorCode((short) 3);
+
+ /** Error while returning partition states. */
+ public static final int CLUSTER_NOT_IDLE_ERR = RECOVERY_ERR_GROUP.registerErrorCode((short) 4);
}
}
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 edaef4609a..bc40b6caa3 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
@@ -58,10 +58,8 @@ import org.apache.ignite.internal.metastorage.Entry;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.metastorage.dsl.Condition;
import org.apache.ignite.internal.metastorage.dsl.Iif;
-import org.apache.ignite.internal.metastorage.dsl.StatementResult;
import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.replicator.ZonePartitionId;
-import org.apache.ignite.internal.util.CollectionUtils;
import org.jetbrains.annotations.Nullable;
/**
@@ -74,8 +72,7 @@ public class RebalanceUtil {
/**
* Status values for methods like
- * {@link #updatePendingAssignmentsKeys(CatalogTableDescriptor, TablePartitionId, Collection, int, long, MetaStorageManager, int, Set)}
- * or {@link #manualPartitionUpdate(TablePartitionId, Collection, Set, int, long, MetaStorageManager, Set)}.
+ * {@link #updatePendingAssignmentsKeys(CatalogTableDescriptor, TablePartitionId, Collection, int, long, MetaStorageManager, int, Set)}.
*/
public enum UpdateStatus {
/**
@@ -314,126 +311,6 @@ public class RebalanceUtil {
return futures;
}
-
- /**
- * Sets force assignments for the zone/table if it's required. The condition for force reassignment is the absence of stable
- * assignments' majority within the set of currently alive nodes. In this case we calculate new assignments that include all alive
- * stable nodes, and try to save ot with a {@link Assignments#force()} flag enabled.
- *
- * @param tableDescriptor Table descriptor.
- * @param zoneDescriptor Zone descriptor.
- * @param partitionIds Partitions IDs to force assignments for. If empty, reassigns all zone's partitions.
- * @param dataNodes Current DZ data nodes.
- * @param aliveNodesConsistentIds Set of alive nodes according to logical topology.
- * @param revision Meta-storage revision to be associated with reassignment.
- * @param metaStorageManager Meta-storage manager.
- * @return A future that will be completed when reassignments data is written into a meta-storage, if that's required.
- */
- public static CompletableFuture<?>[] forceAssignmentsUpdate(
- CatalogTableDescriptor tableDescriptor,
- CatalogZoneDescriptor zoneDescriptor,
- Set<Integer> partitionIds,
- Set<String> dataNodes,
- Set<String> aliveNodesConsistentIds,
- long revision,
- MetaStorageManager metaStorageManager
- ) {
- CompletableFuture<Map<Integer, Assignments>> tableAssignmentsFut = tableAssignments(
- metaStorageManager,
- tableDescriptor.id(),
- partitionIds,
- zoneDescriptor.partitions()
- );
-
- Set<String> aliveDataNodes = CollectionUtils.intersect(dataNodes, aliveNodesConsistentIds);
-
- int[] ids = partitionIds.isEmpty()
- ? IntStream.range(0, zoneDescriptor.partitions()).toArray()
- : partitionIds.stream().mapToInt(Integer::intValue).toArray();
-
- CompletableFuture<?>[] futures = new CompletableFuture[ids.length];
-
- for (int i = 0; i < ids.length; i++) {
- TablePartitionId replicaGrpId = new TablePartitionId(tableDescriptor.id(), ids[i]);
-
- futures[i] = tableAssignmentsFut.thenCompose(tableAssignments ->
- tableAssignments.isEmpty() ? nullCompletedFuture() : manualPartitionUpdate(
- replicaGrpId,
- aliveDataNodes,
- aliveNodesConsistentIds,
- zoneDescriptor.replicas(),
- revision,
- metaStorageManager,
- tableAssignments.get(replicaGrpId.partitionId()).nodes()
- )).thenAccept(res -> {
- LOG.info("Partition {} returned {} status on reset attempt", replicaGrpId, UpdateStatus.valueOf(res));
- }
- );
- }
-
- return futures;
- }
-
- private static CompletableFuture<Integer> manualPartitionUpdate(
- TablePartitionId partId,
- Collection<String> aliveDataNodes,
- Set<String> aliveNodesConsistentIds,
- int replicas,
- long revision,
- MetaStorageManager metaStorageMgr,
- Set<Assignment> currentAssignments
- ) {
- // TODO https://issues.apache.org/jira/browse/IGNITE-21303
- // This is a naive approach that doesn't exclude nodes in error state, if they exist.
- Set<Assignment> partAssignments = new HashSet<>();
- for (Assignment currentAssignment : currentAssignments) {
- if (aliveNodesConsistentIds.contains(currentAssignment.consistentId())) {
- partAssignments.add(currentAssignment);
- }
- }
-
- if (partAssignments.size() >= (replicas / 2 + 1)) {
- return CompletableFuture.completedFuture(ASSIGNMENT_NOT_UPDATED.ordinal());
- }
-
- Set<Assignment> calcAssignments = AffinityUtils.calculateAssignmentForPartition(aliveDataNodes, partId.partitionId(), replicas);
-
- for (Assignment calcAssignment : calcAssignments) {
- if (partAssignments.size() == replicas) {
- break;
- }
-
- partAssignments.add(calcAssignment);
- }
-
- if (currentAssignments.equals(calcAssignments)) {
- return CompletableFuture.completedFuture(ASSIGNMENT_NOT_UPDATED.ordinal());
- }
-
- byte[] partAssignmentsBytes = Assignments.forced(partAssignments).toBytes();
- byte[] revisionBytes = longToBytesKeepingOrder(revision);
-
- ByteArray partChangeTriggerKey = pendingChangeTriggerKey(partId);
- ByteArray partAssignmentsPendingKey = pendingPartAssignmentsKey(partId);
- ByteArray partAssignmentsPlannedKey = plannedPartAssignmentsKey(partId);
-
- Iif iif = iif(
- notExists(partChangeTriggerKey).or(value(partChangeTriggerKey).lt(revisionBytes)),
- iif(
- value(partAssignmentsPendingKey).ne(partAssignmentsBytes),
- ops(
- put(partAssignmentsPendingKey, partAssignmentsBytes),
- put(partChangeTriggerKey, revisionBytes),
- remove(partAssignmentsPlannedKey)
- ).yield(PENDING_KEY_UPDATED.ordinal()),
- ops().yield(ASSIGNMENT_NOT_UPDATED.ordinal())
- ),
- ops().yield(OUTDATED_UPDATE_RECEIVED.ordinal())
- );
-
- return metaStorageMgr.invoke(iif).thenApply(StatementResult::getAsInt);
- }
-
/** Key prefix for pending assignments. */
public static final String PENDING_ASSIGNMENTS_PREFIX = "assignments.pending.";
@@ -740,21 +617,21 @@ public class RebalanceUtil {
* @param numberOfPartitions Number of partitions. Ignored if partition IDs are specified.
* @return Future with table assignments as a value.
*/
- static CompletableFuture<Map<Integer, Assignments>> tableAssignments(
+ public static CompletableFuture<Map<Integer, Assignments>> tableAssignments(
MetaStorageManager metaStorageManager,
int tableId,
Set<Integer> partitionIds,
int numberOfPartitions
) {
- Map<ByteArray, Integer> partitionKeysToPartitionNumber = new HashMap<>();
-
- Collection<Integer> ids = partitionIds.isEmpty()
- ? IntStream.range(0, numberOfPartitions).boxed().collect(toList())
- : partitionIds;
-
- for (Integer partId : ids) {
- partitionKeysToPartitionNumber.put(stablePartAssignmentsKey(new TablePartitionId(tableId, partId)), partId);
- }
+ IntStream partitionIdsStream = partitionIds.isEmpty()
+ ? IntStream.range(0, numberOfPartitions)
+ : partitionIds.stream().mapToInt(Integer::intValue);
+
+ Map<ByteArray, Integer> partitionKeysToPartitionNumber = partitionIdsStream.collect(
+ HashMap::new,
+ (map, partId) -> map.put(stablePartAssignmentsKey(new TablePartitionId(tableId, partId)), partId),
+ Map::putAll
+ );
return metaStorageManager.getAll(partitionKeysToPartitionNumber.keySet())
.thenApply(entries -> {
diff --git a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerScaleUpScaleDownTest.java b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerScaleUpScaleDownTest.java
index d445dc22e5..0633bbe2ca 100644
--- a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerScaleUpScaleDownTest.java
+++ b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerScaleUpScaleDownTest.java
@@ -29,7 +29,6 @@ import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil
import static org.apache.ignite.internal.distributionzones.DistributionZonesUtil.zoneScaleUpChangeTriggerKey;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
import static org.apache.ignite.internal.util.ByteUtils.bytesToLongKeepingOrder;
-import static org.apache.ignite.internal.util.ByteUtils.longToBytes;
import static org.apache.ignite.internal.util.ByteUtils.longToBytesKeepingOrder;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -498,9 +497,11 @@ public class DistributionZoneManagerScaleUpScaleDownTest extends BaseDistributio
byte[] keyDataNodes = zoneDataNodesKey(zoneId).bytes();
if (Arrays.stream(iif.cond().keys()).anyMatch(k -> Arrays.equals(keyScaleDown, k))) {
+ byte[] scaleDownRevision = longToBytesKeepingOrder(revisionOfScaleDown + 100);
+
keyValueStorage.putAll(
List.of(keyScaleDown, keyDataNodes),
- List.of(longToBytes(revisionOfScaleDown + 100), keyValueStorage.get(zoneDataNodesKey(zoneId).bytes()).value()),
+ List.of(scaleDownRevision, keyValueStorage.get(zoneDataNodesKey(zoneId).bytes()).value()),
HybridTimestamp.MIN_VALUE
);
}
diff --git a/modules/distribution-zones/src/testFixtures/java/org/apache/ignite/internal/distributionzones/DistributionZonesTestUtil.java b/modules/distribution-zones/src/testFixtures/java/org/apache/ignite/internal/distributionzones/DistributionZonesTestUtil.java
index b11b730003..5ae827b891 100644
--- a/modules/distribution-zones/src/testFixtures/java/org/apache/ignite/internal/distributionzones/DistributionZonesTestUtil.java
+++ b/modules/distribution-zones/src/testFixtures/java/org/apache/ignite/internal/distributionzones/DistributionZonesTestUtil.java
@@ -281,7 +281,7 @@ public class DistributionZonesTestUtil {
assertValueInStorage(
keyValueStorage,
zoneScaleDownChangeTriggerKey(zoneId).bytes(),
- ByteUtils::bytesToLong,
+ ByteUtils::bytesToLongKeepingOrder,
revision,
2000
);
diff --git a/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItIdempotentCommandCacheTest.java b/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItIdempotentCommandCacheTest.java
index 453a80eca4..69bedbc2ca 100644
--- a/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItIdempotentCommandCacheTest.java
+++ b/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItIdempotentCommandCacheTest.java
@@ -244,7 +244,7 @@ public class ItIdempotentCommandCacheTest extends IgniteAbstractTest {
}
void stopDroppingMessages() {
- ((DefaultMessagingService) clusterService.messagingService()).dropMessages(null);
+ ((DefaultMessagingService) clusterService.messagingService()).stopDroppingMessages();
}
boolean checkValueInStorage(byte[] testKey, byte[] testValueExpected) {
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/DefaultMessagingService.java b/modules/network/src/main/java/org/apache/ignite/internal/network/DefaultMessagingService.java
index 648ef356d7..60fc9f0915 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/DefaultMessagingService.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/DefaultMessagingService.java
@@ -619,7 +619,7 @@ public class DefaultMessagingService extends AbstractMessagingService {
* node's consistent ID.
*/
@TestOnly
- public void dropMessages(BiPredicate<String, NetworkMessage> predicate) {
+ public void dropMessages(BiPredicate<@Nullable String, NetworkMessage> predicate) {
dropMessagesPredicate = predicate;
}
diff --git a/modules/platforms/cpp/ignite/common/error_codes.h b/modules/platforms/cpp/ignite/common/error_codes.h
index 10ff90e3fc..e1a6417e35 100644
--- a/modules/platforms/cpp/ignite/common/error_codes.h
+++ b/modules/platforms/cpp/ignite/common/error_codes.h
@@ -210,7 +210,8 @@ enum class code : underlying_t {
// DisasterRecovery group. Group code: 20
ILLEGAL_PARTITION_ID = 0x140001,
NODES_NOT_FOUND = 0x140002,
- PARTITION_STATE = 0x140003
+ PARTITION_STATE = 0x140003,
+ CLUSTER_NOT_IDLE = 0x140004
};
} // namespace error
diff --git a/modules/platforms/cpp/ignite/odbc/common_types.cpp b/modules/platforms/cpp/ignite/odbc/common_types.cpp
index e76d18a938..e889333d25 100644
--- a/modules/platforms/cpp/ignite/odbc/common_types.cpp
+++ b/modules/platforms/cpp/ignite/odbc/common_types.cpp
@@ -302,6 +302,7 @@ sql_state error_code_to_sql_state(error::code code) {
case error::code::NODES_NOT_FOUND:
case error::code::ILLEGAL_PARTITION_ID:
case error::code::PARTITION_STATE:
+ case error::code::CLUSTER_NOT_IDLE:
return sql_state::SHY000_GENERAL_ERROR;
}
diff --git a/modules/platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs b/modules/platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs
index ea86d3885d..359fa61170 100644
--- a/modules/platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs
+++ b/modules/platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs
@@ -590,6 +590,9 @@ namespace Apache.Ignite
/// <summary> PartitionState error. </summary>
public const int PartitionState = (GroupCode << 16) | (3 & 0xFFFF);
+
+ /// <summary> ClusterNotIdle error. </summary>
+ public const int ClusterNotIdle = (GroupCode << 16) | (4 & 0xFFFF);
}
}
}
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 fe7fbb0d1a..868b1fb0ee 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
@@ -1564,7 +1564,7 @@ public class IgniteImpl implements Ignite {
// TODO: IGNITE-18493 - remove/move this
@TestOnly
- public void dropMessages(BiPredicate<String, NetworkMessage> predicate) {
+ public void dropMessages(BiPredicate<@Nullable String, NetworkMessage> predicate) {
((DefaultMessagingService) clusterSvc.messagingService()).dropMessages(predicate);
}
diff --git a/modules/table/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryReconfigurationTest.java b/modules/table/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryReconfigurationTest.java
index 15655475a0..6086409188 100644
--- a/modules/table/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryReconfigurationTest.java
+++ b/modules/table/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryReconfigurationTest.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.disaster;
import static java.lang.String.format;
+import static java.util.Collections.emptySet;
import static java.util.Map.of;
import static java.util.Objects.requireNonNull;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
@@ -25,6 +26,7 @@ import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.ignite.internal.TestWrappers.unwrapTableImpl;
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.raftConfigurationAppliedKey;
import static org.apache.ignite.internal.replicator.configuration.ReplicationConfigurationSchema.DEFAULT_IDLE_SAFE_TIME_PROP_DURATION;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.assertThrows;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.runRace;
@@ -33,8 +35,11 @@ import static org.apache.ignite.internal.testframework.matchers.CompletableFutur
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willSucceedIn;
import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.anEmptyMap;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.not;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -44,38 +49,65 @@ import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.reflect.Method;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
+import java.util.function.BiPredicate;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.apache.ignite.internal.ClusterPerTestIntegrationTest;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.affinity.Assignments;
import org.apache.ignite.internal.affinity.RendezvousAffinityFunction;
import org.apache.ignite.internal.app.IgniteImpl;
import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
+import org.apache.ignite.internal.lang.ByteArray;
import org.apache.ignite.internal.lang.RunnableX;
+import org.apache.ignite.internal.metastorage.command.MultiInvokeCommand;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.metastorage.dsl.OperationType;
+import org.apache.ignite.internal.metastorage.dsl.Statement;
+import org.apache.ignite.internal.metastorage.dsl.Statement.UpdateStatement;
+import org.apache.ignite.internal.network.NetworkMessage;
+import org.apache.ignite.internal.partition.replicator.network.disaster.LocalPartitionStateEnum;
+import org.apache.ignite.internal.partition.replicator.network.raft.SnapshotMvDataResponse;
import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.raft.Peer;
+import org.apache.ignite.internal.raft.RaftNodeId;
+import org.apache.ignite.internal.raft.WriteCommand;
+import org.apache.ignite.internal.raft.server.impl.JraftServerImpl;
import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.table.TableViewInternal;
import org.apache.ignite.internal.table.distributed.TableManager;
+import org.apache.ignite.internal.table.distributed.disaster.GlobalPartitionState;
+import org.apache.ignite.internal.table.distributed.disaster.GlobalPartitionStateEnum;
import org.apache.ignite.internal.table.distributed.disaster.LocalPartitionStateByNode;
-import org.apache.ignite.internal.testframework.IgniteTestUtils;
+import org.apache.ignite.internal.util.ByteUtils;
import org.apache.ignite.lang.IgniteException;
+import org.apache.ignite.raft.jraft.RaftGroupService;
+import org.apache.ignite.raft.jraft.Status;
+import org.apache.ignite.raft.jraft.error.RaftError;
+import org.apache.ignite.raft.jraft.rpc.WriteActionRequest;
import org.apache.ignite.table.KeyValueView;
import org.apache.ignite.table.Table;
import org.apache.ignite.table.Tuple;
import org.apache.ignite.tx.TransactionException;
-import org.hamcrest.Matchers;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.api.Timeout;
/**
- * Tests for scenarios where majority of peers is not available.
+ * Tests for scenarios where majority of peers is not available. In this class we frequently assert partition distributions, this means that
+ * tests code implicitly depends on test names, because we use method names to generate node names, and we use node names to assign
+ * partitions.
*/
+@Timeout(120)
public class ItDisasterRecoveryReconfigurationTest extends ClusterPerTestIntegrationTest {
/** Scale-down timeout. */
private static final int SCALE_DOWN_TIMEOUT_SECONDS = 2;
@@ -239,10 +271,10 @@ public class ItDisasterRecoveryReconfigurationTest extends ClusterPerTestIntegra
// Should fail because majority was lost.
List<Throwable> fixingPartErrorsBeforeReset = insertValues(table, fixingPartId, 0);
- assertThat(fixingPartErrorsBeforeReset, Matchers.not(empty()));
+ assertThat(fixingPartErrorsBeforeReset, not(empty()));
List<Throwable> anotherPartErrorsBeforeReset = insertValues(table, anotherPartId, 0);
- assertThat(anotherPartErrorsBeforeReset, Matchers.not(empty()));
+ assertThat(anotherPartErrorsBeforeReset, not(empty()));
CompletableFuture<?> updateFuture = node0.disasterRecoveryManager().resetPartitions(
zoneName,
@@ -260,7 +292,7 @@ public class ItDisasterRecoveryReconfigurationTest extends ClusterPerTestIntegra
// Was not specified in reset, shouldn't be fixed. */
List<Throwable> anotherPartErrors = insertValues(table, fixingPartId, 0);
- assertThat(anotherPartErrors, Matchers.not(empty()));
+ assertThat(anotherPartErrors, not(empty()));
}
/**
@@ -299,6 +331,168 @@ public class ItDisasterRecoveryReconfigurationTest extends ClusterPerTestIntegra
assertThat(errors, is(empty()));
}
+ /**
+ * Tests a scenario where all stable nodes are lost, yet we have data on one of pending nodes and perform reset partition operation. In
+ * this case we should use that pending node as a source of data for recovery.
+ *
+ * <p>It goes like this:
+ * <ul>
+ * <li>We have 6 nodes and a partition on nodes 1, 4 and 5.</li>
+ * <li>We stop nodes 4 and 5, leaving node 1 alone in stable assignments.</li>
+ * <li>New distribution is 0, 1 and 3. Rebalance is started via raft snapshots. It transfers data to node 0, but not node 3.</li>
+ * <li>Node 1 is stopped. Data is only present on node 0.</li>
+ * <li>We execute "resetPartitions" and expect that data from node 0 will be available after that.</li>
+ * </ul>
+ */
+ @Test
+ @ZoneParams(nodes = 6, replicas = 3, partitions = 1)
+ public void testIncompleteRebalanceAfterResetPartitions() throws Exception {
+ int partId = 0;
+
+ Assignments assignment013 = Assignments.of(
+ Assignment.forPeer(node(0).name()),
+ Assignment.forPeer(node(1).name()),
+ Assignment.forPeer(node(3).name())
+ );
+
+ IgniteImpl node0 = cluster.node(0);
+ Table table = node0.tables().table(TABLE_NAME);
+
+ awaitPrimaryReplica(node0, partId);
+ assertRealAssignments(node0, partId, 1, 4, 5);
+
+ insertValues(table, partId, 0);
+
+ triggerRaftSnapshot(1, partId);
+ // Second snapshot causes log truncation.
+ triggerRaftSnapshot(1, partId);
+
+ node(1).dropMessages((nodeName, msg) -> node(3).name().equals(nodeName) && msg instanceof SnapshotMvDataResponse);
+
+ stopNodesInParallel(4, 5);
+ waitForScale(node0, 4);
+
+ assertRealAssignments(node0, partId, 0, 1, 3);
+
+ cluster.runningNodes().forEach(node -> {
+ BiPredicate<String, NetworkMessage> newPredicate = (nodeName, msg) -> stableKeySwitchMessage(msg, partId, assignment013);
+ BiPredicate<String, NetworkMessage> oldPredicate = node.dropMessagesPredicate();
+
+ if (oldPredicate == null) {
+ node.dropMessages(newPredicate);
+ } else {
+ node.dropMessages(oldPredicate.or(newPredicate));
+ }
+ });
+
+ CompletableFuture<Void> resetFuture = node0.disasterRecoveryManager().resetPartitions(zoneName, QUALIFIED_TABLE_NAME, emptySet());
+ assertThat(resetFuture, willCompleteSuccessfully());
+
+ waitForPartitionState(node0, GlobalPartitionStateEnum.DEGRADED);
+
+ var localStatesFut = node0.disasterRecoveryManager().localPartitionStates(emptySet(), Set.of(node(3).name()), emptySet());
+ assertThat(localStatesFut, willCompleteSuccessfully());
+
+ Map<TablePartitionId, LocalPartitionStateByNode> localStates = localStatesFut.join();
+ assertThat(localStates, is(not(anEmptyMap())));
+ assertEquals(LocalPartitionStateEnum.INSTALLING_SNAPSHOT, localStates.values().iterator().next().values().iterator().next().state);
+
+ stopNode(1);
+ waitForScale(node0, 3);
+
+ waitForPartitionState(node0, GlobalPartitionStateEnum.DEGRADED);
+
+ resetFuture = node0.disasterRecoveryManager().resetPartitions(zoneName, QUALIFIED_TABLE_NAME, emptySet());
+ assertThat(resetFuture, willCompleteSuccessfully());
+
+ waitForPartitionState(node0, GlobalPartitionStateEnum.AVAILABLE);
+
+ awaitPrimaryReplica(node0, partId);
+ assertRealAssignments(node0, partId, 0, 2, 3);
+
+ // Set time in the future to protect us from "getAsync" from the past.
+ // Should be replaced with "sleep" when clock skew validation is implemented.
+ node0.clock().update(node0.clock().now().addPhysicalTime(
+ SECONDS.toMillis(DEFAULT_IDLE_SAFE_TIME_PROP_DURATION) + node0.clockService().maxClockSkewMillis())
+ );
+
+ // TODO https://issues.apache.org/jira/browse/IGNITE-21303
+ // We need wait quite a bit before data is available. Log shows term mismatches, meaning that right now it only works due to some
+ // miracle. For future improvements we must specify "stable" forced sub-assignments explicitly, instead of calculating them as an
+ // intersection.
+ Thread.sleep(10_000);
+
+ // "forEach" makes "i" effectively final, which is convenient for internal lambda.
+ IntStream.range(0, ENTRIES).forEach(i -> {
+ CompletableFuture<Tuple> fut = table.keyValueView().getAsync(null, Tuple.create(of("id", i)));
+ assertThat(fut, willCompleteSuccessfully());
+
+ assertNotNull(fut.join());
+ });
+ }
+
+ private boolean stableKeySwitchMessage(NetworkMessage msg, int partId, Assignments blockedAssignments) {
+ if (msg instanceof WriteActionRequest) {
+ var writeActionRequest = (WriteActionRequest) msg;
+ WriteCommand command = writeActionRequest.deserializedCommand();
+
+ if (command instanceof MultiInvokeCommand) {
+ MultiInvokeCommand multiInvokeCommand = (MultiInvokeCommand) command;
+
+ Statement andThen = multiInvokeCommand.iif().andThen();
+
+ if (andThen instanceof UpdateStatement) {
+ UpdateStatement updateStatement = (UpdateStatement) andThen;
+ Collection<Operation> operations = updateStatement.update().operations();
+
+ ByteArray raftConfigurationAppliedKey = raftConfigurationAppliedKey(new TablePartitionId(tableId, partId));
+
+ for (Operation operation : operations) {
+ ByteArray opKey = new ByteArray(operation.key());
+
+ if (operation.type() == OperationType.PUT && opKey.equals(raftConfigurationAppliedKey)) {
+ return blockedAssignments.equals(ByteUtils.fromBytes(operation.value()));
+ }
+ }
+ }
+ }
+ }
+
+ return false;
+ }
+
+ private void waitForPartitionState(IgniteImpl node0, GlobalPartitionStateEnum expectedState) throws InterruptedException {
+ assertTrue(waitForCondition(() -> {
+ CompletableFuture<Map<TablePartitionId, GlobalPartitionState>> statesFuture = node0.disasterRecoveryManager()
+ .globalPartitionStates(Set.of(zoneName), emptySet());
+
+ assertThat(statesFuture, willCompleteSuccessfully());
+
+ Map<TablePartitionId, GlobalPartitionState> map = statesFuture.join();
+
+ GlobalPartitionState state = map.values().iterator().next();
+
+ return state.state == expectedState;
+ }, 500, 20_000));
+ }
+
+ private void triggerRaftSnapshot(int nodeIdx, int partId) throws InterruptedException, ExecutionException {
+ //noinspection resource
+ IgniteImpl node = node(nodeIdx);
+
+ var raftNodeId = new RaftNodeId(new TablePartitionId(tableId, partId), new Peer(node.name()));
+ var jraftServer = (JraftServerImpl) node.raftManager().server();
+
+ RaftGroupService raftGroupService = jraftServer.raftGroupService(raftNodeId);
+ assertNotNull(raftGroupService);
+
+ CompletableFuture<Status> fut = new CompletableFuture<>();
+ raftGroupService.getRaftNode().snapshot(fut::complete);
+
+ assertThat(fut, willCompleteSuccessfully());
+ assertEquals(RaftError.SUCCESS, fut.get().getRaftError());
+ }
+
private void awaitPrimaryReplica(IgniteImpl node0, int partId) {
CompletableFuture<ReplicaMeta> awaitPrimaryReplicaFuture = node0.placementDriver()
.awaitPrimaryReplica(new TablePartitionId(tableId, partId), node0.clock().now(), 60, SECONDS);
@@ -358,7 +552,8 @@ public class ItDisasterRecoveryReconfigurationTest extends ClusterPerTestIntegra
private void startNodesInParallel(int... nodeIndexes) {
//noinspection resource
- runRace(IntStream.of(nodeIndexes).<RunnableX>mapToObj(i -> () -> cluster.startNode(i)).toArray(RunnableX[]::new));
+
+ runRace(20_000, IntStream.of(nodeIndexes).<RunnableX>mapToObj(i -> () -> cluster.startNode(i)).toArray(RunnableX[]::new));
}
/**
@@ -372,7 +567,7 @@ public class ItDisasterRecoveryReconfigurationTest extends ClusterPerTestIntegra
private void waitForScale(IgniteImpl node, int targetDataNodesCount) throws InterruptedException {
DistributionZoneManager dzManager = node.distributionZoneManager();
- assertTrue(IgniteTestUtils.waitForCondition(() -> {
+ assertTrue(waitForCondition(() -> {
long causalityToken = node.metaStorageManager().appliedRevision();
long msSafeTime = node.metaStorageManager().timestampByRevision(causalityToken).longValue();
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 c9b7e4927d..61e159e93b 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
@@ -1857,7 +1857,6 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
// This condition can only pass if all stable nodes are dead, and we start new raft group from scratch.
// In this case new initial configuration must match new forced assignments.
- // TODO https://issues.apache.org/jira/browse/IGNITE-21661 Something might not work, extensive testing is required.
if (nonStableNodeAssignments.nodes().isEmpty()) {
nonStableNodeAssignments = Assignments.forced(pendingAssignmentsNodes);
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryManager.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryManager.java
index 968f0412be..3f8205b721 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryManager.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryManager.java
@@ -103,7 +103,7 @@ import org.jetbrains.annotations.Nullable;
*/
public class DisasterRecoveryManager implements IgniteComponent, SystemViewProvider {
/** Logger. */
- private static final IgniteLogger LOG = Loggers.forClass(DisasterRecoveryManager.class);
+ static final IgniteLogger LOG = Loggers.forClass(DisasterRecoveryManager.class);
/** Single key for writing disaster recovery requests into meta-storage. */
static final ByteArray RECOVERY_TRIGGER_KEY = new ByteArray("disaster.recovery.trigger");
@@ -184,7 +184,10 @@ public class DisasterRecoveryManager implements IgniteComponent, SystemViewProvi
watchListener = new WatchListener() {
@Override
public CompletableFuture<Void> onUpdate(WatchEvent event) {
- return handleTriggerKeyUpdate(event);
+ handleTriggerKeyUpdate(event);
+
+ // There is no need to block a watch thread any longer.
+ return nullCompletedFuture();
}
@Override
@@ -249,7 +252,7 @@ public class DisasterRecoveryManager implements IgniteComponent, SystemViewProvi
checkPartitionsRange(partitionIds, Set.of(zone));
- return processNewRequest(new ManualGroupUpdateRequest(UUID.randomUUID(), zone.id(), tableId, partitionIds));
+ return processNewRequest(new ManualGroupUpdateRequest(UUID.randomUUID(), catalog.version(), zone.id(), tableId, partitionIds));
} catch (Throwable t) {
return failedFuture(t);
}
@@ -335,7 +338,7 @@ public class DisasterRecoveryManager implements IgniteComponent, SystemViewProvi
}
}
- private CompletableFuture<Map<TablePartitionId, LocalPartitionStateMessageByNode>> localPartitionStatesInternal(
+ CompletableFuture<Map<TablePartitionId, LocalPartitionStateMessageByNode>> localPartitionStatesInternal(
Set<String> zoneNames,
Set<String> nodeNames,
Set<Integer> partitionIds,
@@ -483,7 +486,7 @@ public class DisasterRecoveryManager implements IgniteComponent, SystemViewProvi
* Handler for {@link #RECOVERY_TRIGGER_KEY} update event. Deserializes the request and delegates the execution to
* {@link DisasterRecoveryRequest#handle(DisasterRecoveryManager, long)}.
*/
- private CompletableFuture<Void> handleTriggerKeyUpdate(WatchEvent watchEvent) {
+ private void handleTriggerKeyUpdate(WatchEvent watchEvent) {
Entry newEntry = watchEvent.entryEvent().newEntry();
byte[] requestBytes = newEntry.value();
@@ -495,7 +498,7 @@ public class DisasterRecoveryManager implements IgniteComponent, SystemViewProvi
} catch (Exception e) {
LOG.warn("Unable to deserialize disaster recovery request.", e);
- return nullCompletedFuture();
+ return;
}
CompletableFuture<Void> operationFuture = ongoingOperationsById.remove(request.operationId());
@@ -504,27 +507,29 @@ public class DisasterRecoveryManager implements IgniteComponent, SystemViewProvi
case SINGLE_NODE:
if (operationFuture == null) {
// We're not the initiator, or timeout has passed. Just ignore it.
- return nullCompletedFuture();
+ return;
}
- return request.handle(this, watchEvent.revision()).whenComplete(copyStateTo(operationFuture));
+ request.handle(this, watchEvent.revision()).whenComplete(copyStateTo(operationFuture));
+
+ break;
case MULTI_NODE:
CompletableFuture<Void> handleFuture = request.handle(this, watchEvent.revision());
if (operationFuture == null) {
// We're not the initiator, or timeout has passed.
- return handleFuture;
+ return;
}
- return handleFuture.whenComplete(copyStateTo(operationFuture));
+ handleFuture.whenComplete(copyStateTo(operationFuture));
+
+ break;
default:
var error = new AssertionError("Unexpected request type: " + request.getClass());
if (operationFuture != null) {
operationFuture.completeExceptionally(error);
}
-
- return failedFuture(error);
}
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/GlobalPartitionState.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/GlobalPartitionState.java
index 754d57a50e..af78a002be 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/GlobalPartitionState.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/GlobalPartitionState.java
@@ -17,16 +17,23 @@
package org.apache.ignite.internal.table.distributed.disaster;
+import org.apache.ignite.internal.tostring.IgniteToStringInclude;
+import org.apache.ignite.internal.tostring.S;
+
/**
* Global partition state.
*/
public class GlobalPartitionState {
+ @IgniteToStringInclude
public final String tableName;
+ @IgniteToStringInclude
public final String zoneName;
+ @IgniteToStringInclude
public final int partitionId;
+ @IgniteToStringInclude
public final GlobalPartitionStateEnum state;
GlobalPartitionState(String tableName, String zoneName, int partitionId, GlobalPartitionStateEnum state) {
@@ -35,4 +42,9 @@ public class GlobalPartitionState {
this.partitionId = partitionId;
this.state = state;
}
+
+ @Override
+ public String toString() {
+ return S.toString(GlobalPartitionState.class, this);
+ }
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/LocalPartitionState.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/LocalPartitionState.java
index 0ef16deebc..d75a1381eb 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/LocalPartitionState.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/LocalPartitionState.java
@@ -18,17 +18,23 @@
package org.apache.ignite.internal.table.distributed.disaster;
import org.apache.ignite.internal.partition.replicator.network.disaster.LocalPartitionStateEnum;
+import org.apache.ignite.internal.tostring.IgniteToStringInclude;
+import org.apache.ignite.internal.tostring.S;
/**
* Local partition state.
*/
public class LocalPartitionState {
+ @IgniteToStringInclude
public final String tableName;
+ @IgniteToStringInclude
public final String zoneName;
+ @IgniteToStringInclude
public final int partitionId;
+ @IgniteToStringInclude
public final LocalPartitionStateEnum state;
LocalPartitionState(String tableName, String zoneName, int partitionId, LocalPartitionStateEnum state) {
@@ -37,4 +43,9 @@ public class LocalPartitionState {
this.partitionId = partitionId;
this.state = state;
}
+
+ @Override
+ public String toString() {
+ return S.toString(LocalPartitionState.class, this);
+ }
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/LocalPartitionStateByNode.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/LocalPartitionStateByNode.java
index a53301cf1b..8e7e5b9036 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/LocalPartitionStateByNode.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/LocalPartitionStateByNode.java
@@ -21,12 +21,14 @@ import java.util.Collection;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
+import org.apache.ignite.internal.tostring.IgniteToStringInclude;
/** Container for LocalPartitionState to node name map. */
public class LocalPartitionStateByNode {
+ @IgniteToStringInclude
private final Map<String, LocalPartitionState> map;
- public LocalPartitionStateByNode(Map<String, LocalPartitionState> map) {
+ LocalPartitionStateByNode(Map<String, LocalPartitionState> map) {
this.map = Map.copyOf(map);
}
@@ -44,4 +46,9 @@ public class LocalPartitionStateByNode {
public Set<String> keySet() {
return map.keySet();
}
+
+ @Override
+ public String toString() {
+ return map.toString();
+ }
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/LocalPartitionStateMessageByNode.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/LocalPartitionStateMessageByNode.java
index 285cb8cbc1..4e0fc5f228 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/LocalPartitionStateMessageByNode.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/LocalPartitionStateMessageByNode.java
@@ -50,4 +50,9 @@ public class LocalPartitionStateMessageByNode {
public void put(String nodeName, LocalPartitionStateMessage state) {
map.put(nodeName, state);
}
+
+ @Override
+ public String toString() {
+ return map.toString();
+ }
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/ManualGroupUpdateRequest.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/ManualGroupUpdateRequest.java
index 261a72c0b9..d02f3bc619 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/ManualGroupUpdateRequest.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/ManualGroupUpdateRequest.java
@@ -17,20 +17,58 @@
package org.apache.ignite.internal.table.distributed.disaster;
+import static java.util.Collections.emptySet;
import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
import static java.util.stream.Collectors.toSet;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.UpdateStatus.ASSIGNMENT_NOT_UPDATED;
+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.plannedPartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.tableAssignments;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.value;
+import static org.apache.ignite.internal.metastorage.dsl.Operations.ops;
+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.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.table.distributed.disaster.DisasterRecoveryRequestType.SINGLE_NODE;
+import static org.apache.ignite.internal.util.ByteUtils.longToBytesKeepingOrder;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.lang.ErrorGroups.DisasterRecovery.CLUSTER_NOT_IDLE_ERR;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Map.Entry;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+import java.util.stream.IntStream;
+import org.apache.ignite.internal.affinity.AffinityUtils;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.affinity.Assignments;
import org.apache.ignite.internal.catalog.Catalog;
import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
import org.apache.ignite.internal.distributionzones.NodeWithAttributes;
-import org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil;
+import org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.UpdateStatus;
import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.metastorage.MetaStorageManager;
+import org.apache.ignite.internal.metastorage.dsl.Iif;
+import org.apache.ignite.internal.metastorage.dsl.StatementResult;
+import org.apache.ignite.internal.partition.replicator.network.disaster.LocalPartitionStateMessage;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.table.distributed.disaster.exceptions.DisasterRecoveryException;
import org.apache.ignite.internal.tostring.S;
+import org.apache.ignite.internal.util.CollectionUtils;
class ManualGroupUpdateRequest implements DisasterRecoveryRequest {
/** Serial version UID. */
@@ -38,14 +76,20 @@ class ManualGroupUpdateRequest implements DisasterRecoveryRequest {
private final UUID operationId;
+ /**
+ * Catalog version at the moment of operation creation. Must match catalog version at the moment of operation execution.
+ */
+ private final int catalogVersion;
+
private final int zoneId;
private final int tableId;
private final Set<Integer> partitionIds;
- ManualGroupUpdateRequest(UUID operationId, int zoneId, int tableId, Set<Integer> partitionIds) {
+ ManualGroupUpdateRequest(UUID operationId, int catalogVersion, int zoneId, int tableId, Set<Integer> partitionIds) {
this.operationId = operationId;
+ this.catalogVersion = catalogVersion;
this.zoneId = zoneId;
this.tableId = tableId;
this.partitionIds = Set.copyOf(partitionIds);
@@ -66,6 +110,10 @@ class ManualGroupUpdateRequest implements DisasterRecoveryRequest {
return SINGLE_NODE;
}
+ public int catalogVersion() {
+ return catalogVersion;
+ }
+
public int tableId() {
return tableId;
}
@@ -79,31 +127,190 @@ class ManualGroupUpdateRequest implements DisasterRecoveryRequest {
HybridTimestamp msSafeTime = disasterRecoveryManager.metaStorageManager.timestampByRevision(msRevision);
int catalogVersion = disasterRecoveryManager.catalogManager.activeCatalogVersion(msSafeTime.longValue());
+
+ if (this.catalogVersion != catalogVersion) {
+ return failedFuture(
+ new DisasterRecoveryException(CLUSTER_NOT_IDLE_ERR, "Cluster is not idle, concurrent DDL update detected.")
+ );
+ }
+
Catalog catalog = disasterRecoveryManager.catalogManager.catalog(catalogVersion);
CatalogZoneDescriptor zoneDescriptor = catalog.zone(zoneId);
CatalogTableDescriptor tableDescriptor = catalog.table(tableId);
+ CompletableFuture<Map<TablePartitionId, LocalPartitionStateMessageByNode>> localStates
+ = disasterRecoveryManager.localPartitionStatesInternal(Set.of(zoneDescriptor.name()), emptySet(), partitionIds, catalog);
+
CompletableFuture<Set<String>> dataNodesFuture = disasterRecoveryManager.dzManager.dataNodes(msRevision, catalogVersion, zoneId);
- return dataNodesFuture.thenCompose(dataNodes -> {
+ return dataNodesFuture.thenCombine(localStates, (dataNodes, localStatesMap) -> {
Set<String> nodeConsistentIds = disasterRecoveryManager.dzManager.logicalTopology()
.stream()
.map(NodeWithAttributes::nodeName)
.collect(toSet());
- CompletableFuture<?>[] futures = RebalanceUtil.forceAssignmentsUpdate(
+ CompletableFuture<?>[] futures = forceAssignmentsUpdate(
tableDescriptor,
zoneDescriptor,
- partitionIds,
dataNodes,
nodeConsistentIds,
msRevision,
- disasterRecoveryManager.metaStorageManager
+ disasterRecoveryManager.metaStorageManager,
+ localStatesMap
);
return allOf(futures);
- });
+ }).thenCompose(Function.identity());
+ }
+
+ /**
+ * Sets force assignments for the zone/table if it's required. The condition for force reassignment is the absence of stable
+ * assignments' majority within the set of currently alive nodes. In this case we calculate new assignments that include all alive
+ * stable nodes, and try to save ot with a {@link Assignments#force()} flag enabled.
+ *
+ * @param tableDescriptor Table descriptor.
+ * @param zoneDescriptor Zone descriptor.
+ * @param dataNodes Current DZ data nodes.
+ * @param aliveNodesConsistentIds Set of alive nodes according to logical topology.
+ * @param revision Meta-storage revision to be associated with reassignment.
+ * @param metaStorageManager Meta-storage manager.
+ * @param localStatesMap Local partition states retrieved by {@link DisasterRecoveryManager#localPartitionStates(Set, Set, Set)}.
+ * @return A future that will be completed when reassignments data is written into a meta-storage, if that's required.
+ */
+ private CompletableFuture<?>[] forceAssignmentsUpdate(
+ CatalogTableDescriptor tableDescriptor,
+ CatalogZoneDescriptor zoneDescriptor,
+ Set<String> dataNodes,
+ Set<String> aliveNodesConsistentIds,
+ long revision,
+ MetaStorageManager metaStorageManager,
+ Map<TablePartitionId, LocalPartitionStateMessageByNode> localStatesMap
+ ) {
+ CompletableFuture<Map<Integer, Assignments>> tableAssignmentsFut = tableAssignments(
+ metaStorageManager,
+ tableDescriptor.id(),
+ partitionIds,
+ zoneDescriptor.partitions()
+ );
+
+ Set<String> aliveDataNodes = CollectionUtils.intersect(dataNodes, aliveNodesConsistentIds);
+
+ int[] partitionIdsArray = partitionIds.isEmpty()
+ ? IntStream.range(0, zoneDescriptor.partitions()).toArray()
+ : partitionIds.stream().mapToInt(Integer::intValue).toArray();
+
+ CompletableFuture<?>[] futures = new CompletableFuture[partitionIdsArray.length];
+
+ for (int partitionId = 0; partitionId < partitionIdsArray.length; partitionId++) {
+ TablePartitionId replicaGrpId = new TablePartitionId(tableDescriptor.id(), partitionIdsArray[partitionId]);
+
+ futures[partitionId] = tableAssignmentsFut.thenCompose(tableAssignments ->
+ tableAssignments.isEmpty() ? nullCompletedFuture() : manualPartitionUpdate(
+ replicaGrpId,
+ aliveDataNodes,
+ aliveNodesConsistentIds,
+ zoneDescriptor.replicas(),
+ revision,
+ metaStorageManager,
+ tableAssignments.get(replicaGrpId.partitionId()).nodes(),
+ localStatesMap.get(replicaGrpId)
+ )).thenAccept(res -> {
+ DisasterRecoveryManager.LOG.info(
+ "Partition {} returned {} status on reset attempt", replicaGrpId, UpdateStatus.valueOf(res)
+ );
+ }
+ );
+ }
+
+ return futures;
+ }
+
+ private static CompletableFuture<Integer> manualPartitionUpdate(
+ TablePartitionId partId,
+ Collection<String> aliveDataNodes,
+ Set<String> aliveNodesConsistentIds,
+ int replicas,
+ long revision,
+ MetaStorageManager metaStorageMgr,
+ Set<Assignment> currentAssignments,
+ LocalPartitionStateMessageByNode localPartitionStateMessageByNode
+ ) {
+ // TODO https://issues.apache.org/jira/browse/IGNITE-21303
+ // This is a naive approach that doesn't exclude nodes in error state, if they exist.
+ Set<Assignment> partAssignments = getAliveNodesWithData(aliveNodesConsistentIds, localPartitionStateMessageByNode);
+
+ Set<Assignment> aliveStableNodes = CollectionUtils.intersect(currentAssignments, partAssignments);
+
+ if (aliveStableNodes.size() >= (replicas / 2 + 1)) {
+ return completedFuture(ASSIGNMENT_NOT_UPDATED.ordinal());
+ }
+
+ enrichAssignments(partId, aliveDataNodes, replicas, partAssignments);
+
+ byte[] partAssignmentsBytes = Assignments.forced(partAssignments).toBytes();
+ byte[] revisionBytes = longToBytesKeepingOrder(revision);
+
+ ByteArray partChangeTriggerKey = pendingChangeTriggerKey(partId);
+ ByteArray partAssignmentsPendingKey = pendingPartAssignmentsKey(partId);
+ ByteArray partAssignmentsPlannedKey = plannedPartAssignmentsKey(partId);
+
+ Iif iif = iif(
+ notExists(partChangeTriggerKey).or(value(partChangeTriggerKey).lt(revisionBytes)),
+ iif(
+ value(partAssignmentsPendingKey).ne(partAssignmentsBytes),
+ ops(
+ put(partAssignmentsPendingKey, partAssignmentsBytes),
+ put(partChangeTriggerKey, revisionBytes),
+ remove(partAssignmentsPlannedKey)
+ ).yield(PENDING_KEY_UPDATED.ordinal()),
+ ops().yield(ASSIGNMENT_NOT_UPDATED.ordinal())
+ ),
+ ops().yield(OUTDATED_UPDATE_RECEIVED.ordinal())
+ );
+
+ return metaStorageMgr.invoke(iif).thenApply(StatementResult::getAsInt);
+ }
+
+ /**
+ * Returns a set of nodes that are both alive and either {@link LocalPartitionStateEnum#HEALTHY} or
+ * {@link LocalPartitionStateEnum#CATCHING_UP}.
+ */
+ private static Set<Assignment> getAliveNodesWithData(
+ Set<String> aliveNodesConsistentIds,
+ LocalPartitionStateMessageByNode localPartitionStateMessageByNode
+ ) {
+ Set<Assignment> partAssignments = new HashSet<>();
+ if (localPartitionStateMessageByNode != null) {
+ for (Entry<String, LocalPartitionStateMessage> entry : localPartitionStateMessageByNode.entrySet()) {
+ if (aliveNodesConsistentIds.contains(entry.getKey())
+ && (entry.getValue().state() == HEALTHY || entry.getValue().state() == CATCHING_UP)
+ ) {
+ partAssignments.add(Assignment.forPeer(entry.getKey()));
+ }
+ }
+ }
+ return partAssignments;
+ }
+
+ /**
+ * Adds more nodes into {@code partAssignments} until it matches the number of replicas or we run out of nodes.
+ */
+ private static void enrichAssignments(
+ TablePartitionId partId,
+ Collection<String> aliveDataNodes,
+ int replicas,
+ Set<Assignment> partAssignments
+ ) {
+ Set<Assignment> calcAssignments = AffinityUtils.calculateAssignmentForPartition(aliveDataNodes, partId.partitionId(), replicas);
+
+ for (Assignment calcAssignment : calcAssignments) {
+ if (partAssignments.size() == replicas) {
+ break;
+ }
+
+ partAssignments.add(calcAssignment);
+ }
}
@Override
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/exceptions/DisasterRecoveryException.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/exceptions/DisasterRecoveryException.java
index e3d4fcbfdc..694be71367 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/exceptions/DisasterRecoveryException.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/exceptions/DisasterRecoveryException.java
@@ -18,16 +18,18 @@
package org.apache.ignite.internal.table.distributed.disaster.exceptions;
import org.apache.ignite.internal.lang.IgniteInternalException;
+import org.apache.ignite.lang.ErrorGroups.DisasterRecovery;
+import org.intellij.lang.annotations.MagicConstant;
/** Common exception for disaster recovery. */
public class DisasterRecoveryException extends IgniteInternalException {
private static final long serialVersionUID = -3565357739782565015L;
- public DisasterRecoveryException(int code, Throwable cause) {
+ public DisasterRecoveryException(@MagicConstant(valuesFromClass = DisasterRecovery.class) int code, Throwable cause) {
super(code, cause);
}
- public DisasterRecoveryException(int code, String message) {
+ public DisasterRecoveryException(@MagicConstant(valuesFromClass = DisasterRecovery.class) int code, String message) {
super(code, message);
}
}