You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by rp...@apache.org on 2025/12/26 07:24:53 UTC
(ignite-3) branch main updated: IGNITE-27458 Stop PartitionReplicaListener being ReplicaListener (#7313)
This is an automated email from the ASF dual-hosted git repository.
rpuch 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 2f74b03afd8 IGNITE-27458 Stop PartitionReplicaListener being ReplicaListener (#7313)
2f74b03afd8 is described below
commit 2f74b03afd84a20eb5b7a7f04e8d09fa1f1db2bc
Author: Roman Puchkovskiy <ro...@gmail.com>
AuthorDate: Fri Dec 26 11:24:48 2025 +0400
IGNITE-27458 Stop PartitionReplicaListener being ReplicaListener (#7313)
---
.../partition/replicator/ReplicaPrimacy.java | 7 +-
...xDistributedTestSingleNodeNoCleanupMessage.java | 29 +---
.../internal/table/distributed/TableManager.java | 2 -
.../replicator/PartitionReplicaListener.java | 37 +-----
.../PartitionReplicaListenerIndexLockingTest.java | 14 +-
...itionReplicaListenerSortedIndexLockingTest.java | 14 +-
.../replication/PartitionReplicaListenerTest.java | 148 ++++++++++-----------
.../ZonePartitionReplicaListenerTest.java | 14 +-
.../storage/InternalTableEstimatedSizeTest.java | 28 +++-
.../apache/ignite/distributed/ItTxTestCluster.java | 2 -
.../table/impl/DummyInternalTableImpl.java | 2 -
11 files changed, 124 insertions(+), 173 deletions(-)
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/ReplicaPrimacy.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/ReplicaPrimacy.java
index 83349dd3967..c2917d753da 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/ReplicaPrimacy.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/ReplicaPrimacy.java
@@ -23,6 +23,7 @@ import org.apache.ignite.internal.partition.replicator.network.replication.ReadO
import org.apache.ignite.internal.replicator.message.PrimaryReplicaRequest;
import org.apache.ignite.internal.replicator.message.ReplicaSafeTimeSyncRequest;
import org.jetbrains.annotations.Nullable;
+import org.jetbrains.annotations.VisibleForTesting;
/**
* Represents replica primacy info. Contains the following information:
@@ -54,14 +55,16 @@ public class ReplicaPrimacy {
/**
* Creates an instance representing information about the primary replica held by this node.
*/
- static ReplicaPrimacy forPrimaryReplicaRequest(long leaseStartTime) {
+ @VisibleForTesting
+ public static ReplicaPrimacy forPrimaryReplicaRequest(long leaseStartTime) {
return new ReplicaPrimacy(leaseStartTime, null);
}
/**
* Creates an instance representing information about whether this node currently holds the primary.
*/
- static ReplicaPrimacy forIsPrimary(boolean isPrimary) {
+ @VisibleForTesting
+ public static ReplicaPrimacy forIsPrimary(boolean isPrimary) {
return new ReplicaPrimacy(null, isPrimary);
}
diff --git a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNodeNoCleanupMessage.java b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNodeNoCleanupMessage.java
index 6c04a27b134..0e22ded3895 100644
--- a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNodeNoCleanupMessage.java
+++ b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNodeNoCleanupMessage.java
@@ -17,7 +17,6 @@
package org.apache.ignite.distributed;
-import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.apache.ignite.internal.replicator.ReplicatorConstants.DEFAULT_IDLE_SAFE_TIME_PROPAGATION_PERIOD_MILLISECONDS;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -28,7 +27,6 @@ import static org.mockito.Mockito.mock;
import java.util.List;
import java.util.Map;
-import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executor;
@@ -48,10 +46,8 @@ import org.apache.ignite.internal.network.InternalClusterNode;
import org.apache.ignite.internal.partition.replicator.schema.ValidationSchemasSource;
import org.apache.ignite.internal.placementdriver.PlacementDriver;
import org.apache.ignite.internal.raft.service.RaftGroupService;
-import org.apache.ignite.internal.replicator.ReplicaResult;
import org.apache.ignite.internal.replicator.ReplicaService;
import org.apache.ignite.internal.replicator.ZonePartitionId;
-import org.apache.ignite.internal.replicator.message.ReplicaRequest;
import org.apache.ignite.internal.schema.SchemaRegistry;
import org.apache.ignite.internal.schema.SchemaSyncService;
import org.apache.ignite.internal.storage.MvPartitionStorage;
@@ -64,7 +60,6 @@ import org.apache.ignite.internal.table.distributed.replicator.PartitionReplicaL
import org.apache.ignite.internal.table.distributed.replicator.TransactionStateResolver;
import org.apache.ignite.internal.table.metrics.TableMetricSource;
import org.apache.ignite.internal.tx.InternalTransaction;
-import org.apache.ignite.internal.tx.LockManager;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.internal.tx.configuration.TransactionConfiguration;
import org.apache.ignite.internal.tx.impl.HeapLockManager;
@@ -72,7 +67,6 @@ import org.apache.ignite.internal.tx.impl.RemotelyTriggeredResourceRegistry;
import org.apache.ignite.internal.tx.impl.TransactionIdGenerator;
import org.apache.ignite.internal.tx.impl.TransactionInflights;
import org.apache.ignite.internal.tx.impl.TxManagerImpl;
-import org.apache.ignite.internal.tx.message.WriteIntentSwitchReplicaRequest;
import org.apache.ignite.internal.tx.test.TestLocalRwTxCounter;
import org.apache.ignite.internal.util.Lazy;
import org.apache.ignite.internal.util.PendingComparableValuesTracker;
@@ -204,10 +198,8 @@ public class ItTxDistributedTestSingleNodeNoCleanupMessage extends TxAbstractTes
transactionStateResolver,
storageUpdateHandler,
validationSchemasSource,
- localNode,
schemaSyncService,
catalogService,
- placementDriver,
clusterNodeResolver,
resourcesRegistry,
schemaRegistry,
@@ -215,22 +207,7 @@ public class ItTxDistributedTestSingleNodeNoCleanupMessage extends TxAbstractTes
lowWatermark,
mock(FailureProcessor.class),
new TableMetricSource(QualifiedName.fromSimple("test_table"))
- ) {
- @Override
- public CompletableFuture<ReplicaResult> invoke(ReplicaRequest request, UUID senderId) {
- if (request instanceof WriteIntentSwitchReplicaRequest) {
- logger().info("Dropping cleanup request: {}", request);
-
- releaseTxLocks(
- ((WriteIntentSwitchReplicaRequest) request).txId(),
- txManager.lockManager()
- );
-
- return completedFuture(new ReplicaResult(null, null));
- }
- return super.invoke(request, senderId);
- }
- };
+ );
}
};
@@ -291,10 +268,6 @@ public class ItTxDistributedTestSingleNodeNoCleanupMessage extends TxAbstractTes
assertEquals(200., accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));
}
- private static void releaseTxLocks(UUID txId, LockManager lockManager) {
- lockManager.releaseAll(txId);
- }
-
@Override
protected int nodes() {
return 1;
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 68fbcf25b0c..18bc6f9650a 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
@@ -1077,10 +1077,8 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
transactionStateResolver,
partitionUpdateHandlers.storageUpdateHandler,
new CatalogValidationSchemasSource(catalogService, schemaManager),
- localNode(),
executorInclinedSchemaSyncService,
catalogService,
- executorInclinedPlacementDriver,
topologyService,
remotelyTriggeredResourceRegistry,
schemaManager.schemaRegistry(table.tableId()),
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 498feb1668c..6169c45def3 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
@@ -94,11 +94,9 @@ import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.lowwatermark.LowWatermark;
import org.apache.ignite.internal.network.ClusterNodeResolver;
-import org.apache.ignite.internal.network.InternalClusterNode;
import org.apache.ignite.internal.partition.replicator.FuturesCleanupResult;
import org.apache.ignite.internal.partition.replicator.ReliableCatalogVersions;
import org.apache.ignite.internal.partition.replicator.ReplicaPrimacy;
-import org.apache.ignite.internal.partition.replicator.ReplicaPrimacyEngine;
import org.apache.ignite.internal.partition.replicator.ReplicaTableProcessor;
import org.apache.ignite.internal.partition.replicator.ReplicationRaftCommandApplicator;
import org.apache.ignite.internal.partition.replicator.TableAwareReplicaRequestPreProcessor;
@@ -133,9 +131,7 @@ import org.apache.ignite.internal.partition.replicator.network.replication.ScanC
import org.apache.ignite.internal.partition.replicator.schema.ValidationSchemasSource;
import org.apache.ignite.internal.partition.replicator.schemacompat.IncompatibleSchemaVersionException;
import org.apache.ignite.internal.partition.replicator.schemacompat.SchemaCompatibilityValidator;
-import org.apache.ignite.internal.placementdriver.LeasePlacementDriver;
import org.apache.ignite.internal.raft.Command;
-import org.apache.ignite.internal.raft.ExecutorInclinedRaftCommandRunner;
import org.apache.ignite.internal.raft.service.RaftCommandRunner;
import org.apache.ignite.internal.replicator.CommandApplicationResult;
import org.apache.ignite.internal.replicator.ReplicaResult;
@@ -145,7 +141,6 @@ import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.replicator.exception.PrimaryReplicaMissException;
import org.apache.ignite.internal.replicator.exception.ReplicationException;
import org.apache.ignite.internal.replicator.exception.UnsupportedReplicaRequestException;
-import org.apache.ignite.internal.replicator.listener.ReplicaListener;
import org.apache.ignite.internal.replicator.message.ReadOnlyDirectReplicaRequest;
import org.apache.ignite.internal.replicator.message.ReplicaMessagesFactory;
import org.apache.ignite.internal.replicator.message.ReplicaRequest;
@@ -206,7 +201,7 @@ import org.jetbrains.annotations.Nullable;
import org.jetbrains.annotations.TestOnly;
/** Partition replication listener. */
-public class PartitionReplicaListener implements ReplicaListener, ReplicaTableProcessor {
+public class PartitionReplicaListener implements ReplicaTableProcessor {
/**
* NB: this listener makes writes to the underlying MV partition storage without taking the partition snapshots read lock. This causes
* the RAFT snapshots transferred to a follower being slightly inconsistent for a limited amount of time.
@@ -257,9 +252,6 @@ public class PartitionReplicaListener implements ReplicaListener, ReplicaTablePr
/** Versioned partition storage. */
private final MvPartitionStorage mvDataStorage;
- /** Raft client. */
- private final RaftCommandRunner raftCommandRunner;
-
/** Tx manager. */
private final TxManager txManager;
@@ -317,7 +309,6 @@ public class PartitionReplicaListener implements ReplicaListener, ReplicaTablePr
private final TableMetricSource metrics;
- private final ReplicaPrimacyEngine replicaPrimacyEngine;
private final TableAwareReplicaRequestPreProcessor tableAwareReplicaRequestPreProcessor;
private final ReliableCatalogVersions reliableCatalogVersions;
private final ReplicationRaftCommandApplicator raftCommandApplicator;
@@ -341,9 +332,7 @@ public class PartitionReplicaListener implements ReplicaListener, ReplicaTablePr
* @param safeTime Safe time clock.
* @param transactionStateResolver Transaction state resolver.
* @param storageUpdateHandler Handler that processes updates writing them to storage.
- * @param localNode Instance of the local node.
* @param catalogService Catalog service.
- * @param placementDriver Placement driver.
* @param clusterNodeResolver Node resolver.
* @param remotelyTriggeredResourceRegistry Resource registry.
* @param indexMetaStorage Index meta storage.
@@ -365,10 +354,8 @@ public class PartitionReplicaListener implements ReplicaListener, ReplicaTablePr
TransactionStateResolver transactionStateResolver,
StorageUpdateHandler storageUpdateHandler,
ValidationSchemasSource validationSchemasSource,
- InternalClusterNode localNode,
SchemaSyncService schemaSyncService,
CatalogService catalogService,
- LeasePlacementDriver placementDriver,
ClusterNodeResolver clusterNodeResolver,
RemotelyTriggeredResourceRegistry remotelyTriggeredResourceRegistry,
SchemaRegistry schemaRegistry,
@@ -378,7 +365,6 @@ public class PartitionReplicaListener implements ReplicaListener, ReplicaTablePr
TableMetricSource metrics
) {
this.mvDataStorage = mvDataStorage;
- this.raftCommandRunner = raftCommandRunner;
this.txManager = txManager;
this.lockManager = lockManager;
this.scanRequestExecutor = scanRequestExecutor;
@@ -399,13 +385,11 @@ public class PartitionReplicaListener implements ReplicaListener, ReplicaTablePr
this.tableLockKey = new TablePartitionId(tableId, replicationGroupId.partitionId());
this.metrics = metrics;
- this.schemaCompatValidator = new SchemaCompatibilityValidator(validationSchemasSource, catalogService, schemaSyncService);
+ schemaCompatValidator = new SchemaCompatibilityValidator(validationSchemasSource, catalogService, schemaSyncService);
indexBuildingProcessor = new PartitionReplicaBuildIndexProcessor(busyLock, tableId, indexMetaStorage, catalogService);
- replicaPrimacyEngine = new ReplicaPrimacyEngine(placementDriver, clockService, replicationGroupId, localNode);
-
- this.tableAwareReplicaRequestPreProcessor = new TableAwareReplicaRequestPreProcessor(
+ tableAwareReplicaRequestPreProcessor = new TableAwareReplicaRequestPreProcessor(
clockService,
schemaCompatValidator,
schemaSyncService
@@ -417,12 +401,6 @@ public class PartitionReplicaListener implements ReplicaListener, ReplicaTablePr
buildIndexReplicaRequestHandler = new BuildIndexReplicaRequestHandler(indexMetaStorage, raftCommandApplicator);
}
- @Override
- public CompletableFuture<ReplicaResult> invoke(ReplicaRequest request, UUID senderId) {
- return replicaPrimacyEngine.validatePrimacy(request)
- .thenCompose(replicaPrimacy -> processRequestInContext(request, replicaPrimacy, senderId));
- }
-
@Override
public CompletableFuture<ReplicaResult> process(ReplicaRequest request, ReplicaPrimacy replicaPrimacy, UUID senderId) {
return processRequestInContext(request, replicaPrimacy, senderId);
@@ -445,15 +423,6 @@ public class PartitionReplicaListener implements ReplicaListener, ReplicaTablePr
}
}
- /** Returns Raft-client. */
- @Override
- public RaftCommandRunner raftClient() {
- if (raftCommandRunner instanceof ExecutorInclinedRaftCommandRunner) {
- return ((ExecutorInclinedRaftCommandRunner) raftCommandRunner).decoratedCommandRunner();
- }
- return raftCommandRunner;
- }
-
private CompletableFuture<?> processRequest(ReplicaRequest request, ReplicaPrimacy replicaPrimacy, UUID senderId) {
boolean hasSchemaVersion = request instanceof SchemaVersionAwareReplicaRequest;
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
index 77a47bb14ef..e78d234f72f 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
@@ -72,9 +72,9 @@ import org.apache.ignite.internal.hlc.TestClockService;
import org.apache.ignite.internal.lowwatermark.TestLowWatermark;
import org.apache.ignite.internal.network.ClusterNodeResolver;
import org.apache.ignite.internal.network.InternalClusterNode;
+import org.apache.ignite.internal.partition.replicator.ReplicaPrimacy;
import org.apache.ignite.internal.partition.replicator.network.PartitionReplicationMessagesFactory;
import org.apache.ignite.internal.partition.replicator.network.replication.RequestType;
-import org.apache.ignite.internal.placementdriver.TestPlacementDriver;
import org.apache.ignite.internal.raft.service.LeaderWithTerm;
import org.apache.ignite.internal.raft.service.RaftGroupService;
import org.apache.ignite.internal.replicator.ZonePartitionId;
@@ -248,8 +248,6 @@ public class PartitionReplicaListenerIndexLockingTest extends IgniteAbstractTest
when(catalog.indexes(anyInt())).thenReturn(List.of(indexDescriptor));
- InternalClusterNode localNode = DummyInternalTableImpl.LOCAL_NODE;
-
partitionReplicaListener = new PartitionReplicaListener(
TEST_MV_PARTITION_STORAGE,
mockRaftClient,
@@ -279,10 +277,8 @@ public class PartitionReplicaListenerIndexLockingTest extends IgniteAbstractTest
TableTestUtils.NOOP_PARTITION_MODIFICATION_COUNTER
),
new DummyValidationSchemasSource(schemaManager),
- localNode,
new AlwaysSyncedSchemaSyncService(),
catalogService,
- new TestPlacementDriver(localNode),
mock(ClusterNodeResolver.class),
new RemotelyTriggeredResourceRegistry(),
schemaManager,
@@ -395,7 +391,7 @@ public class PartitionReplicaListenerIndexLockingTest extends IgniteAbstractTest
throw new AssertionError("Unexpected operation type: " + arg.type);
}
- CompletableFuture<?> fut = partitionReplicaListener.invoke(request, LOCAL_NODE_ID);
+ CompletableFuture<?> fut = partitionReplicaListener.process(request, replicaPrimacy(), LOCAL_NODE_ID);
await(fut);
@@ -422,6 +418,10 @@ public class PartitionReplicaListenerIndexLockingTest extends IgniteAbstractTest
);
}
+ private ReplicaPrimacy replicaPrimacy() {
+ return ReplicaPrimacy.forPrimaryReplicaRequest(HybridTimestamp.MIN_VALUE.longValue());
+ }
+
/** Verifies the mode in which the lock was acquired on the index key for a particular operation. */
@ParameterizedTest
@MethodSource("readWriteMultiTestArguments")
@@ -484,7 +484,7 @@ public class PartitionReplicaListenerIndexLockingTest extends IgniteAbstractTest
throw new AssertionError("Unexpected operation type: " + arg.type);
}
- CompletableFuture<?> fut = partitionReplicaListener.invoke(request, LOCAL_NODE_ID);
+ CompletableFuture<?> fut = partitionReplicaListener.process(request, replicaPrimacy(), LOCAL_NODE_ID);
await(fut);
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerSortedIndexLockingTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerSortedIndexLockingTest.java
index 8d63b89dba4..cc7fe16c6c7 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerSortedIndexLockingTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerSortedIndexLockingTest.java
@@ -73,9 +73,9 @@ import org.apache.ignite.internal.hlc.TestClockService;
import org.apache.ignite.internal.lowwatermark.TestLowWatermark;
import org.apache.ignite.internal.network.ClusterNodeResolver;
import org.apache.ignite.internal.network.InternalClusterNode;
+import org.apache.ignite.internal.partition.replicator.ReplicaPrimacy;
import org.apache.ignite.internal.partition.replicator.network.PartitionReplicationMessagesFactory;
import org.apache.ignite.internal.partition.replicator.network.replication.RequestType;
-import org.apache.ignite.internal.placementdriver.TestPlacementDriver;
import org.apache.ignite.internal.raft.service.LeaderWithTerm;
import org.apache.ignite.internal.raft.service.RaftGroupService;
import org.apache.ignite.internal.replicator.ZonePartitionId;
@@ -223,8 +223,6 @@ public class PartitionReplicaListenerSortedIndexLockingTest extends IgniteAbstra
when(catalog.indexes(anyInt())).thenReturn(List.of(indexDescriptor));
- InternalClusterNode localNode = DummyInternalTableImpl.LOCAL_NODE;
-
partitionReplicaListener = new PartitionReplicaListener(
TEST_MV_PARTITION_STORAGE,
mockRaftClient,
@@ -249,10 +247,8 @@ public class PartitionReplicaListenerSortedIndexLockingTest extends IgniteAbstra
TableTestUtils.NOOP_PARTITION_MODIFICATION_COUNTER
),
new DummyValidationSchemasSource(schemaManager),
- localNode,
new AlwaysSyncedSchemaSyncService(),
catalogService,
- new TestPlacementDriver(localNode),
mock(ClusterNodeResolver.class),
new RemotelyTriggeredResourceRegistry(),
schemaManager,
@@ -293,6 +289,10 @@ public class PartitionReplicaListenerSortedIndexLockingTest extends IgniteAbstra
return txManager;
}
+ private static ReplicaPrimacy validRwPrimacy() {
+ return ReplicaPrimacy.forPrimaryReplicaRequest(1);
+ }
+
@BeforeEach
public void beforeTest() {
((TestSortedIndexStorage) pkStorage.get().storage()).clear();
@@ -365,7 +365,7 @@ public class PartitionReplicaListenerSortedIndexLockingTest extends IgniteAbstra
throw new AssertionError("Unexpected operation type: " + arg.type);
}
- CompletableFuture<?> fut = partitionReplicaListener.invoke(request, LOCAL_NODE_ID);
+ CompletableFuture<?> fut = partitionReplicaListener.process(request, validRwPrimacy(), LOCAL_NODE_ID);
await(fut);
@@ -444,7 +444,7 @@ public class PartitionReplicaListenerSortedIndexLockingTest extends IgniteAbstra
throw new AssertionError("Unexpected operation type: " + arg.type);
}
- CompletableFuture<?> fut = partitionReplicaListener.invoke(request, LOCAL_NODE_ID);
+ CompletableFuture<?> fut = partitionReplicaListener.process(request, validRwPrimacy(), LOCAL_NODE_ID);
await(fut);
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
index dd436d029fe..262e01d38af 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
@@ -135,6 +135,8 @@ import org.apache.ignite.internal.network.InternalClusterNode;
import org.apache.ignite.internal.network.MessagingService;
import org.apache.ignite.internal.network.SingleClusterNodeResolver;
import org.apache.ignite.internal.network.TopologyService;
+import org.apache.ignite.internal.partition.replicator.ReplicaPrimacy;
+import org.apache.ignite.internal.partition.replicator.ReplicaPrimacyEngine;
import org.apache.ignite.internal.partition.replicator.network.PartitionReplicationMessagesFactory;
import org.apache.ignite.internal.partition.replicator.network.command.CatalogVersionAware;
import org.apache.ignite.internal.partition.replicator.network.command.FinishTxCommand;
@@ -473,6 +475,8 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
@Mock
private IndexMetaStorage indexMetaStorage;
+ private ReplicaPrimacyEngine primacyEngine;
+
private static UUID nodeId(int id) {
return new UUID(0, id);
}
@@ -660,10 +664,8 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
TableTestUtils.NOOP_PARTITION_MODIFICATION_COUNTER
),
validationSchemasSource,
- localNode,
schemaSyncService,
catalogService,
- placementDriver,
new SingleClusterNodeResolver(localNode),
new RemotelyTriggeredResourceRegistry(),
new DummySchemaManagerImpl(schemaDescriptor, schemaDescriptorVersion2),
@@ -678,6 +680,8 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
when(lowWatermark.tryLock(any(), any())).thenReturn(true);
+ primacyEngine = new ReplicaPrimacyEngine(placementDriver, clockService, grpId, localNode);
+
reset();
}
@@ -760,7 +764,11 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
}
private CompletableFuture<ReplicaResult> invokeListener(ReplicaRequest request) {
- return partitionReplicaListener.invoke(request, localNode.id());
+ return processWithPrimacy(request);
+ }
+
+ private static ReplicaPrimacy validRoPrimacy() {
+ return ReplicaPrimacy.forIsPrimary(true);
}
private ReadOnlySingleRowPkReplicaRequest readOnlySingleRowPkReplicaRequest(BinaryRow pk, HybridTimestamp readTimestamp) {
@@ -891,7 +899,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
UUID scanTxId = newTxId();
// Request first batch
- CompletableFuture<ReplicaResult> fut = partitionReplicaListener.invoke(
+ CompletableFuture<ReplicaResult> fut = processWithPrimacy(
TABLE_MESSAGES_FACTORY.readWriteScanRetrieveBatchReplicaRequest()
.groupId(zonePartitionIdMessage(grpId))
.tableId(TABLE_ID)
@@ -904,7 +912,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.commitPartitionId(commitPartitionId())
.coordinatorId(localNode.id())
.timestamp(clock.now())
- .build(), localNode.id());
+ .build());
List<BinaryRow> rows = (List<BinaryRow>) fut.get(1, TimeUnit.SECONDS).result();
@@ -912,7 +920,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
assertEquals(4, rows.size());
// Request second batch
- fut = partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readWriteScanRetrieveBatchReplicaRequest()
+ fut = processWithPrimacy(TABLE_MESSAGES_FACTORY.readWriteScanRetrieveBatchReplicaRequest()
.groupId(zonePartitionIdMessage(grpId))
.tableId(TABLE_ID)
.transactionId(scanTxId)
@@ -924,7 +932,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.commitPartitionId(commitPartitionId())
.coordinatorId(localNode.id())
.timestamp(clock.now())
- .build(), localNode.id());
+ .build());
rows = (List<BinaryRow>) fut.get(1, TimeUnit.SECONDS).result();
@@ -932,7 +940,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
assertEquals(2, rows.size());
// Request bounded.
- fut = partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readWriteScanRetrieveBatchReplicaRequest()
+ fut = processWithPrimacy(TABLE_MESSAGES_FACTORY.readWriteScanRetrieveBatchReplicaRequest()
.groupId(zonePartitionIdMessage(grpId))
.tableId(TABLE_ID)
.transactionId(newTxId())
@@ -947,7 +955,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.commitPartitionId(commitPartitionId())
.coordinatorId(localNode.id())
.timestamp(clock.now())
- .build(), localNode.id());
+ .build());
rows = (List<BinaryRow>) fut.get(1, TimeUnit.SECONDS).result();
@@ -955,7 +963,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
assertEquals(2, rows.size());
// Empty result.
- fut = partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readWriteScanRetrieveBatchReplicaRequest()
+ fut = processWithPrimacy(TABLE_MESSAGES_FACTORY.readWriteScanRetrieveBatchReplicaRequest()
.groupId(zonePartitionIdMessage(grpId))
.tableId(TABLE_ID)
.transactionId(newTxId())
@@ -968,7 +976,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.commitPartitionId(commitPartitionId())
.coordinatorId(localNode.id())
.timestamp(clock.now())
- .build(), localNode.id());
+ .build());
rows = (List<BinaryRow>) fut.get(1, TimeUnit.SECONDS).result();
@@ -976,7 +984,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
assertEquals(0, rows.size());
// Lookup.
- fut = partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readWriteScanRetrieveBatchReplicaRequest()
+ fut = processWithPrimacy(TABLE_MESSAGES_FACTORY.readWriteScanRetrieveBatchReplicaRequest()
.groupId(zonePartitionIdMessage(grpId))
.tableId(TABLE_ID)
.transactionId(newTxId())
@@ -989,7 +997,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.commitPartitionId(commitPartitionId())
.coordinatorId(localNode.id())
.timestamp(clock.now())
- .build(), localNode.id());
+ .build());
rows = (List<BinaryRow>) fut.get(1, TimeUnit.SECONDS).result();
@@ -1019,7 +1027,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
UUID scanTxId = newTxId();
// Request first batch
- CompletableFuture<ReplicaResult> fut = partitionReplicaListener.invoke(
+ CompletableFuture<ReplicaResult> fut = partitionReplicaListener.process(
TABLE_MESSAGES_FACTORY.readOnlyScanRetrieveBatchReplicaRequest()
.groupId(zonePartitionIdMessage(grpId))
.tableId(TABLE_ID)
@@ -1029,7 +1037,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.indexToUse(sortedIndexId)
.batchSize(4)
.coordinatorId(localNode.id())
- .build(), localNode.id());
+ .build(), validRoPrimacy(), localNode.id());
List<BinaryRow> rows = (List<BinaryRow>) fut.get(1, TimeUnit.SECONDS).result();
@@ -1037,7 +1045,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
assertEquals(4, rows.size());
// Request second batch
- fut = partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readOnlyScanRetrieveBatchReplicaRequest()
+ fut = partitionReplicaListener.process(TABLE_MESSAGES_FACTORY.readOnlyScanRetrieveBatchReplicaRequest()
.groupId(zonePartitionIdMessage(grpId))
.tableId(TABLE_ID)
.transactionId(scanTxId)
@@ -1046,7 +1054,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.indexToUse(sortedIndexId)
.batchSize(4)
.coordinatorId(localNode.id())
- .build(), localNode.id());
+ .build(), validRoPrimacy(), localNode.id());
rows = (List<BinaryRow>) fut.get(1, TimeUnit.SECONDS).result();
@@ -1054,7 +1062,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
assertEquals(2, rows.size());
// Request bounded.
- fut = partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readOnlyScanRetrieveBatchReplicaRequest()
+ fut = partitionReplicaListener.process(TABLE_MESSAGES_FACTORY.readOnlyScanRetrieveBatchReplicaRequest()
.groupId(zonePartitionIdMessage(grpId))
.tableId(TABLE_ID)
.transactionId(newTxId())
@@ -1066,7 +1074,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.flags(SortedIndexStorage.LESS_OR_EQUAL)
.batchSize(5)
.coordinatorId(localNode.id())
- .build(), localNode.id());
+ .build(), validRoPrimacy(), localNode.id());
rows = (List<BinaryRow>) fut.get(1, TimeUnit.SECONDS).result();
@@ -1074,7 +1082,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
assertEquals(2, rows.size());
// Empty result.
- fut = partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readOnlyScanRetrieveBatchReplicaRequest()
+ fut = partitionReplicaListener.process(TABLE_MESSAGES_FACTORY.readOnlyScanRetrieveBatchReplicaRequest()
.groupId(zonePartitionIdMessage(grpId))
.tableId(TABLE_ID)
.transactionId(newTxId())
@@ -1084,7 +1092,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.lowerBoundPrefix(toIndexBound(5))
.batchSize(5)
.coordinatorId(localNode.id())
- .build(), localNode.id());
+ .build(), validRoPrimacy(), localNode.id());
rows = (List<BinaryRow>) fut.get(1, TimeUnit.SECONDS).result();
@@ -1092,7 +1100,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
assertEquals(0, rows.size());
// Lookup.
- fut = partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readOnlyScanRetrieveBatchReplicaRequest()
+ fut = partitionReplicaListener.process(TABLE_MESSAGES_FACTORY.readOnlyScanRetrieveBatchReplicaRequest()
.groupId(zonePartitionIdMessage(grpId))
.tableId(TABLE_ID)
.transactionId(newTxId())
@@ -1102,7 +1110,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.exactKey(toIndexKey(0))
.batchSize(5)
.coordinatorId(localNode.id())
- .build(), localNode.id());
+ .build(), validRoPrimacy(), localNode.id());
rows = (List<BinaryRow>) fut.get(1, TimeUnit.SECONDS).result();
@@ -1132,7 +1140,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
UUID scanTxId = newTxId();
// Request first batch
- CompletableFuture<ReplicaResult> fut = partitionReplicaListener.invoke(
+ CompletableFuture<ReplicaResult> fut = partitionReplicaListener.process(
TABLE_MESSAGES_FACTORY.readOnlyScanRetrieveBatchReplicaRequest()
.groupId(zonePartitionIdMessage(grpId))
.tableId(TABLE_ID)
@@ -1143,7 +1151,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.exactKey(toIndexKey(0))
.batchSize(3)
.coordinatorId(localNode.id())
- .build(), localNode.id());
+ .build(), validRoPrimacy(), localNode.id());
List<BinaryRow> rows = (List<BinaryRow>) fut.get(1, TimeUnit.SECONDS).result();
@@ -1151,7 +1159,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
assertEquals(3, rows.size());
// Request second batch
- fut = partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readOnlyScanRetrieveBatchReplicaRequest()
+ fut = partitionReplicaListener.process(TABLE_MESSAGES_FACTORY.readOnlyScanRetrieveBatchReplicaRequest()
.groupId(zonePartitionIdMessage(grpId))
.tableId(TABLE_ID)
.transactionId(scanTxId)
@@ -1161,7 +1169,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.exactKey(toIndexKey(0))
.batchSize(1)
.coordinatorId(localNode.id())
- .build(), localNode.id());
+ .build(), validRoPrimacy(), localNode.id());
rows = (List<BinaryRow>) fut.get(1, TimeUnit.SECONDS).result();
@@ -1169,7 +1177,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
assertEquals(1, rows.size());
// Empty result.
- fut = partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readOnlyScanRetrieveBatchReplicaRequest()
+ fut = partitionReplicaListener.process(TABLE_MESSAGES_FACTORY.readOnlyScanRetrieveBatchReplicaRequest()
.groupId(zonePartitionIdMessage(grpId))
.tableId(TABLE_ID)
.transactionId(newTxId())
@@ -1179,7 +1187,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.exactKey(toIndexKey(5))
.batchSize(5)
.coordinatorId(localNode.id())
- .build(), localNode.id());
+ .build(), validRoPrimacy(), localNode.id());
rows = (List<BinaryRow>) fut.get(1, TimeUnit.SECONDS).result();
@@ -1187,7 +1195,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
assertEquals(0, rows.size());
// Lookup.
- fut = partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readOnlyScanRetrieveBatchReplicaRequest()
+ fut = partitionReplicaListener.process(TABLE_MESSAGES_FACTORY.readOnlyScanRetrieveBatchReplicaRequest()
.groupId(zonePartitionIdMessage(grpId))
.tableId(TABLE_ID)
.transactionId(newTxId())
@@ -1197,7 +1205,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.exactKey(toIndexKey(1))
.batchSize(5)
.coordinatorId(localNode.id())
- .build(), localNode.id());
+ .build(), validRoPrimacy(), localNode.id());
rows = (List<BinaryRow>) fut.get(1, TimeUnit.SECONDS).result();
@@ -1343,7 +1351,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
boolean full,
@Nullable String txLabel
) {
- return partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readWriteSingleRowReplicaRequest()
+ return processWithPrimacy(TABLE_MESSAGES_FACTORY.readWriteSingleRowReplicaRequest()
.groupId(zonePartitionIdMessage(grpId))
.tableId(TABLE_ID)
.transactionId(txId)
@@ -1356,8 +1364,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.full(full)
.timestamp(clock.now())
.txLabel(txLabel)
- .build(),
- localNode.id()
+ .build()
);
}
@@ -1376,7 +1383,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
boolean full,
@Nullable String txLabel
) {
- return partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readWriteSingleRowPkReplicaRequest()
+ return processWithPrimacy(TABLE_MESSAGES_FACTORY.readWriteSingleRowPkReplicaRequest()
.groupId(zonePartitionIdMessage(grpId))
.tableId(TABLE_ID)
.transactionId(txId)
@@ -1389,8 +1396,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.full(full)
.timestamp(clock.now())
.txLabel(txLabel)
- .build(),
- localNode.id()
+ .build()
);
}
@@ -1420,7 +1426,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
boolean full,
@Nullable String txLabel
) {
- return partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readWriteMultiRowReplicaRequest()
+ return processWithPrimacy(TABLE_MESSAGES_FACTORY.readWriteMultiRowReplicaRequest()
.groupId(zonePartitionIdMessage(grpId))
.tableId(TABLE_ID)
.transactionId(txId)
@@ -1433,8 +1439,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.full(full)
.timestamp(clock.now())
.txLabel(txLabel)
- .build(),
- localNode.id()
+ .build()
);
}
@@ -1457,7 +1462,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
boolean full,
@Nullable String txLabel
) {
- return partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readWriteMultiRowPkReplicaRequest()
+ return processWithPrimacy(TABLE_MESSAGES_FACTORY.readWriteMultiRowPkReplicaRequest()
.groupId(zonePartitionIdMessage(grpId))
.tableId(TABLE_ID)
.transactionId(txId)
@@ -1470,8 +1475,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.full(full)
.timestamp(clock.now())
.txLabel(txLabel)
- .build(),
- localNode.id()
+ .build()
);
}
@@ -1785,7 +1789,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
}
private CompletableFuture<?> doReplaceRequest(UUID targetTxId, BinaryRow oldRow, BinaryRow newRow, boolean full) {
- return partitionReplicaListener.invoke(TABLE_MESSAGES_FACTORY.readWriteSwapRowReplicaRequest()
+ return processWithPrimacy(TABLE_MESSAGES_FACTORY.readWriteSwapRowReplicaRequest()
.groupId(zonePartitionIdMessage(grpId))
.tableId(TABLE_ID)
.transactionId(targetTxId)
@@ -1798,15 +1802,14 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.coordinatorId(localNode.id())
.full(full)
.timestamp(clock.now())
- .build(),
- localNode.id()
+ .build()
);
}
@Test
public void failsWhenScanByExactMatchReadsTupleWithIncompatibleSchemaFromFuture() {
testFailsWhenReadingFromFutureIncompatibleSchema(
- (targetTxId, key) -> partitionReplicaListener.invoke(
+ (targetTxId, key) -> processWithPrimacy(
TABLE_MESSAGES_FACTORY.readWriteScanRetrieveBatchReplicaRequest()
.groupId(zonePartitionIdMessage(grpId))
.tableId(TABLE_ID)
@@ -1819,8 +1822,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.commitPartitionId(commitPartitionId())
.coordinatorId(localNode.id())
.timestamp(clock.now())
- .build(),
- localNode.id()
+ .build()
)
);
}
@@ -1828,7 +1830,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
@Test
public void failsWhenScanByIndexReadsTupleWithIncompatibleSchemaFromFuture() {
testFailsWhenReadingFromFutureIncompatibleSchema(
- (targetTxId, key) -> partitionReplicaListener.invoke(
+ (targetTxId, key) -> processWithPrimacy(
TABLE_MESSAGES_FACTORY.readWriteScanRetrieveBatchReplicaRequest()
.groupId(zonePartitionIdMessage(grpId))
.tableId(TABLE_ID)
@@ -1840,8 +1842,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.commitPartitionId(commitPartitionId())
.coordinatorId(localNode.id())
.timestamp(clock.now())
- .build(),
- localNode.id()
+ .build()
)
);
}
@@ -1928,7 +1929,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
}
private CompletableFuture<?> doRwScanRetrieveBatchRequest(UUID targetTxId) {
- return partitionReplicaListener.invoke(
+ return processWithPrimacy(
TABLE_MESSAGES_FACTORY.readWriteScanRetrieveBatchReplicaRequest()
.groupId(zonePartitionIdMessage(grpId))
.tableId(TABLE_ID)
@@ -1940,24 +1941,19 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.commitPartitionId(commitPartitionId())
.coordinatorId(localNode.id())
.timestamp(clock.now())
- .build(),
- localNode.id()
+ .build()
);
}
private CompletableFuture<?> doRwScanCloseRequest(UUID targetTxId) {
- ZonePartitionIdMessage serializedMsg =
- zonePartitionIdMessage(new ZonePartitionId(tableDescriptor.zoneId(), grpId.partitionId()));
-
- return partitionReplicaListener.invoke(
+ return processWithPrimacy(
TABLE_MESSAGES_FACTORY.scanCloseReplicaRequest()
- .groupId(serializedMsg)
+ .groupId(zonePartitionIdMessage(new ZonePartitionId(tableDescriptor.zoneId(), grpId.partitionId())))
.tableId(grpId.zoneId())
.transactionId(targetTxId)
.timestamp(beginTimestamp(targetTxId))
.scanId(1)
- .build(),
- localNode.id()
+ .build()
);
}
@@ -2313,7 +2309,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.timestamp(clock.now())
.build();
- return partitionReplicaListener.invoke(message, localNode.id());
+ return processWithPrimacy(message);
}
private void delete(UUID txId, BinaryRow row) {
@@ -2330,7 +2326,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.timestamp(clock.now())
.build();
- assertThat(partitionReplicaListener.invoke(message, localNode.id()), willCompleteSuccessfully());
+ assertThat(processWithPrimacy(message), willCompleteSuccessfully());
}
private BinaryRow roGet(BinaryRow row, HybridTimestamp readTimestamp) {
@@ -2344,7 +2340,8 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
private CompletableFuture<BinaryRow> roGetAsync(BinaryRow row, HybridTimestamp readTimestamp) {
ReadOnlySingleRowPkReplicaRequest message = readOnlySingleRowPkReplicaRequest(row, readTimestamp);
- return partitionReplicaListener.invoke(message, localNode.id()).thenApply(replicaResult -> (BinaryRow) replicaResult.result());
+ return partitionReplicaListener.process(message, validRoPrimacy(), localNode.id())
+ .thenApply(replicaResult -> (BinaryRow) replicaResult.result());
}
private List<BinaryRow> roGetAll(Collection<BinaryRow> rows, HybridTimestamp readTimestamp) {
@@ -2743,7 +2740,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
PrimaryReplicaRequest request = mock(requestClass);
- assertThat(partitionReplicaListener.invoke(request, localNode.id()), willThrow(PrimaryReplicaMissException.class));
+ assertThat(processWithPrimacy(request), willThrow(PrimaryReplicaMissException.class));
}
@ParameterizedTest
@@ -2757,7 +2754,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
PrimaryReplicaRequest request = mock(requestClass);
when(request.enlistmentConsistencyToken()).thenReturn(leaseStartTime - 1000);
- assertThat(partitionReplicaListener.invoke(request, localNode.id()), willThrow(PrimaryReplicaMissException.class));
+ assertThat(processWithPrimacy(request), willThrow(PrimaryReplicaMissException.class));
}
@ParameterizedTest
@@ -2771,21 +2768,12 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
PrimaryReplicaRequest request = mock(requestClass);
when(request.enlistmentConsistencyToken()).thenReturn(leaseStartTime);
- assertThat(partitionReplicaListener.invoke(request, localNode.id()), willThrow(PrimaryReplicaMissException.class));
+ assertThat(processWithPrimacy(request), willThrow(PrimaryReplicaMissException.class));
}
- @ParameterizedTest
- @ValueSource(classes = {PrimaryReplicaRequest.class, TxStateCommitPartitionRequest.class})
- void primaryReplicaRequestsAreRejectedWhenLeaseholderIsDifferent(Class<? extends PrimaryReplicaRequest> requestClass) {
- long leaseStartTime = clock.nowLong();
- placementDriver.setPrimaryReplicaSupplier(
- () -> new TestReplicaMetaImpl(anotherNode, hybridTimestamp(leaseStartTime), HybridTimestamp.MAX_VALUE)
- );
-
- PrimaryReplicaRequest request = mock(requestClass);
- when(request.enlistmentConsistencyToken()).thenReturn(leaseStartTime);
-
- assertThat(partitionReplicaListener.invoke(request, localNode.id()), willThrow(PrimaryReplicaMissException.class));
+ private CompletableFuture<ReplicaResult> processWithPrimacy(ReplicaRequest request) {
+ return primacyEngine.validatePrimacy(request)
+ .thenCompose(primacy -> partitionReplicaListener.process(request, primacy, localNode.id()));
}
private static class RequestContext {
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/ZonePartitionReplicaListenerTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/ZonePartitionReplicaListenerTest.java
index f6a6185d4f5..89b21369687 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/ZonePartitionReplicaListenerTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/ZonePartitionReplicaListenerTest.java
@@ -116,6 +116,7 @@ import org.apache.ignite.internal.network.InternalClusterNode;
import org.apache.ignite.internal.network.MessagingService;
import org.apache.ignite.internal.network.SingleClusterNodeResolver;
import org.apache.ignite.internal.network.TopologyService;
+import org.apache.ignite.internal.partition.replicator.ReplicaPrimacyEngine;
import org.apache.ignite.internal.partition.replicator.ZonePartitionReplicaListener;
import org.apache.ignite.internal.partition.replicator.network.PartitionReplicationMessagesFactory;
import org.apache.ignite.internal.partition.replicator.network.command.FinishTxCommand;
@@ -452,6 +453,8 @@ public class ZonePartitionReplicaListenerTest extends IgniteAbstractTest {
@Mock
private IndexMetaStorage indexMetaStorage;
+ private ReplicaPrimacyEngine primacyEngine;
+
private static UUID nodeId(int id) {
return new UUID(0, id);
}
@@ -658,10 +661,8 @@ public class ZonePartitionReplicaListenerTest extends IgniteAbstractTest {
TableTestUtils.NOOP_PARTITION_MODIFICATION_COUNTER
),
validationSchemasSource,
- localNode,
schemaSyncService,
catalogService,
- placementDriver,
new SingleClusterNodeResolver(localNode),
new RemotelyTriggeredResourceRegistry(),
new DummySchemaManagerImpl(schemaDescriptor, schemaDescriptorVersion2),
@@ -675,6 +676,8 @@ public class ZonePartitionReplicaListenerTest extends IgniteAbstractTest {
when(lowWatermark.tryLock(any(), any())).thenReturn(true);
+ primacyEngine = new ReplicaPrimacyEngine(placementDriver, clockService, grpId, localNode);
+
reset();
}
@@ -1881,7 +1884,12 @@ public class ZonePartitionReplicaListenerTest extends IgniteAbstractTest {
.timestamp(clock.now())
.build();
- return tableReplicaProcessor.invoke(message, localNode.id());
+ return processWithPrimacy(message);
+ }
+
+ private CompletableFuture<ReplicaResult> processWithPrimacy(ReplicaRequest request) {
+ return primacyEngine.validatePrimacy(request)
+ .thenCompose(primacy -> tableReplicaProcessor.process(request, primacy, localNode.id()));
}
private CompletableFuture<ReplicaResult> doReadOnlyMultiGet(Collection<BinaryRow> rows, HybridTimestamp readTimestamp) {
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/storage/InternalTableEstimatedSizeTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/storage/InternalTableEstimatedSizeTest.java
index 8a46eae550f..01e81855b8f 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/storage/InternalTableEstimatedSizeTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/storage/InternalTableEstimatedSizeTest.java
@@ -44,6 +44,7 @@ import static org.mockito.Mockito.when;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -74,12 +75,12 @@ import org.apache.ignite.internal.network.ClusterService;
import org.apache.ignite.internal.network.InternalClusterNode;
import org.apache.ignite.internal.network.MessagingService;
import org.apache.ignite.internal.network.StaticNodeFinder;
+import org.apache.ignite.internal.partition.replicator.ReplicaPrimacyEngine;
import org.apache.ignite.internal.partition.replicator.schema.ValidationSchemasSource;
import org.apache.ignite.internal.placementdriver.PlacementDriver;
import org.apache.ignite.internal.placementdriver.leases.Lease;
import org.apache.ignite.internal.raft.Command;
import org.apache.ignite.internal.raft.service.RaftCommandRunner;
-import org.apache.ignite.internal.replicator.PartitionGroupId;
import org.apache.ignite.internal.replicator.ReplicaService;
import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.ZonePartitionId;
@@ -250,13 +251,30 @@ public class InternalTableEstimatedSizeTest extends BaseIgniteAbstractTest {
))
.collect(toList());
+ Map<ZonePartitionId, ReplicaPrimacyEngine> primacyEngines = new HashMap<>();
+ for (int partitionIndex = 0; partitionIndex < PARTITIONS_NUM; partitionIndex++) {
+ var zonePartitionId = new ZonePartitionId(ZONE_ID, partitionIndex);
+
+ var primacyEngine = new ReplicaPrimacyEngine(
+ placementDriver,
+ clockService,
+ zonePartitionId,
+ node
+ );
+
+ primacyEngines.put(zonePartitionId, primacyEngine);
+ }
+
lenient().doAnswer(invocation -> {
ReplicaRequest request = invocation.getArgument(1);
- var tablePartitionId = (PartitionGroupId) request.groupId().asReplicationGroupId();
+ var zonePartitionId = (ZonePartitionId) request.groupId().asReplicationGroupId();
- return partitionReplicaListeners.get(tablePartitionId.partitionId())
- .invoke(request, node.id())
+ return primacyEngines.get(zonePartitionId)
+ .validatePrimacy(request)
+ .thenCompose(
+ primacy -> partitionReplicaListeners.get(zonePartitionId.partitionId()).process(request, primacy, node.id())
+ )
.thenApply(replicaResult -> new ReplicaMessagesFactory()
.replicaResponse()
.result(replicaResult.result())
@@ -321,10 +339,8 @@ public class InternalTableEstimatedSizeTest extends BaseIgniteAbstractTest {
transactionStateResolver,
storageUpdateHandler,
validationSchemasSource,
- node,
schemaSyncService,
catalogService,
- placementDriver,
clusterNodeResolver,
remotelyTriggeredResourceRegistry,
schemaRegistry,
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 2ec4dea377d..ae79a299b63 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
@@ -1058,10 +1058,8 @@ public class ItTxTestCluster {
transactionStateResolver,
storageUpdateHandler,
validationSchemasSource,
- localNode,
schemaSyncService,
catalogService,
- placementDriver,
clusterNodeResolver,
resourcesRegistry,
schemaRegistry,
diff --git a/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java b/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
index 0caf6b7f99d..f4306d09da0 100644
--- a/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
+++ b/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
@@ -486,10 +486,8 @@ public class DummyInternalTableImpl extends InternalTableImpl {
transactionStateResolver,
storageUpdateHandler,
new DummyValidationSchemasSource(schemaManager),
- LOCAL_NODE,
new AlwaysSyncedSchemaSyncService(),
catalogService,
- placementDriver,
mock(ClusterNodeResolver.class),
resourcesRegistry,
schemaManager,