You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sa...@apache.org on 2024/12/10 07:12:57 UTC
(ignite-3) branch main updated: IGNITE-22036 Replace TableManager.changePeersOnRebalance by the broadcast ReplicaRequest (#4446)
This is an automated email from the ASF dual-hosted git repository.
sanpwc 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 5217133385 IGNITE-22036 Replace TableManager.changePeersOnRebalance by the broadcast ReplicaRequest (#4446)
5217133385 is described below
commit 521713338518487c820329ff4a82d0188751aa7e
Author: Mikhail Efremov <ja...@gmail.com>
AuthorDate: Tue Dec 10 13:12:52 2024 +0600
IGNITE-22036 Replace TableManager.changePeersOnRebalance by the broadcast ReplicaRequest (#4446)
---
.../RebalanceRaftGroupEventsListener.java | 50 +----
.../replicator/ItReplicaLifecycleTest.java | 8 +-
.../network/PartitionReplicationMessageGroup.java | 6 +
.../ChangePeersAndLearnersAsyncReplicaRequest.java | 31 +++
.../placementdriver/TestPlacementDriver.java | 25 ++-
.../placementdriver/TestReplicaMetaImpl.java | 39 +++-
.../ItPlacementDriverReplicaSideTest.java | 4 +-
.../raft/client/TopologyAwareRaftGroupService.java | 25 ++-
.../apache/ignite/internal/replicator/Replica.java | 2 +-
.../ignite/internal/replicator/ReplicaManager.java | 233 +++++++++++++++++----
.../internal/replicator/ReplicaManagerTest.java | 3 +-
.../runner/app/ItIgniteNodeRestartTest.java | 6 +-
.../org/apache/ignite/internal/app/IgniteImpl.java | 6 +-
.../ignite/distributed/ReplicaUnavailableTest.java | 4 +-
.../rebalance/ItRebalanceDistributedTest.java | 38 +++-
.../internal/table/distributed/TableManager.java | 167 +++++++--------
.../replicator/PartitionReplicaListener.java | 96 ++++++++-
.../apache/ignite/distributed/ItTxTestCluster.java | 4 +-
18 files changed, 530 insertions(+), 217 deletions(-)
diff --git a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceRaftGroupEventsListener.java b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceRaftGroupEventsListener.java
index 8beaa8522e..7308dfcea3 100644
--- a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceRaftGroupEventsListener.java
+++ b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceRaftGroupEventsListener.java
@@ -33,7 +33,6 @@ import static org.apache.ignite.internal.metastorage.dsl.Statements.iif;
import static org.apache.ignite.internal.util.CollectionUtils.difference;
import static org.apache.ignite.internal.util.CollectionUtils.intersect;
-import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
@@ -151,54 +150,7 @@ public class RebalanceRaftGroupEventsListener implements RaftGroupEventsListener
/** {@inheritDoc} */
@Override
public void onLeaderElected(long term) {
- if (!busyLock.enterBusy()) {
- return;
- }
-
- try {
- rebalanceScheduler.schedule(() -> {
- if (!busyLock.enterBusy()) {
- return;
- }
-
- try {
- rebalanceAttempts.set(0);
-
- byte[] pendingAssignmentsBytes = metaStorageMgr.get(pendingPartAssignmentsKey(tablePartitionId)).get().value();
-
- if (pendingAssignmentsBytes != null) {
- Set<Assignment> pendingAssignments = Assignments.fromBytes(pendingAssignmentsBytes).nodes();
-
- var peers = new HashSet<String>();
- var learners = new HashSet<String>();
-
- for (Assignment assignment : pendingAssignments) {
- if (assignment.isPeer()) {
- peers.add(assignment.consistentId());
- } else {
- learners.add(assignment.consistentId());
- }
- }
-
- LOG.info(
- "New leader elected. Going to apply new configuration [tablePartitionId={}, peers={}, learners={}]",
- tablePartitionId, peers, learners
- );
-
- PeersAndLearners peersAndLearners = PeersAndLearners.fromConsistentIds(peers, learners);
-
- partitionMover.movePartition(peersAndLearners, term).get();
- }
- } catch (Exception e) {
- // TODO: IGNITE-14693
- LOG.warn("Unable to start rebalance [tablePartitionId, term={}]", e, tablePartitionId, term);
- } finally {
- busyLock.leaveBusy();
- }
- }, 0, TimeUnit.MILLISECONDS);
- } finally {
- busyLock.leaveBusy();
- }
+ // no-op
}
/** {@inheritDoc} */
diff --git a/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItReplicaLifecycleTest.java b/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItReplicaLifecycleTest.java
index 816b570478..f4e561ccfd 100644
--- a/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItReplicaLifecycleTest.java
+++ b/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItReplicaLifecycleTest.java
@@ -29,6 +29,7 @@ import static org.apache.ignite.internal.distributionzones.DistributionZonesTest
import static org.apache.ignite.internal.distributionzones.DistributionZonesTestUtil.assertValueInStorage;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.REBALANCE_SCHEDULER_POOL_SIZE;
import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.STABLE_ASSIGNMENTS_PREFIX;
+import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.pendingPartAssignmentsKey;
import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.stablePartAssignmentsKey;
import static org.apache.ignite.internal.partition.replicator.PartitionReplicaLifecycleManager.FEATURE_FLAG_NAME;
import static org.apache.ignite.internal.partitiondistribution.PartitionDistributionUtils.calculateAssignmentForPartition;
@@ -127,6 +128,7 @@ import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.lowwatermark.LowWatermarkImpl;
import org.apache.ignite.internal.manager.ComponentContext;
import org.apache.ignite.internal.manager.IgniteComponent;
+import org.apache.ignite.internal.metastorage.Entry;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.metastorage.configuration.MetaStorageConfiguration;
import org.apache.ignite.internal.metastorage.dsl.Condition;
@@ -1253,8 +1255,10 @@ public class ItReplicaLifecycleTest extends BaseIgniteAbstractTest {
partitionRaftConfigurer,
view -> new LocalLogStorageFactory(),
ForkJoinPool.commonPool(),
- t -> converter.get().apply(t)
- );
+ t -> converter.get().apply(t),
+ replicaGrpId -> metaStorageManager.get(pendingPartAssignmentsKey((ZonePartitionId) replicaGrpId))
+ .thenApply(Entry::value)
+ );
LongSupplier delayDurationMsSupplier = () -> 10L;
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/PartitionReplicationMessageGroup.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/PartitionReplicationMessageGroup.java
index 1399e74388..626ff37c2b 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/PartitionReplicationMessageGroup.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/PartitionReplicationMessageGroup.java
@@ -43,6 +43,7 @@ import org.apache.ignite.internal.partition.replicator.network.raft.SnapshotTxDa
import org.apache.ignite.internal.partition.replicator.network.replication.BinaryRowMessage;
import org.apache.ignite.internal.partition.replicator.network.replication.BinaryTupleMessage;
import org.apache.ignite.internal.partition.replicator.network.replication.BuildIndexReplicaRequest;
+import org.apache.ignite.internal.partition.replicator.network.replication.ChangePeersAndLearnersAsyncReplicaRequest;
import org.apache.ignite.internal.partition.replicator.network.replication.GetEstimatedSizeRequest;
import org.apache.ignite.internal.partition.replicator.network.replication.ReadOnlyDirectMultiRowReplicaRequest;
import org.apache.ignite.internal.partition.replicator.network.replication.ReadOnlyDirectSingleRowReplicaRequest;
@@ -204,6 +205,11 @@ public interface PartitionReplicationMessageGroup {
*/
short PARTITION_SNAPSHOT_META = 27;
+ /**
+ * Message type for {@link ChangePeersAndLearnersAsyncReplicaRequest}.
+ */
+ short CHANGE_PEERS_AND_LEARNERS_ASYNC_REPLICA_REQUEST = 28;
+
/**
* Message types for partition replicator module RAFT commands.
*
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ChangePeersAndLearnersAsyncReplicaRequest.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ChangePeersAndLearnersAsyncReplicaRequest.java
new file mode 100644
index 0000000000..901b684638
--- /dev/null
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ChangePeersAndLearnersAsyncReplicaRequest.java
@@ -0,0 +1,31 @@
+/*
+ * 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.partition.replicator.network.replication;
+
+import org.apache.ignite.internal.network.annotations.Transferable;
+import org.apache.ignite.internal.partition.replicator.network.PartitionReplicationMessageGroup;
+import org.apache.ignite.internal.replicator.message.PrimaryReplicaRequest;
+
+/**
+ * Request for performing {@link org.apache.ignite.internal.raft.service.RaftGroupService#changePeersAndLearnersAsync} on a primary replica.
+ */
+@Transferable(PartitionReplicationMessageGroup.CHANGE_PEERS_AND_LEARNERS_ASYNC_REPLICA_REQUEST)
+public interface ChangePeersAndLearnersAsyncReplicaRequest extends PrimaryReplicaRequest {
+ /** New peers configuration to rebalance. */
+ byte[] pendingAssignments();
+}
diff --git a/modules/placement-driver-api/src/testFixtures/java/org/apache/ignite/internal/placementdriver/TestPlacementDriver.java b/modules/placement-driver-api/src/testFixtures/java/org/apache/ignite/internal/placementdriver/TestPlacementDriver.java
index bff99cd020..aea47c60cf 100644
--- a/modules/placement-driver-api/src/testFixtures/java/org/apache/ignite/internal/placementdriver/TestPlacementDriver.java
+++ b/modules/placement-driver-api/src/testFixtures/java/org/apache/ignite/internal/placementdriver/TestPlacementDriver.java
@@ -112,8 +112,31 @@ public class TestPlacementDriver extends AbstractEventProducer<PrimaryReplicaEve
return this.primaryReplicaSupplier;
}
- public void setPrimaryReplicaSupplier(Supplier<? extends ReplicaMeta> primaryReplicaSupplier) {
+ /**
+ * Setter for a test primary replica supplier with {@code PRIMARY_REPLICA_ELECTED} event firing that is crucial for some tests internal
+ * logic that depends on the event handling.
+ *
+ * @param primaryReplicaSupplier The supplier that provides {@link TestReplicaMetaImpl} instance with a test primary replica meta
+ * information.
+ */
+ public void setPrimaryReplicaSupplier(Supplier<? extends TestReplicaMetaImpl> primaryReplicaSupplier) {
this.primaryReplicaSupplier = primaryReplicaSupplier;
+
+ TestReplicaMetaImpl replicaMeta = primaryReplicaSupplier.get();
+
+ fireEvent(
+ PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED,
+ new PrimaryReplicaEventParameters(
+ // The only usage of causality token below is IndexBuildController that doesn't use in tests, so the actual value
+ // doesn't matter there yet.
+ 0,
+ replicaMeta.getReplicationGroupId(),
+ replicaMeta.getLeaseholderId(),
+ replicaMeta.getLeaseholder(),
+ replicaMeta.getStartTime()
+ )
+ );
+
}
@Override
diff --git a/modules/placement-driver-api/src/testFixtures/java/org/apache/ignite/internal/placementdriver/TestReplicaMetaImpl.java b/modules/placement-driver-api/src/testFixtures/java/org/apache/ignite/internal/placementdriver/TestReplicaMetaImpl.java
index 7f44b878ba..ca93a1ed88 100644
--- a/modules/placement-driver-api/src/testFixtures/java/org/apache/ignite/internal/placementdriver/TestReplicaMetaImpl.java
+++ b/modules/placement-driver-api/src/testFixtures/java/org/apache/ignite/internal/placementdriver/TestReplicaMetaImpl.java
@@ -23,6 +23,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import java.util.UUID;
import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.network.ClusterNode;
import org.jetbrains.annotations.Nullable;
import org.jetbrains.annotations.TestOnly;
@@ -44,11 +45,15 @@ public class TestReplicaMetaImpl implements ReplicaMeta {
/** Timestamp to expiration the lease. */
private final HybridTimestamp expirationTime;
+ /** Group ID of a primary replica. */
+ private final ReplicationGroupId replicationGroupId;
+
/**
* Creates a new primary meta with unbounded period.
*
- * <p>Notes: Delegates creation to a {@link TestReplicaMetaImpl#TestReplicaMetaImpl(String, UUID, HybridTimestamp, HybridTimestamp)},
- * where {@code leaseholder} is {@link ClusterNode#name()} and {@code leaseholderId} is {@link ClusterNode#id()}.</p>
+ * <p>Notes: Delegates creation to a {@link TestReplicaMetaImpl#TestReplicaMetaImpl(String, UUID, HybridTimestamp, HybridTimestamp,
+ * ReplicationGroupId)}, where {@code leaseholder} is {@link ClusterNode#name()} and {@code leaseholderId} is {@link ClusterNode#id()}.
+ * The replication group is being set to {@code null} value.</p>
*
* @param leaseholder Lease holder, {@code null} if nothing holds the lease.
*/
@@ -63,14 +68,26 @@ public class TestReplicaMetaImpl implements ReplicaMeta {
* @param leaseholderId Lease holder ID, {@code null} if nothing holds the lease.
*/
public TestReplicaMetaImpl(@Nullable String leaseholder, @Nullable UUID leaseholderId) {
- this(leaseholder, leaseholderId, MIN_VALUE, MAX_VALUE);
+ this(leaseholder, leaseholderId, MIN_VALUE, MAX_VALUE, null);
+ }
+
+ /**
+ * Creates a new primary meta with unbounded period.
+ *
+ * @param leaseholder Lease holder consistent ID, {@code null} if nothing holds the lease.
+ * @param leaseholderId Lease holder ID, {@code null} if nothing holds the lease.
+ * @param replicationGroupId ID of the replication group for which one the lease is granted.
+ */
+ public TestReplicaMetaImpl(@Nullable String leaseholder, @Nullable UUID leaseholderId, ReplicationGroupId replicationGroupId) {
+ this(leaseholder, leaseholderId, MIN_VALUE, MAX_VALUE, replicationGroupId);
}
/**
* Creates a new primary meta.
*
- * <p>Notes: Delegates creation to a {@link TestReplicaMetaImpl#TestReplicaMetaImpl(String, UUID, HybridTimestamp, HybridTimestamp)},
- * where {@code leaseholder} is {@link ClusterNode#name()} and {@code leaseholderId} is {@link ClusterNode#id()}.</p>
+ * <p>Notes: Delegates creation to a {@link TestReplicaMetaImpl#TestReplicaMetaImpl(String, UUID, HybridTimestamp, HybridTimestamp,
+ * ReplicationGroupId)}, where {@code leaseholder} is {@link ClusterNode#name()} and {@code leaseholderId} is {@link ClusterNode#id()}.
+ * The replication group is being set to {@code null} value.</p>
*
* @param leaseholder Lease holder, {@code null} if nothing holds the lease.
* @param startTime Start lease timestamp.
@@ -81,7 +98,8 @@ public class TestReplicaMetaImpl implements ReplicaMeta {
leaseholder == null ? null : leaseholder.name(),
leaseholder == null ? null : leaseholder.id(),
startTime,
- expirationTime
+ expirationTime,
+ null
);
}
@@ -92,12 +110,14 @@ public class TestReplicaMetaImpl implements ReplicaMeta {
* @param leaseholderId Lease holder ID, {@code null} if nothing holds the lease.
* @param startTime Start lease timestamp.
* @param expirationTime Lease expiration timestamp.
+ * @param replicationGroupId ID of the replication group for which one the lease is granted.
*/
private TestReplicaMetaImpl(
@Nullable String leaseholder,
@Nullable UUID leaseholderId,
HybridTimestamp startTime,
- HybridTimestamp expirationTime
+ HybridTimestamp expirationTime,
+ ReplicationGroupId replicationGroupId
) {
assertEquals(
leaseholder == null,
@@ -109,6 +129,7 @@ public class TestReplicaMetaImpl implements ReplicaMeta {
this.leaseholderId = leaseholderId;
this.startTime = startTime;
this.expirationTime = expirationTime;
+ this.replicationGroupId = replicationGroupId;
}
@Override
@@ -130,4 +151,8 @@ public class TestReplicaMetaImpl implements ReplicaMeta {
public HybridTimestamp getExpirationTime() {
return expirationTime;
}
+
+ public ReplicationGroupId getReplicationGroupId() {
+ return replicationGroupId;
+ }
}
diff --git a/modules/replicator/src/integrationTest/java/org/apache/ignite/internal/replicator/ItPlacementDriverReplicaSideTest.java b/modules/replicator/src/integrationTest/java/org/apache/ignite/internal/replicator/ItPlacementDriverReplicaSideTest.java
index 1868b5b143..693d57bfb0 100644
--- a/modules/replicator/src/integrationTest/java/org/apache/ignite/internal/replicator/ItPlacementDriverReplicaSideTest.java
+++ b/modules/replicator/src/integrationTest/java/org/apache/ignite/internal/replicator/ItPlacementDriverReplicaSideTest.java
@@ -27,6 +27,7 @@ import static org.apache.ignite.internal.testframework.IgniteTestUtils.testNodeN
import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.apache.ignite.internal.util.CollectionUtils.first;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.IgniteUtils.closeAll;
import static org.apache.ignite.internal.util.IgniteUtils.startAsync;
import static org.apache.ignite.internal.util.IgniteUtils.stopAsync;
@@ -228,7 +229,8 @@ public class ItPlacementDriverReplicaSideTest extends IgniteAbstractTest {
raftManager,
partitionsConfigurer,
new VolatileLogStorageFactoryCreator(nodeName, workDir.resolve("volatile-log-spillout")),
- ForkJoinPool.commonPool()
+ ForkJoinPool.commonPool(),
+ replicaGrpId -> nullCompletedFuture()
);
replicaManagers.put(nodeName, replicaManager);
diff --git a/modules/replicator/src/main/java/org/apache/ignite/internal/raft/client/TopologyAwareRaftGroupService.java b/modules/replicator/src/main/java/org/apache/ignite/internal/raft/client/TopologyAwareRaftGroupService.java
index 3415f4da8f..62a35d7c80 100644
--- a/modules/replicator/src/main/java/org/apache/ignite/internal/raft/client/TopologyAwareRaftGroupService.java
+++ b/modules/replicator/src/main/java/org/apache/ignite/internal/raft/client/TopologyAwareRaftGroupService.java
@@ -305,14 +305,17 @@ public class TopologyAwareRaftGroupService implements RaftGroupService {
* @return Future that is completed when all subscription messages to peers are sent.
*/
public CompletableFuture<Void> subscribeLeader(LeaderElectionListener callback) {
- assert !serverEventHandler.isSubscribed() : "The node already subscribed";
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-23863 do the refactoring to make the code more consistent.
+ if (serverEventHandler.isSubscribed()) {
+ eventsClientListener.addLeaderElectionListener(groupId(), callback);
+ } else {
+ serverEventHandler.setOnLeaderElectedCallback(callback);
+ }
int peers = peers().size();
var futs = new CompletableFuture[peers];
- serverEventHandler.setOnLeaderElectedCallback(callback);
-
for (int i = 0; i < peers; i++) {
Peer peer = peers().get(i);
@@ -340,7 +343,8 @@ public class TopologyAwareRaftGroupService implements RaftGroupService {
ClusterNode leaderHost = clusterService.topologyService().getByConsistentId(leaderWithTerm.leader().consistentId());
if (leaderHost != null) {
- serverEventHandler.onLeaderElected(
+ eventsClientListener.onLeaderElected(
+ groupId(),
leaderHost,
leaderWithTerm.term()
);
@@ -364,6 +368,19 @@ public class TopologyAwareRaftGroupService implements RaftGroupService {
serverEventHandler.setOnLeaderElectedCallback(null);
serverEventHandler.resetLeader();
+ return sendUnsubscribeLeaderMessageAndClearSubscribersMap();
+ }
+
+ /**
+ * Removes the given local callback on RAFT leader elected event from handling list.
+ *
+ * @param callback The callback that should be removed.
+ */
+ public void unsubscribeLeader(LeaderElectionListener callback) {
+ eventsClientListener.removeLeaderElectionListener(groupId(), callback);
+ }
+
+ private CompletableFuture<Void> sendUnsubscribeLeaderMessageAndClearSubscribersMap() {
var peers = peers();
List<CompletableFuture<Boolean>> futs = new ArrayList<>();
diff --git a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/Replica.java b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/Replica.java
index 006ea880dd..c9c478dacf 100644
--- a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/Replica.java
+++ b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/Replica.java
@@ -35,7 +35,7 @@ public interface Replica {
*
* @return RAFT client.
*/
- // TODO:https://issues.apache.org/jira/browse/IGNITE-22036 remove this method
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-23750
@Deprecated(forRemoval = true)
TopologyAwareRaftGroupService raftClient();
diff --git a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java
index ab36f262d9..be83ed32dc 100644
--- a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java
+++ b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java
@@ -22,6 +22,7 @@ import static java.util.concurrent.CompletableFuture.delayedExecutor;
import static java.util.concurrent.CompletableFuture.failedFuture;
import static java.util.concurrent.CompletableFuture.supplyAsync;
import static java.util.stream.Collectors.toSet;
+import static org.apache.ignite.internal.failure.FailureType.CRITICAL_ERROR;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
import static org.apache.ignite.internal.raft.PeersAndLearners.fromAssignments;
import static org.apache.ignite.internal.replicator.LocalReplicaEvent.AFTER_REPLICA_STARTED;
@@ -37,6 +38,7 @@ import static org.apache.ignite.internal.util.CompletableFutures.isCompletedSucc
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.trueCompletedFuture;
import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
import static org.apache.ignite.internal.util.IgniteUtils.shouldSwitchToRequestsExecutor;
import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination;
import static org.apache.ignite.lang.ErrorGroups.Common.INTERNAL_ERR;
@@ -61,6 +63,7 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.LongSupplier;
import java.util.function.Supplier;
@@ -68,7 +71,6 @@ import org.apache.ignite.internal.cluster.management.ClusterManagementGroupManag
import org.apache.ignite.internal.event.AbstractEventProducer;
import org.apache.ignite.internal.failure.FailureContext;
import org.apache.ignite.internal.failure.FailureManager;
-import org.apache.ignite.internal.failure.FailureType;
import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.lang.NodeStoppingException;
@@ -88,6 +90,7 @@ import org.apache.ignite.internal.placementdriver.message.PlacementDriverMessage
import org.apache.ignite.internal.placementdriver.message.PlacementDriverMessagesFactory;
import org.apache.ignite.internal.placementdriver.message.PlacementDriverReplicaMessage;
import org.apache.ignite.internal.placementdriver.message.StopLeaseProlongationMessageResponse;
+import org.apache.ignite.internal.raft.LeaderElectionListener;
import org.apache.ignite.internal.raft.Loza;
import org.apache.ignite.internal.raft.Marshaller;
import org.apache.ignite.internal.raft.Peer;
@@ -244,6 +247,7 @@ public class ReplicaManager extends AbstractEventProducer<LocalReplicaEvent, Loc
* @param volatileLogStorageFactoryCreator Creator for {@link org.apache.ignite.internal.raft.storage.LogStorageFactory} for
* volatile tables.
* @param groupIdConverter Temporary converter to support the zone based partitions in tests.
+ * @param getPendingAssignmentsSupplier The supplier of pending assignments for rebalance failover purposes.
*/
// TODO: https://issues.apache.org/jira/browse/IGNITE-22522 remove this method
@TestOnly
@@ -263,7 +267,8 @@ public class ReplicaManager extends AbstractEventProducer<LocalReplicaEvent, Loc
RaftGroupOptionsConfigurer partitionRaftConfigurer,
LogStorageFactoryCreator volatileLogStorageFactoryCreator,
Executor replicaStartStopExecutor,
- Function<ReplicaRequest, ReplicationGroupId> groupIdConverter
+ Function<ReplicaRequest, ReplicationGroupId> groupIdConverter,
+ Function<ReplicationGroupId, CompletableFuture<byte[]>> getPendingAssignmentsSupplier
) {
this(
nodeName,
@@ -280,7 +285,8 @@ public class ReplicaManager extends AbstractEventProducer<LocalReplicaEvent, Loc
raftManager,
partitionRaftConfigurer,
volatileLogStorageFactoryCreator,
- replicaStartStopExecutor
+ replicaStartStopExecutor,
+ getPendingAssignmentsSupplier
);
this.groupIdConverter = groupIdConverter;
@@ -301,8 +307,11 @@ public class ReplicaManager extends AbstractEventProducer<LocalReplicaEvent, Loc
* @param raftCommandsMarshaller Command marshaller for raft groups creation.
* @param raftGroupServiceFactory A factory for raft-clients creation.
* @param raftManager The manager made up of songs and words to spite all my troubles is not so bad at all.
+ * @param partitionRaftConfigurer Configurer of raft options on raft group creation.
* @param volatileLogStorageFactoryCreator Creator for {@link org.apache.ignite.internal.raft.storage.LogStorageFactory} for
* volatile tables.
+ * @param replicaStartStopExecutor Executor for asynchronous replicas lifecycle management.
+ * @param getPendingAssignmentsSupplier The supplier of pending assignments for rebalance failover purposes.
*/
public ReplicaManager(
String nodeName,
@@ -319,7 +328,8 @@ public class ReplicaManager extends AbstractEventProducer<LocalReplicaEvent, Loc
RaftManager raftManager,
RaftGroupOptionsConfigurer partitionRaftConfigurer,
LogStorageFactoryCreator volatileLogStorageFactoryCreator,
- Executor replicaStartStopExecutor
+ Executor replicaStartStopExecutor,
+ Function<ReplicationGroupId, CompletableFuture<byte[]>> getPendingAssignmentsSupplier
) {
this.clusterNetSvc = clusterNetSvc;
this.cmgMgr = cmgMgr;
@@ -336,7 +346,14 @@ public class ReplicaManager extends AbstractEventProducer<LocalReplicaEvent, Loc
this.raftGroupServiceFactory = raftGroupServiceFactory;
this.raftManager = raftManager;
this.partitionRaftConfigurer = partitionRaftConfigurer;
- this.replicaStateManager = new ReplicaStateManager(replicaStartStopExecutor, clockService, placementDriver, this);
+ this.replicaStateManager = new ReplicaStateManager(
+ replicaStartStopExecutor,
+ clockService,
+ placementDriver,
+ getPendingAssignmentsSupplier,
+ failureManager,
+ this
+ );
// This pool MUST be single-threaded to make sure idle safe time propagation attempts are not reordered on it.
scheduledIdleSafeTimeSyncExecutor = Executors.newScheduledThreadPool(
@@ -954,6 +971,8 @@ public class ReplicaManager extends AbstractEventProducer<LocalReplicaEvent, Loc
return nullCompletedFuture();
}
+ replicaStateManager.stop();
+
blockBusy();
int shutdownTimeoutSeconds = 10;
@@ -1098,7 +1117,7 @@ public class ReplicaManager extends AbstractEventProducer<LocalReplicaEvent, Loc
} catch (Error e) {
LOG.error("Error while trying to send a safe time sync request [groupId={}]", e, entry.getKey());
- failureManager.process(new FailureContext(FailureType.CRITICAL_ERROR, e));
+ failureManager.process(new FailureContext(CRITICAL_ERROR, e));
}
}
}
@@ -1229,15 +1248,27 @@ public class ReplicaManager extends AbstractEventProducer<LocalReplicaEvent, Loc
volatile UUID localNodeId;
+ private final Function<ReplicationGroupId, CompletableFuture<byte[]>> getPendingAssignmentsSupplier;
+
+ private LeaderElectionListener onLeaderElectedFailoverCallback;
+
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ private final FailureManager failureManager;
+
ReplicaStateManager(
Executor replicaStartStopPool,
ClockService clockService,
PlacementDriver placementDriver,
+ Function<ReplicationGroupId, CompletableFuture<byte[]>> getPendingAssignmentsSupplier,
+ FailureManager failureManager,
ReplicaManager replicaManager
) {
this.replicaStartStopPool = replicaStartStopPool;
this.clockService = clockService;
this.placementDriver = placementDriver;
+ this.getPendingAssignmentsSupplier = getPendingAssignmentsSupplier;
+ this.failureManager = failureManager;
this.replicaManager = replicaManager;
}
@@ -1247,55 +1278,181 @@ public class ReplicaManager extends AbstractEventProducer<LocalReplicaEvent, Loc
placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_EXPIRED, this::onPrimaryExpired);
}
+ void stop() {
+ busyLock.block();
+ }
+
private CompletableFuture<Boolean> onPrimaryElected(PrimaryReplicaEventParameters parameters) {
- ReplicationGroupId groupId = parameters.groupId();
- ReplicaStateContext context = getContext(groupId);
+ // Busy lock guarding because on node stop we shouldn't handle primary replica events anymore.
+ if (!busyLock.enterBusy()) {
+ return failedFuture(new NodeStoppingException());
+ }
- synchronized (context) {
- if (localNodeId.equals(parameters.leaseholderId())) {
- assert context.replicaState != ReplicaState.STOPPED
- : "Unexpected primary replica state STOPPED [groupId=" + groupId
- + ", leaseStartTime=" + parameters.startTime() + ", reservedForPrimary=" + context.reservedForPrimary
- + ", contextLeaseStartTime=" + context.leaseStartTime + "].";
- } else if (context.reservedForPrimary) {
- context.assertReservation(groupId, parameters.startTime());
-
- // Unreserve if another replica was elected as primary, only if its lease start time is greater,
- // otherwise it means that event is too late relatively to lease negotiation start and should be ignored.
- if (parameters.startTime().compareTo(context.leaseStartTime) > 0) {
- context.unreserve();
-
- if (context.replicaState == ReplicaState.PRIMARY_ONLY) {
- executeDeferredReplicaStop(groupId, context);
+ try {
+ ReplicationGroupId replicationGroupId = parameters.groupId();
+ ReplicaStateContext context = getContext(replicationGroupId);
+
+ synchronized (context) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ assert context.replicaState != ReplicaState.STOPPED
+ : "Unexpected primary replica state STOPPED [groupId=" + replicationGroupId
+ + ", leaseStartTime=" + parameters.startTime() + ", reservedForPrimary=" + context.reservedForPrimary
+ + ", contextLeaseStartTime=" + context.leaseStartTime + "].";
+
+ registerFailoverCallback(replicationGroupId);
+ } else if (context.reservedForPrimary) {
+ context.assertReservation(replicationGroupId, parameters.startTime());
+
+ // Unreserve if another replica was elected as primary, only if its lease start time is greater,
+ // otherwise it means that event is too late relatively to lease negotiation start and should be ignored.
+ if (parameters.startTime().compareTo(context.leaseStartTime) > 0) {
+ context.unreserve();
+
+ if (context.replicaState == ReplicaState.PRIMARY_ONLY) {
+ executeDeferredReplicaStop(replicationGroupId, context);
+ }
}
}
}
- }
- return falseCompletedFuture();
+ return falseCompletedFuture();
+ } finally {
+ busyLock.leaveBusy();
+ }
}
private CompletableFuture<Boolean> onPrimaryExpired(PrimaryReplicaEventParameters parameters) {
- if (localNodeId.equals(parameters.leaseholderId())) {
- ReplicaStateContext context = replicaContexts.get(parameters.groupId());
+ // Busy lock guarding because on node stop we shouldn't handle primary replica events anymore.
+ if (!busyLock.enterBusy()) {
+ return failedFuture(new NodeStoppingException());
+ }
- if (context != null) {
- synchronized (context) {
- context.assertReservation(parameters.groupId(), parameters.startTime());
- // Unreserve if primary replica expired, only if its lease start time is equal to reservation time,
- // otherwise it means that event is too late relatively to lease negotiation start and should be ignored.
- if (parameters.startTime().equals(context.leaseStartTime)) {
- context.unreserve();
+ try {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ ReplicaStateContext context = replicaContexts.get(parameters.groupId());
+
+ if (context != null) {
+ synchronized (context) {
+ unregisterFailoverCallback(parameters);
+
+ context.assertReservation(parameters.groupId(), parameters.startTime());
+
+ // Unreserve if primary replica expired, only if its lease start time is equal to reservation time,
+ // otherwise it means that event is too late relatively to lease negotiation start and should be ignored.
+ if (parameters.startTime().equals(context.leaseStartTime)) {
+ context.unreserve();
- if (context.replicaState == ReplicaState.RESTART_PLANNED) {
- executeDeferredReplicaStop(parameters.groupId(), context);
+ if (context.replicaState == ReplicaState.RESTART_PLANNED) {
+ executeDeferredReplicaStop(parameters.groupId(), context);
+ }
}
}
}
}
+
+ return falseCompletedFuture();
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ // TODO: move to Replica https://issues.apache.org/jira/browse/IGNITE-23750
+ private void registerFailoverCallback(ReplicationGroupId replicationGroupId) {
+ CompletableFuture<Replica> replicaFuture = replicaManager.replica(replicationGroupId);
+
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-23753
+ // For now we assume that it's a possible situation to get null future on node's stopp process, but it should be fixed with
+ // a correct and consistent replica's state.
+ if (replicaFuture == null) {
+ return;
+ }
+
+ assert replicaFuture.isDone() : "Illegal state: replica future exists and elected as primary, but the future isn't completed"
+ + " [replicationGroupId=" + replicaFuture + "].";
+
+ replicaFuture.thenCompose(inBusyLock(busyLock, () -> (Function<Replica, CompletableFuture<Void>>) electedPrimaryReplica -> {
+ onLeaderElectedFailoverCallback = (leaderNode, term) -> changePeersAndLearnersAsyncIfPendingExists(
+ electedPrimaryReplica,
+ replicationGroupId,
+ term
+ );
+
+ return electedPrimaryReplica.raftClient().subscribeLeader(onLeaderElectedFailoverCallback);
+ })).exceptionally(inBusyLock(busyLock, () -> (Function<Throwable, Void>) e -> {
+ LOG.error("Rebalance failover subscription on elected primary replica failed [groupId=" + replicationGroupId + "].", e);
+
+ failureManager.process(new FailureContext(CRITICAL_ERROR, e));
+
+ return null;
+ }));
+ }
+
+ // TODO: move to Replica https://issues.apache.org/jira/browse/IGNITE-23750
+ private void unregisterFailoverCallback(PrimaryReplicaEventParameters parameters) {
+ CompletableFuture<Replica> replicaFuture = replicaManager.replica(parameters.groupId());
+
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-23753
+ // For now we assume that it's a possible situation to get null future on node's stopp process, but it should be fixed with
+ // a correct and consistent replica's state.
+ if (replicaFuture == null) {
+ return;
}
- return falseCompletedFuture();
+ assert replicaFuture.isDone() : "Illegal state: replica future exists and was elected as primary, but the future isn't"
+ + " completed [replicationGroupId=" + replicaFuture + "].";
+
+ replicaFuture.thenAccept(inBusyLock(busyLock, () -> (Consumer<Replica>) expiredPrimaryReplica -> expiredPrimaryReplica
+ .raftClient()
+ .unsubscribeLeader(onLeaderElectedFailoverCallback)
+ ));
+ }
+
+ private void changePeersAndLearnersAsyncIfPendingExists(
+ Replica primaryReplica,
+ ReplicationGroupId replicationGroupId,
+ long term
+ ) {
+ // Busy lock guarding because on node stop we shouldn't start this detached callback with the failover logic.
+ if (!busyLock.enterBusy()) {
+ return;
+ }
+
+ try {
+ getPendingAssignmentsSupplier.apply(
+ replicationGroupId
+ ).exceptionally(inBusyLock(busyLock, () -> (Function<Throwable, byte[]>) e -> {
+ LOG.error(
+ "Couldn't fetch pending assignments for rebalance failover [groupId={}, term={}].",
+ e,
+ replicationGroupId,
+ term
+ );
+
+ return null;
+ })).thenCompose(inBusyLock(busyLock, () -> (Function<byte[], CompletableFuture<Void>>) pendingsBytes -> {
+ if (pendingsBytes == null) {
+ return nullCompletedFuture();
+ }
+
+ PeersAndLearners newConfiguration = fromAssignments(Assignments.fromBytes(pendingsBytes).nodes());
+
+ LOG.info(
+ "New leader elected. Going to apply new configuration [tablePartitionId={}, peers={}, learners={}]",
+ replicationGroupId,
+ newConfiguration.peers(),
+ newConfiguration.learners()
+ );
+
+ // TODO: add retries on fail https://issues.apache.org/jira/browse/IGNITE-23633
+ return primaryReplica.raftClient().changePeersAndLearnersAsync(newConfiguration, term);
+ })).exceptionally(inBusyLock(busyLock, () -> (Function<Throwable, Void>) e -> {
+ LOG.error("Failover ChangePeersAndLearners failed [groupId={}, term={}].", e, replicationGroupId, term);
+
+ return null;
+ }));
+ } finally {
+ busyLock.leaveBusy();
+ }
}
ReplicaStateContext getContext(ReplicationGroupId groupId) {
diff --git a/modules/replicator/src/test/java/org/apache/ignite/internal/replicator/ReplicaManagerTest.java b/modules/replicator/src/test/java/org/apache/ignite/internal/replicator/ReplicaManagerTest.java
index 96f22952f7..5eacffdc3c 100644
--- a/modules/replicator/src/test/java/org/apache/ignite/internal/replicator/ReplicaManagerTest.java
+++ b/modules/replicator/src/test/java/org/apache/ignite/internal/replicator/ReplicaManagerTest.java
@@ -136,7 +136,8 @@ public class ReplicaManagerTest extends BaseIgniteAbstractTest {
raftManager,
partitionsConfigurer,
volatileLogStorageFactoryCreator,
- ForkJoinPool.commonPool()
+ ForkJoinPool.commonPool(),
+ replicaGrpId -> nullCompletedFuture()
);
assertThat(replicaManager.startAsync(new ComponentContext()), willCompleteSuccessfully());
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
index e49471c596..d768256b28 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
@@ -28,6 +28,7 @@ import static org.apache.ignite.internal.configuration.IgnitePaths.partitionsPat
import static org.apache.ignite.internal.distributionzones.DistributionZonesTestUtil.alterZone;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.REBALANCE_SCHEDULER_POOL_SIZE;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.STABLE_ASSIGNMENTS_PREFIX;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsKey;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
import static org.apache.ignite.internal.network.utils.ClusterServiceTestUtils.defaultChannelTypeRegistry;
import static org.apache.ignite.internal.network.utils.ClusterServiceTestUtils.defaultSerializationRegistry;
@@ -591,7 +592,10 @@ public class ItIgniteNodeRestartTest extends BaseIgniteRestartTest {
raftMgr,
partitionRaftConfigurer,
view -> new LocalLogStorageFactory(),
- threadPoolsManager.tableIoExecutor()
+ threadPoolsManager.tableIoExecutor(),
+ replicaGrpId -> metaStorageMgr.get(pendingPartAssignmentsKey((TablePartitionId) replicaGrpId))
+ .thenApply(Entry::value)
+
);
var resourcesRegistry = new RemotelyTriggeredResourceRegistry();
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 d2976d8b3e..daff737a0d 100644
--- a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
+++ b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
@@ -25,6 +25,7 @@ import static org.apache.ignite.internal.configuration.IgnitePaths.metastoragePa
import static org.apache.ignite.internal.configuration.IgnitePaths.partitionsPath;
import static org.apache.ignite.internal.configuration.IgnitePaths.vaultPath;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.REBALANCE_SCHEDULER_POOL_SIZE;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.pendingPartAssignmentsKey;
import static org.apache.ignite.internal.thread.ThreadOperation.STORAGE_READ;
import static org.apache.ignite.internal.thread.ThreadOperation.STORAGE_WRITE;
import static org.apache.ignite.internal.util.CompletableFutures.copyStateTo;
@@ -195,6 +196,7 @@ import org.apache.ignite.internal.raft.storage.impl.VolatileLogStorageFactoryCre
import org.apache.ignite.internal.raft.util.SharedLogStorageFactoryUtils;
import org.apache.ignite.internal.replicator.ReplicaManager;
import org.apache.ignite.internal.replicator.ReplicaService;
+import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.replicator.configuration.ReplicationConfiguration;
import org.apache.ignite.internal.replicator.configuration.ReplicationExtensionConfiguration;
import org.apache.ignite.internal.rest.RestComponent;
@@ -821,7 +823,9 @@ public class IgniteImpl implements Ignite {
raftMgr,
partitionRaftConfigurer,
volatileLogStorageFactoryCreator,
- threadPoolsManager.tableIoExecutor()
+ threadPoolsManager.tableIoExecutor(),
+ replicaGrpId -> metaStorageMgr.get(pendingPartAssignmentsKey((TablePartitionId) replicaGrpId))
+ .thenApply(org.apache.ignite.internal.metastorage.Entry::value)
);
metricManager.configure(
diff --git a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ReplicaUnavailableTest.java b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ReplicaUnavailableTest.java
index 5ad2fa9614..13218a3e8a 100644
--- a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ReplicaUnavailableTest.java
+++ b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ReplicaUnavailableTest.java
@@ -30,6 +30,7 @@ import static org.apache.ignite.internal.testframework.matchers.CompletableFutur
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willSucceedFast;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willSucceedIn;
import static org.apache.ignite.internal.util.CompletableFutures.emptySetCompletedFuture;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
import static org.apache.ignite.lang.ErrorGroups.Replicator.REPLICA_TIMEOUT_ERR;
import static org.apache.ignite.raft.jraft.test.TestUtils.getLocalAddress;
@@ -215,7 +216,8 @@ public class ReplicaUnavailableTest extends IgniteAbstractTest {
raftManager,
RaftGroupOptionsConfigurer.EMPTY,
view -> new LocalLogStorageFactory(),
- ForkJoinPool.commonPool()
+ ForkJoinPool.commonPool(),
+ replicaGrpId -> nullCompletedFuture()
);
assertThat(replicaManager.startAsync(new ComponentContext()), willCompleteSuccessfully());
diff --git a/modules/table/src/integrationTest/java/org/apache/ignite/internal/rebalance/ItRebalanceDistributedTest.java b/modules/table/src/integrationTest/java/org/apache/ignite/internal/rebalance/ItRebalanceDistributedTest.java
index 98a22c1176..41d986abb8 100644
--- a/modules/table/src/integrationTest/java/org/apache/ignite/internal/rebalance/ItRebalanceDistributedTest.java
+++ b/modules/table/src/integrationTest/java/org/apache/ignite/internal/rebalance/ItRebalanceDistributedTest.java
@@ -244,6 +244,7 @@ import org.apache.ignite.raft.jraft.rpc.RpcRequests;
import org.apache.ignite.raft.jraft.rpc.impl.RaftGroupEventsClientListener;
import org.apache.ignite.sql.IgniteSql;
import org.apache.ignite.table.Table;
+import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -974,20 +975,35 @@ public class ItRebalanceDistributedTest extends BaseIgniteAbstractTest {
return IntStream.range(0, nodes.size()).filter(i -> getNode(i).name.equals(consistentId)).findFirst().orElseThrow();
}
- private void electPrimaryReplica(Node node) {
- Set<Assignment> assignments = getPartitionClusterNodes(node, 0);
+ private void electPrimaryReplica(Node primaryReplicaNode) throws InterruptedException {
+ Node leaseholderNode = getLeaseholderNodeForPartition(primaryReplicaNode, 0);
- String leaseholderConsistentId = assignments.stream().findFirst().get().consistentId();
+ int tableId = getTableId(primaryReplicaNode, TABLE_NAME);
- ClusterNode leaseholder = nodes.stream()
- .filter(n -> n.clusterService.topologyService().localMember().name().equals(leaseholderConsistentId))
- .findFirst()
- .get()
- .clusterService
+ TablePartitionId groupId = new TablePartitionId(tableId, 0);
+
+ assertTrue(waitForCondition(() -> isReplicationGroupStarted(leaseholderNode, groupId), AWAIT_TIMEOUT_MILLIS));
+
+ ClusterNode leaseholder = leaseholderNode.clusterService
.topologyService()
.localMember();
- node.placementDriver.setPrimaryReplicaSupplier(() -> new TestReplicaMetaImpl(leaseholder.name(), leaseholder.id()));
+ nodes.forEach(node -> node.placementDriver.setPrimaryReplicaSupplier(() -> new TestReplicaMetaImpl(
+ leaseholder.name(),
+ leaseholder.id(),
+ new TablePartitionId(getTableId(node, TABLE_NAME), 0)
+ )));
+ }
+
+ private @NotNull Node getLeaseholderNodeForPartition(Node node, int partId) {
+ Set<Assignment> assignments = getPartitionClusterNodes(node, partId);
+
+ String leaseholderConsistentId = assignments.stream().findFirst().get().consistentId();
+
+ return nodes.stream()
+ .filter(n -> n.clusterService.topologyService().localMember().name().equals(leaseholderConsistentId))
+ .findFirst()
+ .get();
}
private static Set<Assignment> getPartitionClusterNodes(Node node, int partNum) {
@@ -1371,7 +1387,9 @@ public class ItRebalanceDistributedTest extends BaseIgniteAbstractTest {
raftManager,
partitionRaftConfigurer,
view -> new LocalLogStorageFactory(),
- ForkJoinPool.commonPool()
+ ForkJoinPool.commonPool(),
+ replicaGrpId -> metaStorageManager.get(pendingPartAssignmentsKey((TablePartitionId) replicaGrpId))
+ .thenApply(Entry::value)
));
LongSupplier delayDurationMsSupplier = () -> 10L;
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 5dba90e2f1..524f896fb9 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
@@ -57,11 +57,9 @@ import static org.apache.ignite.internal.util.CompletableFutures.emptyListComple
import static org.apache.ignite.internal.util.CompletableFutures.falseCompletedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.trueCompletedFuture;
-import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination;
-import static org.apache.ignite.lang.ErrorGroups.Common.INTERNAL_ERR;
import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
import java.nio.file.Path;
@@ -87,6 +85,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiFunction;
+import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.IntSupplier;
import java.util.function.Predicate;
@@ -137,17 +136,18 @@ import org.apache.ignite.internal.network.serialization.MessageSerializationRegi
import org.apache.ignite.internal.partition.replicator.LocalPartitionReplicaEvent;
import org.apache.ignite.internal.partition.replicator.LocalPartitionReplicaEventParameters;
import org.apache.ignite.internal.partition.replicator.PartitionReplicaLifecycleManager;
+import org.apache.ignite.internal.partition.replicator.network.PartitionReplicationMessagesFactory;
+import org.apache.ignite.internal.partition.replicator.network.replication.ChangePeersAndLearnersAsyncReplicaRequest;
import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.partitiondistribution.Assignments;
import org.apache.ignite.internal.partitiondistribution.PartitionDistributionUtils;
import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEvent;
import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEventParameters;
import org.apache.ignite.internal.raft.ExecutorInclinedRaftCommandRunner;
-import org.apache.ignite.internal.raft.Peer;
import org.apache.ignite.internal.raft.PeersAndLearners;
import org.apache.ignite.internal.raft.RaftGroupEventsListener;
-import org.apache.ignite.internal.raft.service.LeaderWithTerm;
import org.apache.ignite.internal.raft.service.RaftCommandRunner;
import org.apache.ignite.internal.raft.service.RaftGroupListener;
import org.apache.ignite.internal.raft.service.RaftGroupService;
@@ -160,6 +160,9 @@ import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.replicator.listener.ReplicaListener;
+import org.apache.ignite.internal.replicator.message.ReplicaMessageUtils;
+import org.apache.ignite.internal.replicator.message.ReplicaMessagesFactory;
+import org.apache.ignite.internal.replicator.message.TablePartitionIdMessage;
import org.apache.ignite.internal.schema.SchemaManager;
import org.apache.ignite.internal.schema.SchemaRegistry;
import org.apache.ignite.internal.schema.SchemaSyncService;
@@ -244,6 +247,12 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
private static final int TX_STATE_STORAGE_FLUSH_DELAY = 100;
private static final IntSupplier TX_STATE_STORAGE_FLUSH_DELAY_SUPPLIER = () -> TX_STATE_STORAGE_FLUSH_DELAY;
+ /** Replica messages factory. */
+ private static final ReplicaMessagesFactory REPLICA_MESSAGES_FACTORY = new ReplicaMessagesFactory();
+
+ /** Table messages factory. */
+ private static final PartitionReplicationMessagesFactory TABLE_MESSAGES_FACTORY = new PartitionReplicationMessagesFactory();
+
private final TopologyService topologyService;
/** Replica manager. */
@@ -1330,16 +1339,23 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
});
}
- private boolean isLocalPeer(Peer peer) {
- return peer.consistentId().equals(localNode().name());
- }
-
private boolean isLocalNodeInAssignments(Collection<Assignment> assignments) {
return assignments.stream().anyMatch(isLocalNodeAssignment);
}
+ private CompletableFuture<Boolean> isLocalNodeIsPrimary(ReplicationGroupId replicationGroupId) {
+ return isLocalNodeIsPrimary(getPrimaryReplica(replicationGroupId));
+ }
+
+ private CompletableFuture<Boolean> isLocalNodeIsPrimary(CompletableFuture<ReplicaMeta> primaryReplicaFuture) {
+ return primaryReplicaFuture.thenApply(primaryReplicaMeta -> primaryReplicaMeta != null
+ && primaryReplicaMeta.getLeaseholder() != null
+ && primaryReplicaMeta.getLeaseholder().equals(localNode().name())
+ );
+ }
+
/**
- * Checks that the local node is primary or not.
+ * Returns future with {@link ReplicaMeta} if primary is presented or null otherwise.
* <br>
* Internally we use there {@link PlacementDriver#getPrimaryReplica} with a penultimate
* safe time value, because metastore is waiting for pending or stable assignments events handling over and only then metastore will
@@ -1351,16 +1367,16 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
* successfully get primary replica for the time stamp before the handling has began. Also there a corner case for tests that are using
* {@code WatchListenerInhibitor#metastorageEventsInhibitor} and it leads to current time equals {@link HybridTimestamp#MIN_VALUE} and
* the skew's subtraction will lead to {@link IllegalArgumentException} from {@link HybridTimestamp}. Then, if we got the minimal
- * possible timestamp, then we also couldn't have any primary replica, then return {@code false}.
+ * possible timestamp, then we also couldn't have any primary replica, then return {@code null} as a future's result.
*
* @param replicationGroupId Replication group ID for that we check is the local node a primary.
- * @return {@code true} is the local node is primary and {@code false} otherwise.
+ * @return completed future with primary replica's meta or {@code null} otherwise.
*/
- private CompletableFuture<Boolean> isLocalNodeIsPrimary(ReplicationGroupId replicationGroupId) {
+ private CompletableFuture<ReplicaMeta> getPrimaryReplica(ReplicationGroupId replicationGroupId) {
HybridTimestamp currentSafeTime = metaStorageMgr.clusterTime().currentSafeTime();
if (HybridTimestamp.MIN_VALUE.equals(currentSafeTime)) {
- return falseCompletedFuture();
+ return nullCompletedFuture();
}
long skewMs = clockService.maxClockSkewMillis();
@@ -1368,10 +1384,7 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
try {
HybridTimestamp previousMetastoreSafeTime = currentSafeTime.subtractPhysicalTime(skewMs);
- return executorInclinedPlacementDriver.getPrimaryReplica(replicationGroupId, previousMetastoreSafeTime)
- .thenApply(replicaMeta -> replicaMeta != null
- && replicaMeta.getLeaseholderId() != null
- && replicaMeta.getLeaseholderId().equals(localNode().id()));
+ return executorInclinedPlacementDriver.getPrimaryReplica(replicationGroupId, previousMetastoreSafeTime);
} catch (IllegalArgumentException e) {
long currentSafeTimeMs = currentSafeTime.longValue();
@@ -2102,20 +2115,15 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
revision,
isRecovery,
assignmentsTimestamp
- ).thenCompose(v -> {
- boolean isLocalNodeInStableOrPending = isNodeInReducedStableOrPendingAssignments(
- replicaGrpId,
- stableAssignments,
- pendingAssignments,
- revision
- );
-
- if (!isLocalNodeInStableOrPending) {
- return nullCompletedFuture();
- }
-
- return changePeersOnRebalance(replicaGrpId, pendingAssignments.nodes(), revision);
- });
+ ).thenAccept(v -> executeIfLocalNodeIsPrimaryForGroup(
+ replicaGrpId,
+ replicaMeta -> sendChangePeersAndLearnersRequest(
+ replicaMeta,
+ replicaGrpId,
+ pendingAssignments,
+ revision
+ )
+ ));
} finally {
busyLock.leaveBusy();
}
@@ -2248,6 +2256,46 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
return newPartitionSet;
}
+ private void executeIfLocalNodeIsPrimaryForGroup(
+ ReplicationGroupId groupId,
+ Consumer<ReplicaMeta> toExecute
+ ) {
+ CompletableFuture<ReplicaMeta> primaryReplicaFuture = getPrimaryReplica(groupId);
+
+ isLocalNodeIsPrimary(primaryReplicaFuture).thenAccept(isPrimary -> {
+ if (isPrimary) {
+ primaryReplicaFuture.thenAccept(toExecute);
+ }
+ });
+ }
+
+ private void sendChangePeersAndLearnersRequest(
+ ReplicaMeta replicaMeta,
+ TablePartitionId replicationGroupId,
+ Assignments pendingAssignments,
+ long currentRevision
+ ) {
+ metaStorageMgr.get(pendingPartAssignmentsKey(replicationGroupId)).thenAccept(latestPendingAssignmentsEntry -> {
+ // Do not change peers of the raft group if this is a stale event.
+ // Note that we start raft node before for the sake of the consistency in a
+ // starting and stopping raft nodes.
+ if (currentRevision < latestPendingAssignmentsEntry.revision()) {
+ return;
+ }
+
+ TablePartitionIdMessage partitionIdMessage = ReplicaMessageUtils
+ .toTablePartitionIdMessage(REPLICA_MESSAGES_FACTORY, replicationGroupId);
+
+ ChangePeersAndLearnersAsyncReplicaRequest request = TABLE_MESSAGES_FACTORY.changePeersAndLearnersAsyncReplicaRequest()
+ .groupId(partitionIdMessage)
+ .pendingAssignments(pendingAssignments.toBytes())
+ .enlistmentConsistencyToken(replicaMeta.getStartTime().longValue())
+ .build();
+
+ replicaSvc.invoke(localNode(), request);
+ });
+ }
+
private boolean isNodeInReducedStableOrPendingAssignments(
TablePartitionId replicaGrpId,
@Nullable Assignments stableAssignments,
@@ -2277,63 +2325,6 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
return true;
}
- private CompletableFuture<Void> changePeersOnRebalance(
- TablePartitionId replicaGrpId,
- Set<Assignment> pendingAssignments,
- long revision
- ) {
- return replicaMgr.replica(replicaGrpId)
- .thenApply(Replica::raftClient)
- .thenCompose(raftClient -> raftClient.refreshAndGetLeaderWithTerm()
- .exceptionally(throwable -> {
- throwable = unwrapCause(throwable);
-
- if (throwable instanceof TimeoutException) {
- LOG.info(
- "Node couldn't get the leader within timeout so the changing peers is skipped [grp={}].",
- replicaGrpId
- );
-
- return LeaderWithTerm.NO_LEADER;
- }
-
- throw new IgniteInternalException(
- INTERNAL_ERR,
- "Failed to get a leader for the RAFT replication group [get=" + replicaGrpId + "].",
- throwable
- );
- })
- .thenCompose(leaderWithTerm -> {
- if (leaderWithTerm.isEmpty() || !isLocalPeer(leaderWithTerm.leader())) {
- return nullCompletedFuture();
- }
-
- // run update of raft configuration if this node is a leader
- LOG.info("Current node={} is the leader of partition raft group={}. "
- + "Initiate rebalance process for partition={}, table={}",
- leaderWithTerm.leader(),
- replicaGrpId,
- replicaGrpId.partitionId(),
- tables.get(replicaGrpId.tableId()).name()
- );
-
- return metaStorageMgr.get(pendingPartAssignmentsKey(replicaGrpId))
- .thenCompose(latestPendingAssignmentsEntry -> {
- // Do not change peers of the raft group if this is a stale event.
- // Note that we start raft node before for the sake of the consistency in a
- // starting and stopping raft nodes.
- if (revision < latestPendingAssignmentsEntry.revision()) {
- return nullCompletedFuture();
- }
-
- PeersAndLearners newConfiguration = fromAssignments(pendingAssignments);
-
- return raftClient.changePeersAndLearnersAsync(newConfiguration, leaderWithTerm.term());
- });
- })
- );
- }
-
private SnapshotStorageFactory createSnapshotStorageFactory(
TablePartitionId replicaGrpId,
PartitionUpdateHandlers partitionUpdateHandlers,
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
index 0e63c388e5..2fe68fa5fb 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
@@ -30,6 +30,8 @@ import static org.apache.ignite.internal.partition.replicator.network.replicatio
import static org.apache.ignite.internal.partition.replicator.network.replication.RequestType.RW_GET_ALL;
import static org.apache.ignite.internal.partition.replicator.network.replication.RequestType.RW_REPLACE;
import static org.apache.ignite.internal.partition.replicator.network.replication.RequestType.RW_SCAN;
+import static org.apache.ignite.internal.partitiondistribution.Assignments.fromBytes;
+import static org.apache.ignite.internal.raft.PeersAndLearners.fromAssignments;
import static org.apache.ignite.internal.replicator.message.ReplicaMessageUtils.toTablePartitionIdMessage;
import static org.apache.ignite.internal.table.distributed.index.MetaIndexStatus.BUILDING;
import static org.apache.ignite.internal.table.distributed.index.MetaIndexStatus.REGISTERED;
@@ -52,10 +54,12 @@ import static org.apache.ignite.internal.util.CompletableFutures.falseCompletedF
import static org.apache.ignite.internal.util.CompletableFutures.isCompletedSuccessfully;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.trueCompletedFuture;
+import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
import static org.apache.ignite.internal.util.IgniteUtils.findAny;
import static org.apache.ignite.internal.util.IgniteUtils.findFirst;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
+import static org.apache.ignite.lang.ErrorGroups.Common.INTERNAL_ERR;
import static org.apache.ignite.lang.ErrorGroups.Replicator.CURSOR_CLOSE_ERR;
import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_ALREADY_FINISHED_ERR;
import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_COMMIT_ERR;
@@ -114,6 +118,7 @@ import org.apache.ignite.internal.partition.replicator.network.command.WriteInte
import org.apache.ignite.internal.partition.replicator.network.replication.BinaryRowMessage;
import org.apache.ignite.internal.partition.replicator.network.replication.BinaryTupleMessage;
import org.apache.ignite.internal.partition.replicator.network.replication.BuildIndexReplicaRequest;
+import org.apache.ignite.internal.partition.replicator.network.replication.ChangePeersAndLearnersAsyncReplicaRequest;
import org.apache.ignite.internal.partition.replicator.network.replication.GetEstimatedSizeRequest;
import org.apache.ignite.internal.partition.replicator.network.replication.ReadOnlyDirectMultiRowReplicaRequest;
import org.apache.ignite.internal.partition.replicator.network.replication.ReadOnlyDirectSingleRowReplicaRequest;
@@ -131,11 +136,15 @@ import org.apache.ignite.internal.partition.replicator.network.replication.ReadW
import org.apache.ignite.internal.partition.replicator.network.replication.RequestType;
import org.apache.ignite.internal.partition.replicator.network.replication.ScanCloseReplicaRequest;
import org.apache.ignite.internal.partition.replicator.network.replication.UpdateMinimumActiveTxBeginTimeReplicaRequest;
+import org.apache.ignite.internal.partitiondistribution.Assignments;
import org.apache.ignite.internal.placementdriver.PlacementDriver;
import org.apache.ignite.internal.placementdriver.ReplicaMeta;
import org.apache.ignite.internal.raft.Command;
import org.apache.ignite.internal.raft.ExecutorInclinedRaftCommandRunner;
+import org.apache.ignite.internal.raft.PeersAndLearners;
+import org.apache.ignite.internal.raft.service.LeaderWithTerm;
import org.apache.ignite.internal.raft.service.RaftCommandRunner;
+import org.apache.ignite.internal.raft.service.RaftGroupService;
import org.apache.ignite.internal.replicator.CommandApplicationResult;
import org.apache.ignite.internal.replicator.ReplicaResult;
import org.apache.ignite.internal.replicator.TablePartitionId;
@@ -274,7 +283,7 @@ public class PartitionReplicaListener implements ReplicaListener {
private final MvPartitionStorage mvDataStorage;
/** Raft client. */
- private final RaftCommandRunner raftClient;
+ private final RaftCommandRunner raftCommandRunner;
/** Tx manager. */
private final TxManager txManager;
@@ -354,7 +363,7 @@ public class PartitionReplicaListener implements ReplicaListener {
* The constructor.
*
* @param mvDataStorage Data storage.
- * @param raftClient Raft client.
+ * @param raftCommandRunner Raft client.
* @param txManager Transaction manager.
* @param lockManager Lock manager.
* @param partId Partition id.
@@ -376,7 +385,7 @@ public class PartitionReplicaListener implements ReplicaListener {
*/
public PartitionReplicaListener(
MvPartitionStorage mvDataStorage,
- RaftCommandRunner raftClient,
+ RaftCommandRunner raftCommandRunner,
TxManager txManager,
LockManager lockManager,
Executor scanRequestExecutor,
@@ -401,7 +410,7 @@ public class PartitionReplicaListener implements ReplicaListener {
IndexMetaStorage indexMetaStorage
) {
this.mvDataStorage = mvDataStorage;
- this.raftClient = raftClient;
+ this.raftCommandRunner = raftCommandRunner;
this.txManager = txManager;
this.lockManager = lockManager;
this.scanRequestExecutor = scanRequestExecutor;
@@ -483,10 +492,10 @@ public class PartitionReplicaListener implements ReplicaListener {
/** Returns Raft-client. */
@Override
public RaftCommandRunner raftClient() {
- if (raftClient instanceof ExecutorInclinedRaftCommandRunner) {
- return ((ExecutorInclinedRaftCommandRunner) raftClient).decoratedCommandRunner();
+ if (raftCommandRunner instanceof ExecutorInclinedRaftCommandRunner) {
+ return ((ExecutorInclinedRaftCommandRunner) raftCommandRunner).decoratedCommandRunner();
}
- return raftClient;
+ return raftCommandRunner;
}
private CompletableFuture<?> processRequest(ReplicaRequest request, @Nullable Boolean isPrimary, UUID senderId,
@@ -523,6 +532,10 @@ public class PartitionReplicaListener implements ReplicaListener {
return processGetEstimatedSizeRequest();
}
+ if (request instanceof ChangePeersAndLearnersAsyncReplicaRequest) {
+ return processChangePeersAndLearnersReplicaRequest((ChangePeersAndLearnersAsyncReplicaRequest) request);
+ }
+
@Nullable HybridTimestamp opTs = getTxOpTimestamp(request);
@Nullable HybridTimestamp opTsIfDirectRo = (request instanceof ReadOnlyDirectReplicaRequest) ? opTs : null;
@Nullable HybridTimestamp txTs = getTxStartTimestamp(request);
@@ -589,6 +602,67 @@ public class PartitionReplicaListener implements ReplicaListener {
return triggerTxRecovery(txId, senderId);
}
+ private CompletableFuture<Void> processChangePeersAndLearnersReplicaRequest(ChangePeersAndLearnersAsyncReplicaRequest request) {
+ TablePartitionId replicaGrpId = (TablePartitionId) request.groupId().asReplicationGroupId();
+
+ RaftGroupService raftClient = raftCommandRunner instanceof RaftGroupService
+ ? (RaftGroupService) raftCommandRunner
+ : ((RaftGroupService) ((ExecutorInclinedRaftCommandRunner) raftCommandRunner).decoratedCommandRunner());
+
+ return raftClient.refreshAndGetLeaderWithTerm()
+ .exceptionally(throwable -> {
+ throwable = unwrapCause(throwable);
+
+ if (throwable instanceof TimeoutException) {
+ LOG.info(
+ "Node couldn't get the leader within timeout so the changing peers is skipped [grp={}].",
+ replicaGrpId
+ );
+
+ return LeaderWithTerm.NO_LEADER;
+ }
+
+ throw new IgniteInternalException(
+ INTERNAL_ERR,
+ "Failed to get a leader for the RAFT replication group [get=" + replicaGrpId + "].",
+ throwable
+ );
+ })
+ .thenCompose(leaderWithTerm -> {
+ if (leaderWithTerm.isEmpty() || !isTokenStillValidPrimary(request.enlistmentConsistencyToken())) {
+ return nullCompletedFuture();
+ }
+
+ // run update of raft configuration if this node is a leader
+ LOG.debug("Current node={} is the leader of partition raft group={}. "
+ + "Initiate rebalance process for partition={}, table={}",
+ leaderWithTerm.leader(),
+ replicaGrpId,
+ replicaGrpId.partitionId(),
+ catalogService.table(replicaGrpId.tableId(), catalogService.latestCatalogVersion()).name()
+ );
+
+ return raftClient.changePeersAndLearnersAsync(peersConfigurationFromMessage(request), leaderWithTerm.term());
+ });
+ }
+
+ private boolean isTokenStillValidPrimary(long suspectedEnlistmentConsistencyToken) {
+ HybridTimestamp currentTime = clockService.current();
+
+ ReplicaMeta meta = placementDriver.getCurrentPrimaryReplica(replicationGroupId, currentTime);
+
+ return meta != null
+ && isLocalPeer(meta.getLeaseholderId())
+ && clockService.before(currentTime, meta.getExpirationTime())
+ && suspectedEnlistmentConsistencyToken == meta.getStartTime().longValue();
+ }
+
+ private static PeersAndLearners peersConfigurationFromMessage(ChangePeersAndLearnersAsyncReplicaRequest request) {
+ Assignments pendingAssignments = fromBytes(request.pendingAssignments());
+
+ return fromAssignments(pendingAssignments.nodes());
+ }
+
/**
* Run cleanup on a node.
*
@@ -2648,7 +2722,7 @@ public class PartitionReplicaListener implements ReplicaListener {
}
int attemptsCounter0 = attemptsCounter;
- raftClient.run(cmd).whenComplete((res, ex) -> {
+ raftCommandRunner.run(cmd).whenComplete((res, ex) -> {
if (ex != null) {
if (ex instanceof SafeTimeReorderException || ex.getCause() instanceof SafeTimeReorderException) {
assert cmd instanceof SafeTimePropagatingCommand;
@@ -2914,7 +2988,7 @@ public class PartitionReplicaListener implements ReplicaListener {
* @param rowsToUpdate All {@link BinaryRow}s represented as {@link TimedBinaryRowMessage}s to be updated.
* @param catalogVersion Validated catalog version associated with given operation.
* @param leaseStartTime Lease start time.
- * @return Raft future, see {@link #applyCmdWithExceptionHandling(Command, CompletableFuture)}.
+ * @return Raft future, see {@link #applyCmdWithExceptionHandling(Command)}.
*/
private CompletableFuture<CommandApplicationResult> applyUpdateAllCommand(
ReadWriteMultiRowReplicaRequest request,
@@ -4038,7 +4112,7 @@ public class PartitionReplicaListener implements ReplicaListener {
return txRwOperationTracker.awaitCompleteTxRwOperations(registeredChangeInfo.catalogVersion())
.thenCompose(unused -> safeTime.waitFor(hybridTimestamp(buildingChangeInfo.activationTimestamp())))
- .thenCompose(unused -> raftClient.run(toBuildIndexCommand(request, buildingChangeInfo)));
+ .thenCompose(unused -> raftCommandRunner.run(toBuildIndexCommand(request, buildingChangeInfo)));
}
private List<Integer> indexIdsAtRwTxBeginTs(UUID txId) {
@@ -4068,7 +4142,7 @@ public class PartitionReplicaListener implements ReplicaListener {
.txIds(request.transactionIds())
.build();
- return raftClient.run(cmd);
+ return raftCommandRunner.run(cmd);
}
private CompletableFuture<?> processMinimumActiveTxTimeReplicaRequest(UpdateMinimumActiveTxBeginTimeReplicaRequest request) {
diff --git a/modules/table/src/testFixtures/java/org/apache/ignite/distributed/ItTxTestCluster.java b/modules/table/src/testFixtures/java/org/apache/ignite/distributed/ItTxTestCluster.java
index c620783d26..2a0638675c 100644
--- a/modules/table/src/testFixtures/java/org/apache/ignite/distributed/ItTxTestCluster.java
+++ b/modules/table/src/testFixtures/java/org/apache/ignite/distributed/ItTxTestCluster.java
@@ -28,6 +28,7 @@ import static org.apache.ignite.internal.replicator.ReplicatorConstants.DEFAULT_
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.apache.ignite.internal.util.CollectionUtils.first;
import static org.apache.ignite.internal.util.CompletableFutures.emptySetCompletedFuture;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.IgniteUtils.startAsync;
import static org.apache.ignite.internal.util.IgniteUtils.stopAsync;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -469,7 +470,8 @@ public class ItTxTestCluster {
raftSrv,
partitionRaftConfigurer,
new VolatileLogStorageFactoryCreator(nodeName, workDir.resolve("volatile-log-spillout")),
- ForkJoinPool.commonPool()
+ ForkJoinPool.commonPool(),
+ replicaGrpId -> nullCompletedFuture()
);
assertThat(replicaMgr.startAsync(new ComponentContext()), willCompleteSuccessfully());