rpuch commented on code in PR #5584:
URL: https://github.com/apache/ignite-3/pull/5584#discussion_r2046676755
##########
modules/table/src/integrationTest/java/org/apache/ignite/internal/table/distributed/disaster/AbstractHighAvailablePartitionsRecoveryTest.java:
##########
@@ -286,11 +293,26 @@ private static Set<String>
assignmentsFromPendingEntry(Entry entry) {
}
private Set<Assignment> getPartitionClusterNodes(IgniteImpl node, String
tableName, int partNum) {
+ if (enabledColocation()) {
+ int zoneId =
TableTestUtils.getZoneIdByTableNameStrict(node.catalogManager(), tableName,
clock.nowLong());
+ return getZonePartitionClusterNodes(node, zoneId, partNum);
+ } else {
+ return getTablePartitionClusterNodes(node, tableName, partNum);
+ }
+ }
+
+ private Set<Assignment> getTablePartitionClusterNodes(IgniteImpl node,
String tableName, int partNum) {
return Optional.ofNullable(getTableId(node.catalogManager(),
tableName, clock.nowLong()))
.map(tableId ->
stablePartitionAssignments(node.metaStorageManager(), tableId, partNum).join())
.orElse(Set.of());
}
+ private Set<Assignment> getZonePartitionClusterNodes(IgniteImpl node,
@Nullable Integer zoneId, int partNum) {
+ return Optional.ofNullable(zoneId)
+ .map(id ->
ZoneRebalanceUtil.zonePartitionAssignments(node.metaStorageManager(), id,
partNum).join())
Review Comment:
Please use `get()` instead of `join()` as the latter is uninterruptible
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/GroupUpdateRequestHandler.java:
##########
@@ -0,0 +1,676 @@
+/*
+ * 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.table.distributed.disaster;
+
+import static java.util.Collections.emptyMap;
+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.tableStableAssignments;
+import static
org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.zoneStableAssignments;
+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.partitiondistribution.PartitionDistributionUtils.calculateAssignmentForPartition;
+import static
org.apache.ignite.internal.partitiondistribution.PendingAssignmentsCalculator.pendingAssignmentsCalculator;
+import static
org.apache.ignite.internal.table.distributed.disaster.DisasterRecoveryManager.tableState;
+import static
org.apache.ignite.internal.table.distributed.disaster.DisasterRecoveryManager.zoneState;
+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.ArrayList;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+import org.apache.ignite.internal.catalog.Catalog;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.distributionzones.NodeWithAttributes;
+import org.apache.ignite.internal.distributionzones.rebalance.AssignmentUtil;
+import org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil;
+import
org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.UpdateStatus;
+import
org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.metastorage.MetaStorageManager;
+import org.apache.ignite.internal.metastorage.dsl.Iif;
+import
org.apache.ignite.internal.partition.replicator.network.disaster.LocalPartitionStateEnum;
+import
org.apache.ignite.internal.partition.replicator.network.disaster.LocalPartitionStateMessage;
+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.PartitionGroupId;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import
org.apache.ignite.internal.table.distributed.disaster.exceptions.DisasterRecoveryException;
+import org.apache.ignite.internal.util.CollectionUtils;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * A colocation-aware handler for {@link GroupUpdateRequest}.
+ */
+abstract class GroupUpdateRequestHandler<T extends PartitionGroupId> {
+ private static final IgniteLogger LOG =
Loggers.forClass(GroupUpdateRequest.class);
+
+ public static GroupUpdateRequestHandler<?> handler(GroupUpdateRequest
request) {
+ return request.colocationEnabled()
+ ? new ZoneGroupUpdateRequestHandler(request)
+ : new TableGroupUpdateRequestHandler(request);
+ }
+
+ private final GroupUpdateRequest request;
+
+ GroupUpdateRequestHandler(GroupUpdateRequest request) {
+ this.request = request;
+ }
+
+ public CompletableFuture<Void> handle(DisasterRecoveryManager
disasterRecoveryManager, long msRevision, HybridTimestamp msTimestamp) {
+ int catalogVersion =
disasterRecoveryManager.catalogManager.activeCatalogVersion(msTimestamp.longValue());
+
+ if (request.catalogVersion() != catalogVersion) {
+ return failedFuture(
+ new DisasterRecoveryException(CLUSTER_NOT_IDLE_ERR,
"Cluster is not idle, concurrent DDL update detected.")
+ );
+ }
+
+ Catalog catalog =
disasterRecoveryManager.catalogManager.catalog(catalogVersion);
+
+ int zoneId = zoneId();
+
+ CatalogZoneDescriptor zoneDescriptor = catalog.zone(zoneId);
+
+ Set<Integer> allZonePartitionsToReset = new HashSet<>();
+
request.partitionIds().values().forEach(allZonePartitionsToReset::addAll);
+
+ CompletableFuture<Set<String>> dataNodesFuture =
+ disasterRecoveryManager.dzManager.dataNodes(msTimestamp,
catalogVersion, zoneId);
+
+ CompletableFuture<Map<T, LocalPartitionStateMessageByNode>>
localStatesFuture =
+ localStatesFuture(disasterRecoveryManager,
Set.of(zoneDescriptor.name()), allZonePartitionsToReset, catalog);
+
+ return dataNodesFuture.thenCombine(localStatesFuture, (dataNodes,
localStatesMap) -> {
+ Set<String> nodeConsistentIds =
disasterRecoveryManager.dzManager.logicalTopology(msRevision)
+ .stream()
+ .map(NodeWithAttributes::nodeName)
+ .collect(toSet());
+
+ List<CompletableFuture<Void>> tableFuts = new
ArrayList<>(request.partitionIds().size());
Review Comment:
`tableFuts` is not about just tables anymore as `partitionIds` might also be
about zones
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/GroupUpdateRequestHandler.java:
##########
@@ -0,0 +1,676 @@
+/*
+ * 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.table.distributed.disaster;
+
+import static java.util.Collections.emptyMap;
+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.tableStableAssignments;
+import static
org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.zoneStableAssignments;
+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.partitiondistribution.PartitionDistributionUtils.calculateAssignmentForPartition;
+import static
org.apache.ignite.internal.partitiondistribution.PendingAssignmentsCalculator.pendingAssignmentsCalculator;
+import static
org.apache.ignite.internal.table.distributed.disaster.DisasterRecoveryManager.tableState;
+import static
org.apache.ignite.internal.table.distributed.disaster.DisasterRecoveryManager.zoneState;
+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.ArrayList;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+import org.apache.ignite.internal.catalog.Catalog;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.distributionzones.NodeWithAttributes;
+import org.apache.ignite.internal.distributionzones.rebalance.AssignmentUtil;
+import org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil;
+import
org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.UpdateStatus;
+import
org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.metastorage.MetaStorageManager;
+import org.apache.ignite.internal.metastorage.dsl.Iif;
+import
org.apache.ignite.internal.partition.replicator.network.disaster.LocalPartitionStateEnum;
+import
org.apache.ignite.internal.partition.replicator.network.disaster.LocalPartitionStateMessage;
+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.PartitionGroupId;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import
org.apache.ignite.internal.table.distributed.disaster.exceptions.DisasterRecoveryException;
+import org.apache.ignite.internal.util.CollectionUtils;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * A colocation-aware handler for {@link GroupUpdateRequest}.
+ */
+abstract class GroupUpdateRequestHandler<T extends PartitionGroupId> {
+ private static final IgniteLogger LOG =
Loggers.forClass(GroupUpdateRequest.class);
+
+ public static GroupUpdateRequestHandler<?> handler(GroupUpdateRequest
request) {
+ return request.colocationEnabled()
+ ? new ZoneGroupUpdateRequestHandler(request)
+ : new TableGroupUpdateRequestHandler(request);
+ }
+
+ private final GroupUpdateRequest request;
+
+ GroupUpdateRequestHandler(GroupUpdateRequest request) {
+ this.request = request;
+ }
+
+ public CompletableFuture<Void> handle(DisasterRecoveryManager
disasterRecoveryManager, long msRevision, HybridTimestamp msTimestamp) {
+ int catalogVersion =
disasterRecoveryManager.catalogManager.activeCatalogVersion(msTimestamp.longValue());
+
+ if (request.catalogVersion() != catalogVersion) {
+ return failedFuture(
+ new DisasterRecoveryException(CLUSTER_NOT_IDLE_ERR,
"Cluster is not idle, concurrent DDL update detected.")
+ );
+ }
+
+ Catalog catalog =
disasterRecoveryManager.catalogManager.catalog(catalogVersion);
+
+ int zoneId = zoneId();
+
+ CatalogZoneDescriptor zoneDescriptor = catalog.zone(zoneId);
+
+ Set<Integer> allZonePartitionsToReset = new HashSet<>();
+
request.partitionIds().values().forEach(allZonePartitionsToReset::addAll);
+
+ CompletableFuture<Set<String>> dataNodesFuture =
+ disasterRecoveryManager.dzManager.dataNodes(msTimestamp,
catalogVersion, zoneId);
+
+ CompletableFuture<Map<T, LocalPartitionStateMessageByNode>>
localStatesFuture =
+ localStatesFuture(disasterRecoveryManager,
Set.of(zoneDescriptor.name()), allZonePartitionsToReset, catalog);
+
+ return dataNodesFuture.thenCombine(localStatesFuture, (dataNodes,
localStatesMap) -> {
+ Set<String> nodeConsistentIds =
disasterRecoveryManager.dzManager.logicalTopology(msRevision)
+ .stream()
+ .map(NodeWithAttributes::nodeName)
+ .collect(toSet());
+
+ List<CompletableFuture<Void>> tableFuts = new
ArrayList<>(request.partitionIds().size());
+
+ for (Entry<Integer, Set<Integer>> partitionEntry :
request.partitionIds().entrySet()) {
+
+ int[] partitionIdsArray =
AssignmentUtil.partitionIds(partitionEntry.getValue(),
zoneDescriptor.partitions());
+
+ tableFuts.add(forceAssignmentsUpdate(
+ partitionEntry.getKey(),
+ zoneDescriptor,
+ dataNodes,
+ nodeConsistentIds,
+ msRevision,
+ msTimestamp,
+ disasterRecoveryManager.metaStorageManager,
+ localStatesMap,
+ catalog.time(),
+ partitionIdsArray,
+ request.manualUpdate()
+ ));
+ }
+
+ return allOf(tableFuts.toArray(new CompletableFuture[]{}));
+ })
+ .thenCompose(Function.identity())
+ .whenComplete((unused, throwable) -> {
+ // TODO: IGNITE-23635 Add fail handling for failed resetPeers
+ if (throwable != null) {
+ LOG.error("Failed to reset partition", throwable);
+ }
+ });
+ }
+
+ protected int zoneId() {
+ return request.zoneId();
+ }
+
+ /**
+ * 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 replicationId Table id or Zone id.
+ * @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 timestamp Meta-storage timestamp to be associated with
reassignment.
+ * @param metaStorageManager Meta-storage manager.
+ * @param localStatesMap Local partition states retrieved by
+ * {@link DisasterRecoveryManager#localTablePartitionStates(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<Void> forceAssignmentsUpdate(
+ int replicationId,
+ CatalogZoneDescriptor zoneDescriptor,
+ Set<String> dataNodes,
+ Set<String> aliveNodesConsistentIds,
+ long revision,
+ HybridTimestamp timestamp,
+ MetaStorageManager metaStorageManager,
+ Map<T, LocalPartitionStateMessageByNode> localStatesMap,
+ long assignmentsTimestamp,
+ int[] partitionIds,
+ boolean manualUpdate
+ ) {
+ CompletableFuture<Map<Integer, Assignments>> stableAssignments =
+ stableAssignments(metaStorageManager, replicationId,
partitionIds);
+ return stableAssignments
+ .thenCompose(assignments -> {
+ if (assignments.isEmpty()) {
+ return nullCompletedFuture();
+ }
+
+ return updateAssignments(
+ replicationId,
+ zoneDescriptor,
+ dataNodes,
+ aliveNodesConsistentIds,
+ revision,
+ timestamp,
+ metaStorageManager,
+ localStatesMap,
+ assignmentsTimestamp,
+ partitionIds,
+ assignments,
+ manualUpdate
+ );
+ });
+ }
+
+ private CompletableFuture<Void> updateAssignments(
+ int entityId,
Review Comment:
Here it's `entityId`, in previous method it's `replicationId`. Would it make
sense to use the same term for consistency?
##########
modules/table/src/integrationTest/java/org/apache/ignite/internal/table/distributed/disaster/AbstractHighAvailablePartitionsRecoveryTest.java:
##########
@@ -286,11 +293,26 @@ private static Set<String>
assignmentsFromPendingEntry(Entry entry) {
}
private Set<Assignment> getPartitionClusterNodes(IgniteImpl node, String
tableName, int partNum) {
+ if (enabledColocation()) {
+ int zoneId =
TableTestUtils.getZoneIdByTableNameStrict(node.catalogManager(), tableName,
clock.nowLong());
+ return getZonePartitionClusterNodes(node, zoneId, partNum);
+ } else {
+ return getTablePartitionClusterNodes(node, tableName, partNum);
+ }
+ }
+
+ private Set<Assignment> getTablePartitionClusterNodes(IgniteImpl node,
String tableName, int partNum) {
return Optional.ofNullable(getTableId(node.catalogManager(),
tableName, clock.nowLong()))
.map(tableId ->
stablePartitionAssignments(node.metaStorageManager(), tableId, partNum).join())
.orElse(Set.of());
}
+ private Set<Assignment> getZonePartitionClusterNodes(IgniteImpl node,
@Nullable Integer zoneId, int partNum) {
Review Comment:
It seems that `zoneId` cannot be `null` here, so its type should be `int`
and `Nullable` use should be avoided
##########
modules/table/src/integrationTest/java/org/apache/ignite/internal/table/distributed/disaster/AbstractHighAvailablePartitionsRecoveryTest.java:
##########
@@ -471,6 +492,16 @@ private void
awaitForAllNodesTableGroupInitialization(Set<Integer> tableIds, int
}, 10_000));
}
+ private static void incrementReplicaCount(RaftNodeId raftNodeId,
IgniteImpl igniteImpl, AtomicInteger numberOfInitializedReplicas) {
Review Comment:
```suggestion
private static void incrementReplicaCountIfHasLog(RaftNodeId raftNodeId,
IgniteImpl igniteImpl, AtomicInteger numberOfInitializedReplicas) {
```
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryManager.java:
##########
@@ -405,7 +420,8 @@ private CompletableFuture<Void> resetPartitions(
String zoneName,
Map<Integer, Set<Integer>> partitionIds,
boolean manualUpdate,
- long triggerRevision
+ long triggerRevision,
Review Comment:
Javadoc of partitionIds parameter is now stale: it says that it's about
per-table partitions, but it might be per-zone partitions as well. Same thing
has to be addressed at least in `GroupUpdateRequest#partitionIds`.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/GroupUpdateRequestHandler.java:
##########
@@ -0,0 +1,676 @@
+/*
+ * 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.table.distributed.disaster;
+
+import static java.util.Collections.emptyMap;
+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.tableStableAssignments;
+import static
org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.zoneStableAssignments;
+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.partitiondistribution.PartitionDistributionUtils.calculateAssignmentForPartition;
+import static
org.apache.ignite.internal.partitiondistribution.PendingAssignmentsCalculator.pendingAssignmentsCalculator;
+import static
org.apache.ignite.internal.table.distributed.disaster.DisasterRecoveryManager.tableState;
+import static
org.apache.ignite.internal.table.distributed.disaster.DisasterRecoveryManager.zoneState;
+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.ArrayList;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+import org.apache.ignite.internal.catalog.Catalog;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.distributionzones.NodeWithAttributes;
+import org.apache.ignite.internal.distributionzones.rebalance.AssignmentUtil;
+import org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil;
+import
org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.UpdateStatus;
+import
org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.metastorage.MetaStorageManager;
+import org.apache.ignite.internal.metastorage.dsl.Iif;
+import
org.apache.ignite.internal.partition.replicator.network.disaster.LocalPartitionStateEnum;
+import
org.apache.ignite.internal.partition.replicator.network.disaster.LocalPartitionStateMessage;
+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.PartitionGroupId;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import
org.apache.ignite.internal.table.distributed.disaster.exceptions.DisasterRecoveryException;
+import org.apache.ignite.internal.util.CollectionUtils;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * A colocation-aware handler for {@link GroupUpdateRequest}.
+ */
+abstract class GroupUpdateRequestHandler<T extends PartitionGroupId> {
+ private static final IgniteLogger LOG =
Loggers.forClass(GroupUpdateRequest.class);
+
+ public static GroupUpdateRequestHandler<?> handler(GroupUpdateRequest
request) {
Review Comment:
Let's put this method after `request` field
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/GroupUpdateRequestHandler.java:
##########
@@ -0,0 +1,676 @@
+/*
+ * 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.table.distributed.disaster;
+
+import static java.util.Collections.emptyMap;
+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.tableStableAssignments;
+import static
org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.zoneStableAssignments;
+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.partitiondistribution.PartitionDistributionUtils.calculateAssignmentForPartition;
+import static
org.apache.ignite.internal.partitiondistribution.PendingAssignmentsCalculator.pendingAssignmentsCalculator;
+import static
org.apache.ignite.internal.table.distributed.disaster.DisasterRecoveryManager.tableState;
+import static
org.apache.ignite.internal.table.distributed.disaster.DisasterRecoveryManager.zoneState;
+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.ArrayList;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+import org.apache.ignite.internal.catalog.Catalog;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.distributionzones.NodeWithAttributes;
+import org.apache.ignite.internal.distributionzones.rebalance.AssignmentUtil;
+import org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil;
+import
org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.UpdateStatus;
+import
org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.metastorage.MetaStorageManager;
+import org.apache.ignite.internal.metastorage.dsl.Iif;
+import
org.apache.ignite.internal.partition.replicator.network.disaster.LocalPartitionStateEnum;
+import
org.apache.ignite.internal.partition.replicator.network.disaster.LocalPartitionStateMessage;
+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.PartitionGroupId;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import
org.apache.ignite.internal.table.distributed.disaster.exceptions.DisasterRecoveryException;
+import org.apache.ignite.internal.util.CollectionUtils;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * A colocation-aware handler for {@link GroupUpdateRequest}.
+ */
+abstract class GroupUpdateRequestHandler<T extends PartitionGroupId> {
+ private static final IgniteLogger LOG =
Loggers.forClass(GroupUpdateRequest.class);
+
+ public static GroupUpdateRequestHandler<?> handler(GroupUpdateRequest
request) {
+ return request.colocationEnabled()
+ ? new ZoneGroupUpdateRequestHandler(request)
+ : new TableGroupUpdateRequestHandler(request);
+ }
+
+ private final GroupUpdateRequest request;
+
+ GroupUpdateRequestHandler(GroupUpdateRequest request) {
+ this.request = request;
+ }
+
+ public CompletableFuture<Void> handle(DisasterRecoveryManager
disasterRecoveryManager, long msRevision, HybridTimestamp msTimestamp) {
+ int catalogVersion =
disasterRecoveryManager.catalogManager.activeCatalogVersion(msTimestamp.longValue());
+
+ if (request.catalogVersion() != catalogVersion) {
+ return failedFuture(
+ new DisasterRecoveryException(CLUSTER_NOT_IDLE_ERR,
"Cluster is not idle, concurrent DDL update detected.")
+ );
+ }
+
+ Catalog catalog =
disasterRecoveryManager.catalogManager.catalog(catalogVersion);
+
+ int zoneId = zoneId();
+
+ CatalogZoneDescriptor zoneDescriptor = catalog.zone(zoneId);
+
+ Set<Integer> allZonePartitionsToReset = new HashSet<>();
+
request.partitionIds().values().forEach(allZonePartitionsToReset::addAll);
+
+ CompletableFuture<Set<String>> dataNodesFuture =
+ disasterRecoveryManager.dzManager.dataNodes(msTimestamp,
catalogVersion, zoneId);
+
+ CompletableFuture<Map<T, LocalPartitionStateMessageByNode>>
localStatesFuture =
+ localStatesFuture(disasterRecoveryManager,
Set.of(zoneDescriptor.name()), allZonePartitionsToReset, catalog);
+
+ return dataNodesFuture.thenCombine(localStatesFuture, (dataNodes,
localStatesMap) -> {
+ Set<String> nodeConsistentIds =
disasterRecoveryManager.dzManager.logicalTopology(msRevision)
+ .stream()
+ .map(NodeWithAttributes::nodeName)
+ .collect(toSet());
+
+ List<CompletableFuture<Void>> tableFuts = new
ArrayList<>(request.partitionIds().size());
+
+ for (Entry<Integer, Set<Integer>> partitionEntry :
request.partitionIds().entrySet()) {
+
+ int[] partitionIdsArray =
AssignmentUtil.partitionIds(partitionEntry.getValue(),
zoneDescriptor.partitions());
+
+ tableFuts.add(forceAssignmentsUpdate(
+ partitionEntry.getKey(),
+ zoneDescriptor,
+ dataNodes,
+ nodeConsistentIds,
+ msRevision,
+ msTimestamp,
+ disasterRecoveryManager.metaStorageManager,
+ localStatesMap,
+ catalog.time(),
+ partitionIdsArray,
+ request.manualUpdate()
+ ));
+ }
+
+ return allOf(tableFuts.toArray(new CompletableFuture[]{}));
+ })
+ .thenCompose(Function.identity())
+ .whenComplete((unused, throwable) -> {
+ // TODO: IGNITE-23635 Add fail handling for failed resetPeers
+ if (throwable != null) {
+ LOG.error("Failed to reset partition", throwable);
+ }
+ });
+ }
+
+ protected int zoneId() {
Review Comment:
```suggestion
protected final int zoneId() {
```
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/GroupUpdateRequestHandler.java:
##########
@@ -0,0 +1,676 @@
+/*
+ * 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.table.distributed.disaster;
+
+import static java.util.Collections.emptyMap;
+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.tableStableAssignments;
+import static
org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.zoneStableAssignments;
+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.partitiondistribution.PartitionDistributionUtils.calculateAssignmentForPartition;
+import static
org.apache.ignite.internal.partitiondistribution.PendingAssignmentsCalculator.pendingAssignmentsCalculator;
+import static
org.apache.ignite.internal.table.distributed.disaster.DisasterRecoveryManager.tableState;
+import static
org.apache.ignite.internal.table.distributed.disaster.DisasterRecoveryManager.zoneState;
+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.ArrayList;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+import org.apache.ignite.internal.catalog.Catalog;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.distributionzones.NodeWithAttributes;
+import org.apache.ignite.internal.distributionzones.rebalance.AssignmentUtil;
+import org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil;
+import
org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.UpdateStatus;
+import
org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.metastorage.MetaStorageManager;
+import org.apache.ignite.internal.metastorage.dsl.Iif;
+import
org.apache.ignite.internal.partition.replicator.network.disaster.LocalPartitionStateEnum;
+import
org.apache.ignite.internal.partition.replicator.network.disaster.LocalPartitionStateMessage;
+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.PartitionGroupId;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import
org.apache.ignite.internal.table.distributed.disaster.exceptions.DisasterRecoveryException;
+import org.apache.ignite.internal.util.CollectionUtils;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * A colocation-aware handler for {@link GroupUpdateRequest}.
+ */
+abstract class GroupUpdateRequestHandler<T extends PartitionGroupId> {
+ private static final IgniteLogger LOG =
Loggers.forClass(GroupUpdateRequest.class);
+
+ public static GroupUpdateRequestHandler<?> handler(GroupUpdateRequest
request) {
+ return request.colocationEnabled()
+ ? new ZoneGroupUpdateRequestHandler(request)
+ : new TableGroupUpdateRequestHandler(request);
+ }
+
+ private final GroupUpdateRequest request;
+
+ GroupUpdateRequestHandler(GroupUpdateRequest request) {
+ this.request = request;
+ }
+
+ public CompletableFuture<Void> handle(DisasterRecoveryManager
disasterRecoveryManager, long msRevision, HybridTimestamp msTimestamp) {
+ int catalogVersion =
disasterRecoveryManager.catalogManager.activeCatalogVersion(msTimestamp.longValue());
+
+ if (request.catalogVersion() != catalogVersion) {
+ return failedFuture(
+ new DisasterRecoveryException(CLUSTER_NOT_IDLE_ERR,
"Cluster is not idle, concurrent DDL update detected.")
+ );
+ }
+
+ Catalog catalog =
disasterRecoveryManager.catalogManager.catalog(catalogVersion);
+
+ int zoneId = zoneId();
+
+ CatalogZoneDescriptor zoneDescriptor = catalog.zone(zoneId);
+
+ Set<Integer> allZonePartitionsToReset = new HashSet<>();
+
request.partitionIds().values().forEach(allZonePartitionsToReset::addAll);
Review Comment:
How can this work correctly for per-table partitions? If for table 1 we need
to reset partitions 11 and 12 and for table 2 partitions 21 and 22, this will
contain 11, 12, 21, 22, but information about owning tables will be lost
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
