You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sk...@apache.org on 2025/12/08 15:36:00 UTC
(ignite-3) branch main updated: IGNITE-27175 Remove code related to non-colocated case (transactions module) (#7136)
This is an automated email from the ASF dual-hosted git repository.
sk0x50 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 1659454e042 IGNITE-27175 Remove code related to non-colocated case (transactions module) (#7136)
1659454e042 is described below
commit 1659454e04233dc560d3607c594c4d69817b1d6c
Author: Slava Koptilin <sl...@gmail.com>
AuthorDate: Mon Dec 8 17:35:54 2025 +0200
IGNITE-27175 Remove code related to non-colocated case (transactions module) (#7136)
---
.../tx/ClientTransactionCommitRequest.java | 56 ++++++---
.../ignite/client/fakes/FakeInternalTable.java | 4 +-
.../apache/ignite/client/fakes/FakeTxManager.java | 21 ++--
.../partition/replicator/TxRecoveryEngine.java | 8 +-
.../handlers/TxCleanupRecoveryRequestHandler.java | 6 +-
.../handlers/TxFinishReplicaRequestHandler.java | 35 ++----
.../handlers/TxRecoveryMessageHandler.java | 6 +-
.../network/PartitionReplicationMessageGroup.java | 3 +
.../network/command/FinishTxCommandV1.java | 1 +
.../network/command/WriteIntentSwitchCommand.java | 1 +
.../replicator/raft/RaftTxFinishMarker.java | 4 +-
.../raft/handlers/FinishTxCommandHandler.java | 18 +--
.../TxCleanupRecoveryRequestHandlerTest.java | 24 ++--
.../runner/app/ItIgniteNodeRestartTest.java | 1 -
.../org/apache/ignite/internal/app/IgniteImpl.java | 1 -
.../sql/engine/exec/ScannableTableImpl.java | 4 +-
.../internal/sql/engine/exec/TxAttributes.java | 3 +-
.../sql/engine/framework/NoOpTransaction.java | 7 +-
.../ItInternalTableReadWriteScanTest.java | 9 +-
.../ignite/internal/table/ItColocationTest.java | 7 +-
.../ignite/internal/table/InternalTable.java | 4 +-
.../apache/ignite/internal/table/TxContext.java | 10 +-
.../internal/table/distributed/TableManager.java | 2 +-
.../replicator/TransactionStateResolver.java | 16 +--
.../distributed/storage/InternalTableImpl.java | 71 ++++++------
.../PartitionRaftCommandsSerializationTest.java | 10 +-
.../distributed/storage/InternalTableImplTest.java | 10 +-
.../apache/ignite/distributed/ItTxTestCluster.java | 3 -
.../disaster/ItDisasterRecoveryManagerTest.java | 7 +-
.../internal/tx/ItTransactionMetricsTest.java | 9 +-
.../tx/readonly/ItReadOnlyTxInPastTest.java | 9 +-
.../ItTransactionPrimaryChangeTest.java | 8 +-
.../tx/distributed/ItTransactionRecoveryTest.java | 129 +++++++--------------
.../tx/distributed/ItTxResourcesVacuumTest.java | 51 +++-----
.../TransactionExpirationRegistryBenchmark.java | 11 +-
.../ignite/internal/tx/InternalTransaction.java | 10 +-
.../org/apache/ignite/internal/tx/TxManager.java | 13 +--
.../ignite/internal/tx/TxMetaSerializer.java | 62 ++++------
.../org/apache/ignite/internal/tx/TxStateMeta.java | 19 ++-
.../ignite/internal/tx/TxStateMetaAbandoned.java | 10 +-
.../ignite/internal/tx/TxStateMetaFinishing.java | 4 +-
.../internal/tx/impl/EnlistedPartitionGroup.java | 16 +--
.../ignite/internal/tx/impl/OrphanDetector.java | 4 +-
.../tx/impl/PersistentTxStateVacuumizer.java | 12 +-
.../internal/tx/impl/PlacementDriverHelper.java | 38 +++---
.../tx/impl/PrimaryReplicaExpiredException.java | 4 +-
.../tx/impl/PublicApiThreadingTransaction.java | 10 +-
.../tx/impl/ReadOnlyImplicitTransactionImpl.java | 11 +-
.../internal/tx/impl/ReadOnlyTransactionImpl.java | 11 +-
.../internal/tx/impl/ReadWriteTransactionImpl.java | 36 ++----
.../tx/impl/RemoteReadWriteTransaction.java | 7 +-
.../internal/tx/impl/TransactionInflights.java | 18 +--
.../internal/tx/impl/TxCleanupRequestHandler.java | 26 ++---
.../internal/tx/impl/TxCleanupRequestSender.java | 36 +++---
.../ignite/internal/tx/impl/TxManagerImpl.java | 79 ++++---------
.../ignite/internal/tx/impl/TxMessageSender.java | 36 +++---
.../tx/impl/VolatileTxStateMetaStorage.java | 12 +-
.../internal/tx/message/CleanupReplicatedInfo.java | 14 +--
.../tx/message/CleanupReplicatedInfoMessage.java | 12 +-
.../tx/message/EnlistedPartitionGroupMessage.java | 4 +-
.../TableWriteIntentSwitchReplicaRequest.java | 3 +-
.../tx/message/TxFinishReplicaRequest.java | 6 +-
.../tx/message/TxStateMetaAbandonedMessage.java | 4 +-
.../tx/message/TxStateMetaFinishingMessage.java | 4 +-
.../internal/tx/message/TxStateMetaMessage.java | 6 +-
.../message/WriteIntentSwitchReplicatedInfo.java | 14 +--
.../TxStateMetaRocksDbPartitionStorage.java | 55 +++------
.../rocksdb/TxStateRocksDbPartitionStorage.java | 45 ++-----
.../state/rocksdb/TxStateRocksDbSharedStorage.java | 26 ++---
.../state/rocksdb/TxStateRocksDbStorage.java | 14 +--
.../apache/ignite/internal/tx/TxCleanupTest.java | 3 +-
.../apache/ignite/internal/tx/TxManagerTest.java | 17 ++-
.../ignite/internal/tx/TxMetaSerializerTest.java | 48 ++------
.../internal/tx/impl/OrphanDetectorTest.java | 33 +++---
.../tx/impl/ReadWriteTransactionImplTest.java | 7 +-
...ransactionExpirationRegistryConcurrentTest.java | 12 +-
.../RocksDbTxStatePartitionStorageTest.java | 14 +--
.../TxStateMetaRocksDbPartitionStorageTest.java | 34 +-----
.../rocksdb/TxStateRocksDbSharedStorageTest.java | 8 +-
.../test/TestTxStatePartitionStorageTest.java | 2 +-
.../state/AbstractTxStatePartitionStorageTest.java | 62 +++++-----
.../internal/tx/test/ItTransactionTestUtils.java | 9 +-
82 files changed, 594 insertions(+), 905 deletions(-)
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionCommitRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionCommitRequest.java
index 05c26aaf347..161281673eb 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionCommitRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionCommitRequest.java
@@ -30,9 +30,7 @@ import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.hlc.HybridTimestampTracker;
import org.apache.ignite.internal.lang.IgniteInternalCheckedException;
-import org.apache.ignite.internal.lang.IgniteTuple3;
-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.table.IgniteTablesInternal;
import org.apache.ignite.internal.table.InternalTable;
import org.apache.ignite.internal.table.TableViewInternal;
@@ -72,14 +70,14 @@ public class ClientTransactionCommitRequest {
if (enableDirectMapping && !tx.isReadOnly()) {
int cnt = in.unpackInt(); // Number of direct enlistments.
- List<IgniteTuple3<TablePartitionId, String, Long>> list = new ArrayList<>();
+ List<Enlistment> list = new ArrayList<>();
for (int i = 0; i < cnt; i++) {
int tableId = in.unpackInt();
int partId = in.unpackInt();
String consistentId = in.unpackString();
long token = in.unpackLong();
- list.add(new IgniteTuple3<>(new TablePartitionId(tableId, partId), consistentId, token));
+ list.add(new Enlistment(tableId, partId, consistentId, token));
}
if (cnt > 0) {
@@ -91,30 +89,30 @@ public class ClientTransactionCommitRequest {
Exception ex = null;
- for (IgniteTuple3<TablePartitionId, String, Long> enlistment : list) {
- TableViewInternal table = igniteTables.cachedTable(enlistment.get1().tableId());
+ for (Enlistment enlistment : list) {
+ TableViewInternal table = igniteTables.cachedTable(enlistment.tableId());
if (table == null) {
- ex = new TransactionException(TX_COMMIT_ERR, "Table not found [id=" + enlistment.get1().tableId() + ']');
+ ex = new TransactionException(TX_COMMIT_ERR, "Table not found [id=" + enlistment.tableId() + ']');
break;
}
- if (!merge(table.internalTable(), enlistment.get1().partitionId(), enlistment.get2(), enlistment.get3(), tx, true)) {
- ex = new TransactionException(TX_COMMIT_ERR, "Invalid enlistment token [id=" + enlistment.get1().tableId() + ']');
+ if (!merge(table.internalTable(), enlistment.partitionId(), enlistment.consistentId(), enlistment.token(), tx, true)) {
+ ex = new TransactionException(TX_COMMIT_ERR, "Invalid enlistment token [id=" + enlistment.tableId() + ']');
break;
}
}
if (ex != null) {
// Perform enlistment for rollback.
- for (IgniteTuple3<TablePartitionId, String, Long> enlistment : list) {
- TableViewInternal table = igniteTables.cachedTable(enlistment.get1().tableId());
+ for (Enlistment enlistment : list) {
+ TableViewInternal table = igniteTables.cachedTable(enlistment.tableId());
if (table == null) {
continue;
}
- merge(table.internalTable(), enlistment.get1().partitionId(), enlistment.get2(), enlistment.get3(), tx, false);
+ merge(table.internalTable(), enlistment.partitionId(), enlistment.consistentId(), enlistment.token(), tx, false);
}
Exception finalEx = ex;
@@ -161,7 +159,7 @@ public class ClientTransactionCommitRequest {
* @return {@code True} if merged.
*/
public static boolean merge(InternalTable table, int partId, String consistentId, long token, InternalTransaction tx, boolean commit) {
- ReplicationGroupId replicationGroupId = table.targetReplicationGroupId(partId);
+ ZonePartitionId replicationGroupId = table.targetReplicationGroupId(partId);
PendingTxPartitionEnlistment existing = tx.enlistedPartition(replicationGroupId);
if (existing == null) {
tx.enlist(replicationGroupId, table.tableId(), consistentId, token);
@@ -172,4 +170,34 @@ public class ClientTransactionCommitRequest {
return true;
}
+
+ private static class Enlistment {
+ final int tableId;
+ final int partitionId;
+ final String consistentId;
+ final long token;
+
+ Enlistment(int tableId, int partitionId, String consistentId, long token) {
+ this.tableId = tableId;
+ this.partitionId = partitionId;
+ this.consistentId = consistentId;
+ this.token = token;
+ }
+
+ int tableId() {
+ return tableId;
+ }
+
+ int partitionId() {
+ return partitionId;
+ }
+
+ String consistentId() {
+ return consistentId;
+ }
+
+ long token() {
+ return token;
+ }
+ }
}
diff --git a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeInternalTable.java b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeInternalTable.java
index a7d0ac8319c..26a31e393dd 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeInternalTable.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeInternalTable.java
@@ -48,7 +48,7 @@ import org.apache.ignite.internal.lang.IgniteInternalException;
import org.apache.ignite.internal.network.ClusterNodeImpl;
import org.apache.ignite.internal.network.InternalClusterNode;
import org.apache.ignite.internal.placementdriver.ReplicaMeta;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.schema.BinaryRowEx;
import org.apache.ignite.internal.schema.BinaryTuple;
@@ -503,7 +503,7 @@ public class FakeInternalTable implements InternalTable, StreamerReceiverRunner
}
@Override
- public ReplicationGroupId targetReplicationGroupId(int partId) {
+ public ZonePartitionId targetReplicationGroupId(int partId) {
return null; // Not supported yet.
}
diff --git a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeTxManager.java b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeTxManager.java
index 80f5c1993aa..88a5d8b1618 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeTxManager.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeTxManager.java
@@ -29,7 +29,6 @@ import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.hlc.HybridTimestampTracker;
import org.apache.ignite.internal.manager.ComponentContext;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.tx.InternalTxOptions;
@@ -88,7 +87,7 @@ public class FakeTxManager implements TxManager {
}
@Override
- public PendingTxPartitionEnlistment enlistedPartition(ReplicationGroupId replicationGroupId) {
+ public PendingTxPartitionEnlistment enlistedPartition(ZonePartitionId replicationGroupId) {
return null;
}
@@ -98,18 +97,18 @@ public class FakeTxManager implements TxManager {
}
@Override
- public boolean assignCommitPartition(ReplicationGroupId replicationGroupId) {
+ public boolean assignCommitPartition(ZonePartitionId replicationGroupId) {
return false;
}
@Override
- public ReplicationGroupId commitPartition() {
+ public ZonePartitionId commitPartition() {
return null;
}
@Override
public void enlist(
- ReplicationGroupId replicationGroupId,
+ ZonePartitionId replicationGroupId,
int tableId,
String primaryNodeConsistentId,
long consistencyToken
@@ -219,11 +218,11 @@ public class FakeTxManager implements TxManager {
@Override
public CompletableFuture<Void> finish(
HybridTimestampTracker timestampTracker,
- ReplicationGroupId commitPartition,
+ ZonePartitionId commitPartition,
boolean commitIntent,
boolean timeoutExceeded,
boolean recovery,
- Map<ReplicationGroupId, PendingTxPartitionEnlistment> enlistedGroups,
+ Map<ZonePartitionId, PendingTxPartitionEnlistment> enlistedGroups,
UUID txId
) {
return nullCompletedFuture();
@@ -231,8 +230,8 @@ public class FakeTxManager implements TxManager {
@Override
public CompletableFuture<Void> cleanup(
- ReplicationGroupId commitPartitionId,
- Map<ReplicationGroupId, PartitionEnlistment> enlistedPartitions,
+ ZonePartitionId commitPartitionId,
+ Map<ZonePartitionId, PartitionEnlistment> enlistedPartitions,
boolean commit,
@Nullable HybridTimestamp commitTimestamp,
UUID txId
@@ -242,7 +241,7 @@ public class FakeTxManager implements TxManager {
@Override
public CompletableFuture<Void> cleanup(
- ReplicationGroupId commitPartitionId,
+ ZonePartitionId commitPartitionId,
Collection<EnlistedPartitionGroup> enlistedPartitions,
boolean commit,
@Nullable HybridTimestamp commitTimestamp,
@@ -252,7 +251,7 @@ public class FakeTxManager implements TxManager {
}
@Override
- public CompletableFuture<Void> cleanup(ReplicationGroupId commitPartitionId, String node, UUID txId) {
+ public CompletableFuture<Void> cleanup(ZonePartitionId commitPartitionId, String node, UUID txId) {
return nullCompletedFuture();
}
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/TxRecoveryEngine.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/TxRecoveryEngine.java
index 0ca53daf1d0..b2d452f0794 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/TxRecoveryEngine.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/TxRecoveryEngine.java
@@ -26,7 +26,7 @@ import java.util.function.Function;
import org.apache.ignite.internal.hlc.HybridTimestampTracker;
import org.apache.ignite.internal.network.ClusterNodeResolver;
import org.apache.ignite.internal.network.InternalClusterNode;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.tx.PendingTxPartitionEnlistment;
import org.apache.ignite.internal.tx.TxManager;
@@ -37,14 +37,14 @@ public class TxRecoveryEngine {
private final TxManager txManager;
private final ClusterNodeResolver clusterNodeResolver;
- private final ReplicationGroupId replicationGroupId;
+ private final ZonePartitionId replicationGroupId;
private final Function<InternalClusterNode, PendingTxPartitionEnlistment> abandonedTxRecoveryEnlistmentFactory;
/** Constructor. */
public TxRecoveryEngine(
TxManager txManager,
ClusterNodeResolver clusterNodeResolver,
- ReplicationGroupId replicationGroupId,
+ ZonePartitionId replicationGroupId,
Function<InternalClusterNode, PendingTxPartitionEnlistment> abandonedTxRecoveryEnlistmentFactory
) {
this.txManager = txManager;
@@ -83,7 +83,7 @@ public class TxRecoveryEngine {
* @param txId Transaction id.
* @param nodeId Node id (inconsistent).
*/
- public CompletableFuture<Void> runCleanupOnNode(ReplicationGroupId commitPartitionId, UUID txId, UUID nodeId) {
+ public CompletableFuture<Void> runCleanupOnNode(ZonePartitionId commitPartitionId, UUID txId, UUID nodeId) {
// Get node id of the sender to send back cleanup requests.
String nodeConsistentId = clusterNodeResolver.getConsistentIdById(nodeId);
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/TxCleanupRecoveryRequestHandler.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/TxCleanupRecoveryRequestHandler.java
index 2bca8098759..0174c0d1e1d 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/TxCleanupRecoveryRequestHandler.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/TxCleanupRecoveryRequestHandler.java
@@ -33,7 +33,7 @@ import org.apache.ignite.internal.lang.IgniteBiTuple;
import org.apache.ignite.internal.lang.IgniteInternalException;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.internal.tx.TxMeta;
import org.apache.ignite.internal.tx.message.TxCleanupRecoveryRequest;
@@ -52,14 +52,14 @@ public class TxCleanupRecoveryRequestHandler {
private final TxStatePartitionStorage txStatePartitionStorage;
private final TxManager txManager;
private final FailureProcessor failureProcessor;
- private final ReplicationGroupId replicationGroupId;
+ private final ZonePartitionId replicationGroupId;
/** Constructor. */
public TxCleanupRecoveryRequestHandler(
TxStatePartitionStorage txStatePartitionStorage,
TxManager txManager,
FailureProcessor failureProcessor,
- ReplicationGroupId replicationGroupId
+ ZonePartitionId replicationGroupId
) {
this.txStatePartitionStorage = txStatePartitionStorage;
this.txManager = txManager;
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/TxFinishReplicaRequestHandler.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/TxFinishReplicaRequestHandler.java
index 00cf323dee6..8265d0b77c0 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/TxFinishReplicaRequestHandler.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/TxFinishReplicaRequestHandler.java
@@ -20,7 +20,7 @@ package org.apache.ignite.internal.partition.replicator.handlers;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.stream.Collectors.toList;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
-import static org.apache.ignite.internal.replicator.message.ReplicaMessageUtils.toReplicationGroupIdMessage;
+import static org.apache.ignite.internal.replicator.message.ReplicaMessageUtils.toZonePartitionIdMessage;
import static org.apache.ignite.internal.tx.TxState.ABORTED;
import static org.apache.ignite.internal.tx.TxState.COMMITTED;
import static org.apache.ignite.internal.tx.TxState.isFinalState;
@@ -50,9 +50,9 @@ import org.apache.ignite.internal.partition.replicator.schema.ValidationSchemasS
import org.apache.ignite.internal.partition.replicator.schemacompat.CompatValidationResult;
import org.apache.ignite.internal.partition.replicator.schemacompat.SchemaCompatibilityValidator;
import org.apache.ignite.internal.raft.service.RaftCommandRunner;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.replicator.message.ReplicaMessagesFactory;
-import org.apache.ignite.internal.replicator.message.ReplicationGroupIdMessage;
+import org.apache.ignite.internal.replicator.message.ZonePartitionIdMessage;
import org.apache.ignite.internal.schema.SchemaSyncService;
import org.apache.ignite.internal.tx.IncompatibleSchemaAbortException;
import org.apache.ignite.internal.tx.MismatchingTransactionOutcomeInternalException;
@@ -88,7 +88,7 @@ public class TxFinishReplicaRequestHandler {
private final TxStatePartitionStorage txStatePartitionStorage;
private final ClockService clockService;
private final TxManager txManager;
- private final ReplicationGroupId replicationGroupId;
+ private final ZonePartitionId replicationGroupId;
private final SchemaCompatibilityValidator schemaCompatValidator;
private final ReliableCatalogVersions reliableCatalogVersions;
@@ -104,7 +104,7 @@ public class TxFinishReplicaRequestHandler {
SchemaSyncService schemaSyncService,
CatalogService catalogService,
RaftCommandRunner raftCommandRunner,
- ReplicationGroupId replicationGroupId
+ ZonePartitionId replicationGroupId
) {
this.txStatePartitionStorage = txStatePartitionStorage;
this.clockService = clockService;
@@ -130,8 +130,7 @@ public class TxFinishReplicaRequestHandler {
* @return future result of the operation.
*/
public CompletableFuture<TransactionResult> handle(TxFinishReplicaRequest request) {
- // TODO: https://issues.apache.org/jira/browse/IGNITE-19170 Use ZonePartitionIdMessage and remove cast
- Map<ReplicationGroupId, PartitionEnlistment> enlistedGroups = asReplicationGroupIdToPartitionMap(request.groups());
+ Map<ZonePartitionId, PartitionEnlistment> enlistedGroups = asReplicationGroupIdToPartitionMap(request.groups());
UUID txId = request.txId();
@@ -155,12 +154,12 @@ public class TxFinishReplicaRequestHandler {
}
}
- private static Map<ReplicationGroupId, PartitionEnlistment> asReplicationGroupIdToPartitionMap(
- Map<ReplicationGroupIdMessage, PartitionEnlistmentMessage> messages
+ private static Map<ZonePartitionId, PartitionEnlistment> asReplicationGroupIdToPartitionMap(
+ Map<ZonePartitionIdMessage, PartitionEnlistmentMessage> messages
) {
- var result = new HashMap<ReplicationGroupId, PartitionEnlistment>(IgniteUtils.capacity(messages.size()));
+ var result = new HashMap<ZonePartitionId, PartitionEnlistment>(IgniteUtils.capacity(messages.size()));
- for (Entry<ReplicationGroupIdMessage, PartitionEnlistmentMessage> e : messages.entrySet()) {
+ for (Entry<ZonePartitionIdMessage, PartitionEnlistmentMessage> e : messages.entrySet()) {
result.put(e.getKey().asReplicationGroupId(), e.getValue().asPartition());
}
@@ -168,7 +167,7 @@ public class TxFinishReplicaRequestHandler {
}
private CompletableFuture<TransactionResult> finishAndCleanup(
- Map<ReplicationGroupId, PartitionEnlistment> enlistedPartitions,
+ Map<ZonePartitionId, PartitionEnlistment> enlistedPartitions,
boolean commit,
@Nullable HybridTimestamp commitTimestamp,
UUID txId
@@ -341,18 +340,8 @@ public class TxFinishReplicaRequestHandler {
private static EnlistedPartitionGroupMessage enlistedPartitionGroupMessage(EnlistedPartitionGroup enlistedPartitionGroup) {
return TX_MESSAGES_FACTORY.enlistedPartitionGroupMessage()
- .groupId(replicationGroupId(enlistedPartitionGroup.groupId()))
+ .groupId(toZonePartitionIdMessage(REPLICA_MESSAGES_FACTORY, enlistedPartitionGroup.groupId()))
.tableIds(enlistedPartitionGroup.tableIds())
.build();
}
-
- /**
- * Method to convert from {@link ReplicationGroupId} object to command-based {@link ReplicationGroupIdMessage} object.
- *
- * @param replicationGroupId {@link ReplicationGroupId} object to convert to {@link ReplicationGroupIdMessage}.
- * @return {@link ReplicationGroupIdMessage} object converted from argument.
- */
- private static ReplicationGroupIdMessage replicationGroupId(ReplicationGroupId replicationGroupId) {
- return toReplicationGroupIdMessage(REPLICA_MESSAGES_FACTORY, replicationGroupId);
- }
}
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/TxRecoveryMessageHandler.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/TxRecoveryMessageHandler.java
index b69b276d7ea..032cdc8ec6e 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/TxRecoveryMessageHandler.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/TxRecoveryMessageHandler.java
@@ -24,7 +24,7 @@ import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.partition.replicator.TxRecoveryEngine;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.tx.TxMeta;
import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
import org.apache.ignite.internal.tx.storage.state.TxStatePartitionStorage;
@@ -36,13 +36,13 @@ public class TxRecoveryMessageHandler {
private static final IgniteLogger LOG = Loggers.forClass(TxRecoveryMessageHandler.class);
private final TxStatePartitionStorage txStatePartitionStorage;
- private final ReplicationGroupId replicationGroupId;
+ private final ZonePartitionId replicationGroupId;
private final TxRecoveryEngine txRecoveryEngine;
/** Constructor. */
public TxRecoveryMessageHandler(
TxStatePartitionStorage txStatePartitionStorage,
- ReplicationGroupId replicationGroupId,
+ ZonePartitionId replicationGroupId,
TxRecoveryEngine txRecoveryEngine
) {
this.txStatePartitionStorage = txStatePartitionStorage;
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 d5919c8c55e..51c154310c3 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
@@ -230,6 +230,7 @@ public interface PartitionReplicationMessageGroup {
*
* @see #FINISH_TX_V2
*/
+ @Deprecated
short FINISH_TX_V1 = 40;
/**
@@ -237,6 +238,7 @@ public interface PartitionReplicationMessageGroup {
*
* @see #WRITE_INTENT_SWITCH_V2
*/
+ @Deprecated
short WRITE_INTENT_SWITCH_V1 = 41;
/**
@@ -258,6 +260,7 @@ public interface PartitionReplicationMessageGroup {
*
* @see #UPDATE_V2
*/
+ @Deprecated
short BUILD_INDEX_V1 = 44;
/** Message type for {@link UpdateMinimumActiveTxBeginTimeCommand}. */
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/command/FinishTxCommandV1.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/command/FinishTxCommandV1.java
index b01b603b2c6..5b9381effa2 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/command/FinishTxCommandV1.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/command/FinishTxCommandV1.java
@@ -27,6 +27,7 @@ import org.apache.ignite.internal.replicator.message.TablePartitionIdMessage;
*
* <p>This command is replaced with {@link FinishTxCommandV2} and only exists in the source code for backward compatibility.</p>
*/
+@Deprecated(forRemoval = true)
@Transferable(Commands.FINISH_TX_V1)
public interface FinishTxCommandV1 extends FinishTxCommand {
/** Returns ordered replication groups IDs. */
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/command/WriteIntentSwitchCommand.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/command/WriteIntentSwitchCommand.java
index c819bc5b18e..17c55e08842 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/command/WriteIntentSwitchCommand.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/command/WriteIntentSwitchCommand.java
@@ -27,6 +27,7 @@ import org.jetbrains.annotations.Nullable;
*
* <p>This command is replaced with {@link WriteIntentSwitchCommandV2} and only exists in the source code for backward compatibility.
*/
+@Deprecated(forRemoval = true)
@Transferable(PartitionReplicationMessageGroup.Commands.WRITE_INTENT_SWITCH_V1)
public interface WriteIntentSwitchCommand extends PartitionCommand {
/**
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/RaftTxFinishMarker.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/RaftTxFinishMarker.java
index 88f59ac95bf..dfee0250969 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/RaftTxFinishMarker.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/RaftTxFinishMarker.java
@@ -22,7 +22,7 @@ import static org.apache.ignite.internal.tx.TxState.COMMITTED;
import java.util.UUID;
import org.apache.ignite.internal.hlc.HybridTimestamp;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.internal.tx.TxStateMeta;
import org.jetbrains.annotations.Nullable;
@@ -49,7 +49,7 @@ public class RaftTxFinishMarker {
UUID txId,
boolean commit,
@Nullable HybridTimestamp commitTimestamp,
- @Nullable ReplicationGroupId commitPartitionId
+ @Nullable ZonePartitionId commitPartitionId
) {
txManager.updateTxMeta(txId, old -> TxStateMeta.builder(old, commit ? COMMITTED : ABORTED)
.commitPartitionId(commitPartitionId)
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/handlers/FinishTxCommandHandler.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/handlers/FinishTxCommandHandler.java
index 4412627c6b3..423742c88eb 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/handlers/FinishTxCommandHandler.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/handlers/FinishTxCommandHandler.java
@@ -17,7 +17,6 @@
package org.apache.ignite.internal.partition.replicator.raft.handlers;
-import static java.util.stream.Collectors.toList;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
import static org.apache.ignite.internal.tx.TxState.ABORTED;
import static org.apache.ignite.internal.tx.TxState.COMMITTED;
@@ -30,13 +29,11 @@ import org.apache.ignite.internal.lang.IgniteInternalException;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.partition.replicator.network.command.FinishTxCommand;
-import org.apache.ignite.internal.partition.replicator.network.command.FinishTxCommandV1;
import org.apache.ignite.internal.partition.replicator.network.command.FinishTxCommandV2;
import org.apache.ignite.internal.partition.replicator.raft.CommandResult;
import org.apache.ignite.internal.partition.replicator.raft.RaftTxFinishMarker;
import org.apache.ignite.internal.partition.replicator.raft.UnexpectedTransactionStateException;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
-import org.apache.ignite.internal.replicator.message.TablePartitionIdMessage;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.tx.TransactionResult;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.internal.tx.TxMeta;
@@ -53,14 +50,14 @@ public class FinishTxCommandHandler extends AbstractCommandHandler<FinishTxComma
private static final IgniteLogger LOG = Loggers.forClass(FinishTxCommandHandler.class);
private final TxStatePartitionStorage txStatePartitionStorage;
- private final ReplicationGroupId replicationGroupId;
+ private final ZonePartitionId replicationGroupId;
private final RaftTxFinishMarker txFinishMarker;
/** Constructor. */
public FinishTxCommandHandler(
TxStatePartitionStorage txStatePartitionStorage,
- ReplicationGroupId replicationGroupId,
+ ZonePartitionId replicationGroupId,
TxManager txManager
) {
this.txStatePartitionStorage = txStatePartitionStorage;
@@ -148,17 +145,8 @@ public class FinishTxCommandHandler extends AbstractCommandHandler<FinishTxComma
private static List<EnlistedPartitionGroup> enlistedPartitions(FinishTxCommand command) {
if (command instanceof FinishTxCommandV2) {
return fromPartitionMessages(((FinishTxCommandV2) command).partitions());
- } else if (command instanceof FinishTxCommandV1) {
- return enlistedPartitions((FinishTxCommandV1) command);
}
throw new IllegalArgumentException("Unknown command: " + command);
}
-
- private static List<EnlistedPartitionGroup> enlistedPartitions(FinishTxCommandV1 command) {
- return command.partitionIds().stream()
- .map(TablePartitionIdMessage::asTablePartitionId)
- .map(EnlistedPartitionGroup::new)
- .collect(toList());
- }
}
diff --git a/modules/partition-replicator/src/test/java/org/apache/ignite/internal/partition/replicator/handlers/TxCleanupRecoveryRequestHandlerTest.java b/modules/partition-replicator/src/test/java/org/apache/ignite/internal/partition/replicator/handlers/TxCleanupRecoveryRequestHandlerTest.java
index d7fa4166849..86e35bcae84 100644
--- a/modules/partition-replicator/src/test/java/org/apache/ignite/internal/partition/replicator/handlers/TxCleanupRecoveryRequestHandlerTest.java
+++ b/modules/partition-replicator/src/test/java/org/apache/ignite/internal/partition/replicator/handlers/TxCleanupRecoveryRequestHandlerTest.java
@@ -18,7 +18,7 @@
package org.apache.ignite.internal.partition.replicator.handlers;
import static java.util.stream.Collectors.toUnmodifiableList;
-import static org.apache.ignite.internal.replicator.message.ReplicaMessageUtils.toReplicationGroupIdMessage;
+import static org.apache.ignite.internal.replicator.message.ReplicaMessageUtils.toZonePartitionIdMessage;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -44,8 +44,6 @@ import org.apache.ignite.internal.failure.FailureProcessor;
import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.lang.IgniteBiTuple;
-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.message.ReplicaMessagesFactory;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
@@ -71,7 +69,7 @@ class TxCleanupRecoveryRequestHandlerTest extends BaseIgniteAbstractTest {
private final TxMessagesFactory txMessagesFactory = new TxMessagesFactory();
private final ReplicaMessagesFactory replicaMessagesFactory = new ReplicaMessagesFactory();
- private final ReplicationGroupId replicationGroupId = new ZonePartitionId(0, 0);
+ private final ZonePartitionId replicationGroupId = new ZonePartitionId(0, 0);
@Mock
private TxStatePartitionStorage txStatePartitionStorage;
@@ -86,8 +84,8 @@ class TxCleanupRecoveryRequestHandlerTest extends BaseIgniteAbstractTest {
private final HybridClock clock = new HybridClockImpl();
- private final TablePartitionId partition1Id = new TablePartitionId(1, 1);
- private final TablePartitionId partition2Id = new TablePartitionId(2, 2);
+ private final ZonePartitionId partition1Id = new ZonePartitionId(0, 1);
+ private final ZonePartitionId partition2Id = new ZonePartitionId(0, 2);
@BeforeEach
void setUp() {
@@ -114,16 +112,17 @@ class TxCleanupRecoveryRequestHandlerTest extends BaseIgniteAbstractTest {
when(iterator.hasNext()).thenThrow(exceptionFactory.get());
TxCleanupRecoveryRequest request = txMessagesFactory.txCleanupRecoveryRequest()
- .groupId(toReplicationGroupIdMessage(replicaMessagesFactory, replicationGroupId))
+ .groupId(toZonePartitionIdMessage(replicaMessagesFactory, replicationGroupId))
.build();
assertThat(handler.handle(request), willCompleteSuccessfully());
verify(failureProcessor, never()).process(any());
}
- private static List<EnlistedPartitionGroup> tableEnlistedPartitions(TablePartitionId... tablePartitionIds) {
- return Arrays.stream(tablePartitionIds)
- .map(tablePartitionId -> new EnlistedPartitionGroup(tablePartitionId, Set.of(tablePartitionId.tableId())))
+ private static List<EnlistedPartitionGroup> zoneEnlistedPartitions(ZonePartitionId... zonePartitionIds) {
+ return Arrays.stream(zonePartitionIds)
+ // Assume that partition ID equals to table ID and both tables belong to the same distribution zone.
+ .map(zonePartitionId -> new EnlistedPartitionGroup(zonePartitionId, Set.of(zonePartitionId.partitionId())))
.collect(toUnmodifiableList());
}
@@ -145,7 +144,7 @@ class TxCleanupRecoveryRequestHandlerTest extends BaseIgniteAbstractTest {
});
TxCleanupRecoveryRequest request = txMessagesFactory.txCleanupRecoveryRequest()
- .groupId(toReplicationGroupIdMessage(replicaMessagesFactory, replicationGroupId))
+ .groupId(toZonePartitionIdMessage(replicaMessagesFactory, replicationGroupId))
.build();
assertThat(handler.handle(request), willCompleteSuccessfully());
@@ -166,7 +165,7 @@ class TxCleanupRecoveryRequestHandlerTest extends BaseIgniteAbstractTest {
private Cursor<IgniteBiTuple<UUID, TxMeta>> mockTxCursor(int txCount) {
List<IgniteBiTuple<UUID, TxMeta>> tasks = new ArrayList<>();
for (int i = 0; i < txCount; i++) {
- TxMeta txMeta = new TxMeta(TxState.COMMITTED, tableEnlistedPartitions(partition1Id, partition2Id), clock.now());
+ TxMeta txMeta = new TxMeta(TxState.COMMITTED, zoneEnlistedPartitions(partition1Id, partition2Id), clock.now());
tasks.add(new IgniteBiTuple<>(UUID.randomUUID(), txMeta));
}
@@ -174,5 +173,4 @@ class TxCleanupRecoveryRequestHandlerTest extends BaseIgniteAbstractTest {
when(cursor.iterator()).thenReturn(tasks.iterator());
return cursor;
}
-
}
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 588ac269b40..609e9d8bae0 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
@@ -642,7 +642,6 @@ public class ItIgniteNodeRestartTest extends BaseIgniteRestartTest {
lowWatermark,
threadPoolsManager.commonScheduler(),
failureProcessor,
- nodeProperties,
metricManager
);
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 75c483c322c..d1b05303026 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
@@ -1068,7 +1068,6 @@ public class IgniteImpl implements Ignite {
lowWatermark,
threadPoolsManager.commonScheduler(),
failureManager,
- nodeProperties,
metricManager
);
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ScannableTableImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ScannableTableImpl.java
index 7f5807bd043..049f0fdc577 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ScannableTableImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ScannableTableImpl.java
@@ -27,7 +27,7 @@ import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Flow.Publisher;
import org.apache.ignite.internal.hlc.HybridTimestamp;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.schema.BinaryRowEx;
import org.apache.ignite.internal.schema.BinaryTuple;
@@ -218,7 +218,7 @@ public class ScannableTableImpl implements ScannableTable {
return TxContext.readOnly(txAttributes.id(), txAttributes.coordinatorId(), timestamp);
}
- ReplicationGroupId commitPartition = txAttributes.commitPartition();
+ ZonePartitionId commitPartition = txAttributes.commitPartition();
assert commitPartition != null;
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/TxAttributes.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/TxAttributes.java
index 95a5eea7c54..e0dcb848be0 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/TxAttributes.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/TxAttributes.java
@@ -60,8 +60,7 @@ public class TxAttributes {
return new TxAttributes(tx.id(), readTime, tx.coordinatorId());
}
- // TODO https://issues.apache.org/jira/browse/IGNITE-27175 remove cast to ZonePartitionId
- return new TxAttributes(tx.id(), (ZonePartitionId) tx.commitPartition(), tx.coordinatorId());
+ return new TxAttributes(tx.id(), tx.commitPartition(), tx.coordinatorId());
}
/**
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/NoOpTransaction.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/NoOpTransaction.java
index 7c68f0ee258..11d6ecbd644 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/NoOpTransaction.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/NoOpTransaction.java
@@ -28,7 +28,6 @@ import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.network.ClusterNodeImpl;
import org.apache.ignite.internal.network.InternalClusterNode;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.tx.PendingTxPartitionEnlistment;
@@ -158,7 +157,7 @@ public final class NoOpTransaction implements InternalTransaction {
}
@Override
- public PendingTxPartitionEnlistment enlistedPartition(ReplicationGroupId tablePartitionId) {
+ public PendingTxPartitionEnlistment enlistedPartition(ZonePartitionId replicationGroupId) {
return enlistment;
}
@@ -168,7 +167,7 @@ public final class NoOpTransaction implements InternalTransaction {
}
@Override
- public boolean assignCommitPartition(ReplicationGroupId replicationGroupId) {
+ public boolean assignCommitPartition(ZonePartitionId replicationGroupId) {
return true;
}
@@ -203,7 +202,7 @@ public final class NoOpTransaction implements InternalTransaction {
@Override
public void enlist(
- ReplicationGroupId replicationGroupId,
+ ZonePartitionId replicationGroupId,
int tableId,
String primaryNodeConsistentId,
long consistencyToken
diff --git a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItInternalTableReadWriteScanTest.java b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItInternalTableReadWriteScanTest.java
index 29ee38582d6..ea137024c5f 100644
--- a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItInternalTableReadWriteScanTest.java
+++ b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItInternalTableReadWriteScanTest.java
@@ -23,7 +23,6 @@ import java.util.concurrent.Flow.Publisher;
import org.apache.ignite.internal.hlc.HybridTimestampTracker;
import org.apache.ignite.internal.network.InternalClusterNode;
import org.apache.ignite.internal.replicator.PartitionGroupId;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.table.InternalTable;
@@ -79,15 +78,15 @@ public class ItInternalTableReadWriteScanTest extends ItAbstractInternalTableSca
InternalTransaction tx = internalTbl.txManager().beginExplicitRw(HYBRID_TIMESTAMP_TRACKER, InternalTxOptions.defaults());
int partId = ((PartitionGroupId) internalTbl.groupId()).partitionId();
- ReplicationGroupId tblPartId = new ZonePartitionId(zoneId, partId);
+ var zonePartitionId = new ZonePartitionId(zoneId, partId);
long term = 1L;
- tx.assignCommitPartition(tblPartId);
+ tx.assignCommitPartition(zonePartitionId);
- InternalClusterNode primaryReplicaNode = getPrimaryReplica(tblPartId);
+ InternalClusterNode primaryReplicaNode = getPrimaryReplica(zonePartitionId);
- tx.enlist(tblPartId, internalTbl.tableId(), primaryReplicaNode.name(), term);
+ tx.enlist(zonePartitionId, internalTbl.tableId(), primaryReplicaNode.name(), term);
return tx;
}
diff --git a/modules/table/src/integrationTest/java/org/apache/ignite/internal/table/ItColocationTest.java b/modules/table/src/integrationTest/java/org/apache/ignite/internal/table/ItColocationTest.java
index 336df84b531..5e346560bb3 100644
--- a/modules/table/src/integrationTest/java/org/apache/ignite/internal/table/ItColocationTest.java
+++ b/modules/table/src/integrationTest/java/org/apache/ignite/internal/table/ItColocationTest.java
@@ -81,7 +81,6 @@ import org.apache.ignite.internal.placementdriver.TestPlacementDriver;
import org.apache.ignite.internal.raft.Command;
import org.apache.ignite.internal.raft.service.RaftGroupService;
import org.apache.ignite.internal.replicator.ReplicaService;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.replicator.message.ReplicaMessagesFactory;
import org.apache.ignite.internal.replicator.message.ReplicaRequest;
@@ -227,11 +226,11 @@ public class ItColocationTest extends BaseIgniteAbstractTest {
@Override
public CompletableFuture<Void> finish(
HybridTimestampTracker observableTimestampTracker,
- ReplicationGroupId commitPartition,
+ ZonePartitionId commitPartition,
boolean commitIntent,
boolean timeoutExceeded,
boolean recovery,
- Map<ReplicationGroupId, PendingTxPartitionEnlistment> enlistedGroups,
+ Map<ZonePartitionId, PendingTxPartitionEnlistment> enlistedGroups,
UUID txId
) {
return nullCompletedFuture();
@@ -241,7 +240,7 @@ public class ItColocationTest extends BaseIgniteAbstractTest {
assertThat(txManager.startAsync(new ComponentContext()), willCompleteSuccessfully());
Int2ObjectMap<RaftGroupService> partRafts = new Int2ObjectOpenHashMap<>();
- Map<ReplicationGroupId, RaftGroupService> groupRafts = new HashMap<>();
+ Map<ZonePartitionId, RaftGroupService> groupRafts = new HashMap<>();
for (int i = 0; i < PARTS; ++i) {
RaftGroupService r = mock(RaftGroupService.class);
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java b/modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java
index bef61054490..638dd2af769 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java
@@ -27,7 +27,7 @@ import java.util.concurrent.ScheduledExecutorService;
import org.apache.ignite.internal.close.ManuallyCloseable;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.network.InternalClusterNode;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.schema.BinaryRowEx;
import org.apache.ignite.internal.storage.MvPartitionStorage;
@@ -460,7 +460,7 @@ public interface InternalTable extends ManuallyCloseable {
* @param partId Partition id.
* @return The id.
*/
- ReplicationGroupId targetReplicationGroupId(int partId);
+ ZonePartitionId targetReplicationGroupId(int partId);
/**
* Returns a metric source for this table.
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/TxContext.java b/modules/table/src/main/java/org/apache/ignite/internal/table/TxContext.java
index 117b058b88e..23ebd45b717 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/TxContext.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/TxContext.java
@@ -20,7 +20,7 @@ package org.apache.ignite.internal.table;
import java.util.Objects;
import java.util.UUID;
import org.apache.ignite.internal.hlc.HybridTimestamp;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.tostring.IgniteToStringBuilder;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.tx.TransactionIds;
@@ -53,7 +53,7 @@ public abstract class TxContext {
public static TxContext readWrite(
UUID txId,
UUID txCoordinatorId,
- ReplicationGroupId commitPartition,
+ ZonePartitionId commitPartition,
long enlistmentConsistencyToken
) {
return new ReadWrite(txId, txCoordinatorId, commitPartition, enlistmentConsistencyToken);
@@ -136,10 +136,10 @@ public abstract class TxContext {
/** Read-write transaction context. */
public static class ReadWrite extends TxContext {
- private final ReplicationGroupId commitPartition;
+ private final ZonePartitionId commitPartition;
private final long enlistmentConsistencyToken;
- private ReadWrite(UUID txId, UUID txCoordinatorId, ReplicationGroupId commitPartition, long enlistmentConsistencyToken) {
+ private ReadWrite(UUID txId, UUID txCoordinatorId, ZonePartitionId commitPartition, long enlistmentConsistencyToken) {
super(txId, txCoordinatorId);
Objects.requireNonNull(commitPartition, "Commit partition is mandatory for RW transaction");
@@ -170,7 +170,7 @@ public abstract class TxContext {
}
/** Returns transaction commit partition. */
- public ReplicationGroupId commitPartition() {
+ public ZonePartitionId commitPartition() {
return commitPartition;
}
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 c9c3730e7e6..0ced11ce45d 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
@@ -2019,7 +2019,7 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
}
private void destroyTxStateStoragesForTablesNotIn(Set<Integer> aliveTableIds) {
- Set<Integer> tableIdsOnDisk = sharedTxStateStorage.tableOrZoneIdsOnDisk();
+ Set<Integer> tableIdsOnDisk = sharedTxStateStorage.zoneIdsOnDisk();
for (int tableId : difference(tableIdsOnDisk, aliveTableIds)) {
sharedTxStateStorage.destroyStorage(tableId);
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/TransactionStateResolver.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/TransactionStateResolver.java
index 00189c63500..755b0441934 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/TransactionStateResolver.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/TransactionStateResolver.java
@@ -39,7 +39,7 @@ import org.apache.ignite.internal.network.MessagingService;
import org.apache.ignite.internal.network.NetworkMessage;
import org.apache.ignite.internal.network.RecipientLeftException;
import org.apache.ignite.internal.placementdriver.PlacementDriver;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.replicator.exception.PrimaryReplicaMissException;
import org.apache.ignite.internal.replicator.message.ReplicaMessagesFactory;
import org.apache.ignite.internal.tx.TransactionMeta;
@@ -141,7 +141,7 @@ public class TransactionStateResolver {
*/
public CompletableFuture<TransactionMeta> resolveTxState(
UUID txId,
- ReplicationGroupId commitGrpId,
+ ZonePartitionId commitGrpId,
@Nullable HybridTimestamp timestamp
) {
return resolveTxState(txId, commitGrpId, timestamp, AWAIT_PRIMARY_REPLICA_TIMEOUT, SECONDS);
@@ -159,7 +159,7 @@ public class TransactionStateResolver {
*/
public CompletableFuture<TransactionMeta> resolveTxState(
UUID txId,
- ReplicationGroupId commitGrpId,
+ ZonePartitionId commitGrpId,
@Nullable HybridTimestamp timestamp,
long awaitCommitPartitionAvailabilityTimeout,
TimeUnit awaitCommitPartitionAvailabilityTimeUnit
@@ -207,7 +207,7 @@ public class TransactionStateResolver {
private void resolveDistributiveTxState(
UUID txId,
@Nullable TxStateMeta localMeta,
- ReplicationGroupId commitGrpId,
+ ZonePartitionId commitGrpId,
@Nullable HybridTimestamp timestamp,
long awaitCommitPartitionAvailabilityTimeout,
TimeUnit awaitCommitPartitionAvailabilityTimeUnit,
@@ -249,7 +249,7 @@ public class TransactionStateResolver {
private void resolveTxStateFromTxCoordinator(
UUID txId,
@Nullable UUID coordinatorId,
- ReplicationGroupId commitGrpId,
+ ZonePartitionId commitGrpId,
HybridTimestamp timestamp,
CompletableFuture<TransactionMeta> txMetaFuture
) {
@@ -280,7 +280,7 @@ public class TransactionStateResolver {
private void resolveTxStateFromCommitPartition(
UUID txId,
- ReplicationGroupId commitGrpId,
+ ZonePartitionId commitGrpId,
CompletableFuture<TransactionMeta> txMetaFuture
) {
resolveTxStateFromCommitPartition(txId, commitGrpId, AWAIT_PRIMARY_REPLICA_TIMEOUT, SECONDS, txMetaFuture);
@@ -288,7 +288,7 @@ public class TransactionStateResolver {
private void resolveTxStateFromCommitPartition(
UUID txId,
- ReplicationGroupId commitGrpId,
+ ZonePartitionId commitGrpId,
long awaitPrimaryReplicaTimeout,
TimeUnit awaitPrimaryReplicaTimeUnit,
CompletableFuture<TransactionMeta> txMetaFuture
@@ -324,7 +324,7 @@ public class TransactionStateResolver {
*/
private void sendAndRetry(
CompletableFuture<TransactionMeta> resFut,
- ReplicationGroupId replicaGrp,
+ ZonePartitionId replicaGrp,
UUID txId,
long awaitPrimaryReplicaTimeout,
TimeUnit awaitPrimaryReplicaTimeUnit
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
index 586d6a4d610..14bc1cf914d 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
@@ -41,7 +41,7 @@ import static org.apache.ignite.internal.partition.replicator.network.replicatio
import static org.apache.ignite.internal.partition.replicator.network.replication.RequestType.RW_REPLACE_IF_EXIST;
import static org.apache.ignite.internal.partition.replicator.network.replication.RequestType.RW_UPSERT;
import static org.apache.ignite.internal.partition.replicator.network.replication.RequestType.RW_UPSERT_ALL;
-import static org.apache.ignite.internal.replicator.message.ReplicaMessageUtils.toReplicationGroupIdMessage;
+import static org.apache.ignite.internal.replicator.message.ReplicaMessageUtils.toZonePartitionIdMessage;
import static org.apache.ignite.internal.table.distributed.TableUtils.isDirectFlowApplicable;
import static org.apache.ignite.internal.table.distributed.storage.RowBatch.allResultFutures;
import static org.apache.ignite.internal.util.CompletableFutures.completedOrFailedFuture;
@@ -101,7 +101,6 @@ import org.apache.ignite.internal.partition.replicator.network.replication.ScanC
import org.apache.ignite.internal.placementdriver.PlacementDriver;
import org.apache.ignite.internal.placementdriver.ReplicaMeta;
import org.apache.ignite.internal.replicator.ReplicaService;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.replicator.exception.PrimaryReplicaMissException;
import org.apache.ignite.internal.replicator.exception.ReplicationException;
@@ -321,7 +320,7 @@ public class InternalTableImpl implements InternalTable {
private <R> CompletableFuture<R> enlistInTx(
BinaryRowEx row,
@Nullable InternalTransaction tx,
- IgniteTriFunction<InternalTransaction, ReplicationGroupId, Long, ReplicaRequest> fac,
+ IgniteTriFunction<InternalTransaction, ZonePartitionId, Long, ReplicaRequest> fac,
BiPredicate<R, ReplicaRequest> noWriteChecker
) {
return enlistInTx(row, tx, fac, noWriteChecker, null);
@@ -340,7 +339,7 @@ public class InternalTableImpl implements InternalTable {
private <R> CompletableFuture<R> enlistInTx(
BinaryRowEx row,
@Nullable InternalTransaction tx,
- IgniteTriFunction<InternalTransaction, ReplicationGroupId, Long, ReplicaRequest> fac,
+ IgniteTriFunction<InternalTransaction, ZonePartitionId, Long, ReplicaRequest> fac,
BiPredicate<R, ReplicaRequest> noWriteChecker,
@Nullable Long txStartTs
) {
@@ -360,7 +359,7 @@ public class InternalTableImpl implements InternalTable {
int partId = partitionId(row);
- ReplicationGroupId partGroupId = targetReplicationGroupId(partId);
+ ZonePartitionId partGroupId = targetReplicationGroupId(partId);
PendingTxPartitionEnlistment enlistment = actualTx.enlistedPartition(partGroupId);
@@ -470,7 +469,7 @@ public class InternalTableImpl implements InternalTable {
int partitionId = partitionRowBatch.getIntKey();
RowBatch rowBatch = partitionRowBatch.getValue();
- ReplicationGroupId replicationGroupId = targetReplicationGroupId(partitionId);
+ ZonePartitionId replicationGroupId = targetReplicationGroupId(partitionId);
PendingTxPartitionEnlistment enlistment = actualTx.enlistedPartition(replicationGroupId);
@@ -556,7 +555,7 @@ public class InternalTableImpl implements InternalTable {
@Nullable BinaryTuplePrefix upperBound,
int flags
) {
- ReplicationGroupId replicationGroupId = targetReplicationGroupId(partId);
+ ZonePartitionId replicationGroupId = targetReplicationGroupId(partId);
PendingTxPartitionEnlistment enlistment = tx.enlistedPartition(replicationGroupId);
@@ -808,13 +807,13 @@ public class InternalTableImpl implements InternalTable {
private <R> CompletableFuture<R> evaluateReadOnlyPrimaryNode(
@Nullable InternalTransaction tx,
BinaryRowEx row,
- BiFunction<ReplicationGroupId, Long, ReplicaRequest> op
+ BiFunction<ZonePartitionId, Long, ReplicaRequest> op
) {
InternalTransaction actualTx = startImplicitRoTxIfNeeded(tx);
int partId = partitionId(row);
- ReplicationGroupId replicationGroupId = targetReplicationGroupId(partId);
+ ZonePartitionId replicationGroupId = targetReplicationGroupId(partId);
return sendReadOnlyToPrimaryReplica(actualTx, replicationGroupId, op);
}
@@ -829,13 +828,13 @@ public class InternalTableImpl implements InternalTable {
*/
private <R> CompletableFuture<R> evaluateReadOnlyPrimaryNode(
Collection<BinaryRowEx> rows,
- BiFunction<ReplicationGroupId, Long, ReplicaRequest> op
+ BiFunction<ZonePartitionId, Long, ReplicaRequest> op
) {
InternalTransaction actualTx = txManager.beginImplicitRo(observableTimestampTracker);
int partId = partitionId(rows.iterator().next());
- ReplicationGroupId replicationGroupId = targetReplicationGroupId(partId);
+ ZonePartitionId replicationGroupId = targetReplicationGroupId(partId);
return sendReadOnlyToPrimaryReplica(actualTx, replicationGroupId, op);
}
@@ -851,8 +850,8 @@ public class InternalTableImpl implements InternalTable {
*/
private <R> CompletableFuture<R> sendReadOnlyToPrimaryReplica(
InternalTransaction tx,
- ReplicationGroupId replicationGroupId,
- BiFunction<ReplicationGroupId, Long, ReplicaRequest> op
+ ZonePartitionId replicationGroupId,
+ BiFunction<ZonePartitionId, Long, ReplicaRequest> op
) {
ReplicaMeta meta = placementDriver.getCurrentPrimaryReplica(replicationGroupId, tx.schemaTimestamp());
@@ -968,7 +967,7 @@ public class InternalTableImpl implements InternalTable {
InternalClusterNode recipientNode
) {
int partId = partitionId(keyRow);
- ReplicationGroupId replicationGroupId = targetReplicationGroupId(partId);
+ ZonePartitionId replicationGroupId = targetReplicationGroupId(partId);
return replicaSvc.invoke(recipientNode, TABLE_MESSAGES_FACTORY.readOnlySingleRowPkReplicaRequest()
.groupId(serializeReplicationGroupId(replicationGroupId))
@@ -1058,7 +1057,7 @@ public class InternalTableImpl implements InternalTable {
for (Int2ObjectMap.Entry<RowBatch> partitionRowBatch : rowBatchByPartitionId.int2ObjectEntrySet()) {
int partitionId = partitionRowBatch.getIntKey();
- ReplicationGroupId replicationGroupId = targetReplicationGroupId(partitionId);
+ ZonePartitionId replicationGroupId = targetReplicationGroupId(partitionId);
ReadOnlyMultiRowPkReplicaRequest request = TABLE_MESSAGES_FACTORY.readOnlyMultiRowPkReplicaRequest()
.groupId(serializeReplicationGroupId(replicationGroupId))
@@ -1084,7 +1083,7 @@ public class InternalTableImpl implements InternalTable {
RequestType requestType,
Collection<? extends BinaryRow> rows,
InternalTransaction tx,
- ReplicationGroupId groupId,
+ ZonePartitionId groupId,
Long enlistmentConsistencyToken,
boolean full
) {
@@ -1208,7 +1207,7 @@ public class InternalTableImpl implements InternalTable {
@Nullable Long txStartTs
) {
InternalTransaction tx = txManager.beginImplicitRw(observableTimestampTracker);
- ReplicationGroupId replicationGroupId = targetReplicationGroupId(partition);
+ ZonePartitionId replicationGroupId = targetReplicationGroupId(partition);
assert rows.stream().allMatch(row -> partitionId(row) == partition) : "Invalid batch for partition " + partition;
@@ -1327,7 +1326,7 @@ public class InternalTableImpl implements InternalTable {
Collection<? extends BinaryRow> rows,
@Nullable BitSet deleted,
InternalTransaction tx,
- ReplicationGroupId groupId,
+ ZonePartitionId groupId,
Long enlistmentConsistencyToken,
boolean full
) {
@@ -1663,7 +1662,7 @@ public class InternalTableImpl implements InternalTable {
TxContext.ReadOnly txContext = (TxContext.ReadOnly) opCtx.txContext();
- ReplicationGroupId replicationGroupId = targetReplicationGroupId(partId);
+ ZonePartitionId replicationGroupId = targetReplicationGroupId(partId);
return new PartitionScanPublisher<>(new ReadOnlyInflightBatchRequestTracker(transactionInflights, txContext.txId())) {
@Override
@@ -1747,7 +1746,7 @@ public class InternalTableImpl implements InternalTable {
if (tx.implicit()) {
opFut = completedOrFailedFuture(null, th);
} else {
- var replicationGrpId = targetReplicationGroupId(partId);
+ ZonePartitionId replicationGrpId = targetReplicationGroupId(partId);
PendingTxPartitionEnlistment enlistment = tx.enlistedPartition(replicationGrpId);
opFut = enlistment != null ? completeScan(
@@ -1790,7 +1789,7 @@ public class InternalTableImpl implements InternalTable {
TxContext.ReadWrite txContext = (TxContext.ReadWrite) opCtx.txContext();
- ReplicationGroupId replicationGroupId = targetReplicationGroupId(partId);
+ ZonePartitionId replicationGroupId = targetReplicationGroupId(partId);
return new PartitionScanPublisher<>(READ_WRITE_INFLIGHT_BATCH_REQUEST_TRACKER) {
@Override
@@ -1836,7 +1835,7 @@ public class InternalTableImpl implements InternalTable {
*/
private CompletableFuture<Void> completeScan(
UUID txId,
- ReplicationGroupId replicaGrpId,
+ ZonePartitionId replicaGrpId,
long scanId,
Throwable th,
String recipientConsistentId,
@@ -2020,13 +2019,13 @@ public class InternalTableImpl implements InternalTable {
protected CompletableFuture<PendingTxPartitionEnlistment> enlist(int partId, InternalTransaction tx) {
HybridTimestamp now = tx.schemaTimestamp();
- ReplicationGroupId replicationGroupId = targetReplicationGroupId(partId);
+ ZonePartitionId replicationGroupId = targetReplicationGroupId(partId);
tx.assignCommitPartition(replicationGroupId);
ReplicaMeta meta = placementDriver.getCurrentPrimaryReplica(replicationGroupId, now);
Function<ReplicaMeta, PendingTxPartitionEnlistment> enlistClo = replicaMeta -> {
- ReplicationGroupId partGroupId = targetReplicationGroupId(partId);
+ ZonePartitionId partGroupId = targetReplicationGroupId(partId);
String leaseHolderNodeId = replicaMeta.getLeaseholder();
@@ -2053,7 +2052,7 @@ public class InternalTableImpl implements InternalTable {
return partitionMeta(targetReplicationGroupId(partitionIndex), clockService.current()).thenApply(this::getClusterNode);
}
- private CompletableFuture<ReplicaMeta> partitionMeta(ReplicationGroupId replicationGroupId, HybridTimestamp at) {
+ private CompletableFuture<ReplicaMeta> partitionMeta(ZonePartitionId replicationGroupId, HybridTimestamp at) {
return awaitPrimaryReplica(replicationGroupId, at)
.exceptionally(e -> {
throw withCause(
@@ -2150,7 +2149,7 @@ public class InternalTableImpl implements InternalTable {
* @return Cluster node to evaluate read-only request.
*/
protected CompletableFuture<InternalClusterNode> evaluateReadOnlyRecipientNode(int partId, HybridTimestamp readTimestamp) {
- ReplicationGroupId replicationGroupId = targetReplicationGroupId(partId);
+ ZonePartitionId replicationGroupId = targetReplicationGroupId(partId);
return awaitPrimaryReplica(replicationGroupId, readTimestamp)
.handle((res, e) -> {
@@ -2167,7 +2166,7 @@ public class InternalTableImpl implements InternalTable {
}
private static TransactionException createFailedGetPrimaryReplicaTransactionException(
- ReplicationGroupId replicationGroupId,
+ ZonePartitionId replicationGroupId,
HybridTimestamp readTimestamp
) {
String errorMessage = format(
@@ -2201,7 +2200,7 @@ public class InternalTableImpl implements InternalTable {
var invokeFutures = new CompletableFuture<?>[partitions];
for (int partId = 0; partId < partitions; partId++) {
- ReplicationGroupId replicaGroupId = targetReplicationGroupId(partId);
+ ZonePartitionId replicaGroupId = targetReplicationGroupId(partId);
ReplicationGroupIdMessage partitionIdMessage = serializeReplicationGroupId(replicaGroupId);
Function<ReplicaMeta, ReplicaRequest> requestFactory = replicaMeta ->
@@ -2220,12 +2219,12 @@ public class InternalTableImpl implements InternalTable {
}
@Override
- public final ReplicationGroupId targetReplicationGroupId(int partitionIndex) {
+ public final ZonePartitionId targetReplicationGroupId(int partitionIndex) {
return new ZonePartitionId(zoneId, partitionIndex);
}
- private static ZonePartitionIdMessage serializeReplicationGroupId(ReplicationGroupId replicationGroupId) {
- return toReplicationGroupIdMessage(REPLICA_MESSAGES_FACTORY, replicationGroupId);
+ private static ZonePartitionIdMessage serializeReplicationGroupId(ZonePartitionId replicationGroupId) {
+ return toZonePartitionIdMessage(REPLICA_MESSAGES_FACTORY, replicationGroupId);
}
@Override
@@ -2234,7 +2233,7 @@ public class InternalTableImpl implements InternalTable {
}
private <T> CompletableFuture<T> sendToPrimaryWithRetry(
- ReplicationGroupId replicationGroupId,
+ ZonePartitionId replicationGroupId,
HybridTimestamp hybridTimestamp,
int numRetries,
Function<ReplicaMeta, ReplicaRequest> requestFactory
@@ -2317,7 +2316,7 @@ public class InternalTableImpl implements InternalTable {
private ReplicaRequest upsertAllInternal(
Collection<? extends BinaryRow> keyRows0,
InternalTransaction txo,
- ReplicationGroupId groupId,
+ ZonePartitionId groupId,
Long enlistmentConsistencyToken,
boolean full
) {
@@ -2328,7 +2327,7 @@ public class InternalTableImpl implements InternalTable {
Collection<? extends BinaryRow> keyRows0,
@Nullable BitSet deleted,
InternalTransaction txo,
- ReplicationGroupId groupId,
+ ZonePartitionId groupId,
Long enlistmentConsistencyToken,
boolean full
) {
@@ -2346,7 +2345,7 @@ public class InternalTableImpl implements InternalTable {
return matchAny(unwrapCause(e), ACQUIRE_LOCK_ERR, REPLICA_MISS_ERR, GROUP_OVERLOADED_ERR);
}
- private CompletableFuture<ReplicaMeta> awaitPrimaryReplica(ReplicationGroupId replicationGroupId, HybridTimestamp timestamp) {
+ private CompletableFuture<ReplicaMeta> awaitPrimaryReplica(ZonePartitionId replicationGroupId, HybridTimestamp timestamp) {
return placementDriver.awaitPrimaryReplica(
replicationGroupId,
timestamp,
@@ -2376,7 +2375,7 @@ public class InternalTableImpl implements InternalTable {
ReplicaRequest create(
Collection<? extends BinaryRow> keyRows,
InternalTransaction tx,
- ReplicationGroupId groupId,
+ ZonePartitionId groupId,
Long enlistmentConsistencyToken,
Boolean full
);
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/command/PartitionRaftCommandsSerializationTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/command/PartitionRaftCommandsSerializationTest.java
index 7c250beb002..8b3a65ce475 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/command/PartitionRaftCommandsSerializationTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/command/PartitionRaftCommandsSerializationTest.java
@@ -46,8 +46,8 @@ 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.raft.Command;
import org.apache.ignite.internal.replicator.message.ReplicaMessagesFactory;
-import org.apache.ignite.internal.replicator.message.ReplicationGroupIdMessage;
import org.apache.ignite.internal.replicator.message.TablePartitionIdMessage;
+import org.apache.ignite.internal.replicator.message.ZonePartitionIdMessage;
import org.apache.ignite.internal.schema.Column;
import org.apache.ignite.internal.schema.SchemaDescriptor;
import org.apache.ignite.internal.schema.marshaller.KvMarshaller;
@@ -245,11 +245,11 @@ public class PartitionRaftCommandsSerializationTest extends IgniteAbstractTest {
@Test
public void testFinishTxCommand() {
- ArrayList<ReplicationGroupIdMessage> grps = new ArrayList<>(10);
+ ArrayList<ZonePartitionIdMessage> grps = new ArrayList<>(10);
for (int i = 0; i < 10; i++) {
- grps.add(REPLICA_MESSAGES_FACTORY.tablePartitionIdMessage()
- .tableId(1)
+ grps.add(REPLICA_MESSAGES_FACTORY.zonePartitionIdMessage()
+ .zoneId(1)
.partitionId(i)
.build());
}
@@ -269,7 +269,7 @@ public class PartitionRaftCommandsSerializationTest extends IgniteAbstractTest {
assertEquals(cmd.commitTimestamp(), readCmd.commitTimestamp());
}
- private static EnlistedPartitionGroupMessage partitionGroupMessage(ReplicationGroupIdMessage groupIdMessage, Set<Integer> tableIds) {
+ private static EnlistedPartitionGroupMessage partitionGroupMessage(ZonePartitionIdMessage groupIdMessage, Set<Integer> tableIds) {
return TX_MESSAGES_FACTORY.enlistedPartitionGroupMessage()
.groupId(groupIdMessage)
.tableIds(tableIds)
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImplTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImplTest.java
index bed39739520..865b4d8adb9 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImplTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImplTest.java
@@ -20,7 +20,6 @@ package org.apache.ignite.internal.table.distributed.storage;
import static java.util.UUID.randomUUID;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.apache.ignite.internal.lang.IgniteSystemProperties.COLOCATION_FEATURE_FLAG;
-import static org.apache.ignite.internal.lang.IgniteSystemProperties.colocationEnabled;
import static org.apache.ignite.internal.table.distributed.storage.InternalTableImpl.collectMultiRowsResponsesWithRestoreOrder;
import static org.apache.ignite.internal.table.distributed.storage.InternalTableImpl.collectRejectedRowsResponses;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
@@ -376,15 +375,15 @@ public class InternalTableImplTest extends BaseIgniteAbstractTest {
}
private PendingTxPartitionEnlistment extractSingleEnlistmentForZone() {
- Map<ReplicationGroupId, PendingTxPartitionEnlistment> capturedEnlistments = extractEnlistmentsFromTxFinish();
+ Map<ZonePartitionId, PendingTxPartitionEnlistment> capturedEnlistments = extractEnlistmentsFromTxFinish();
assertThat(capturedEnlistments, is(aMapWithSize(1)));
PendingTxPartitionEnlistment enlistment = capturedEnlistments.get(new ZonePartitionId(ZONE_ID, 0));
assertThat(enlistment, is(notNullValue()));
return enlistment;
}
- private Map<ReplicationGroupId, PendingTxPartitionEnlistment> extractEnlistmentsFromTxFinish() {
- ArgumentCaptor<Map<ReplicationGroupId, PendingTxPartitionEnlistment>> enlistmentsCaptor = ArgumentCaptor.captor();
+ private Map<ZonePartitionId, PendingTxPartitionEnlistment> extractEnlistmentsFromTxFinish() {
+ ArgumentCaptor<Map<ZonePartitionId, PendingTxPartitionEnlistment>> enlistmentsCaptor = ArgumentCaptor.captor();
verify(txManager).finish(any(), any(), anyBoolean(), anyBoolean(), anyBoolean(), enlistmentsCaptor.capture(), any());
@@ -398,8 +397,7 @@ public class InternalTableImplTest extends BaseIgniteAbstractTest {
TestTransactionIds.newTransactionId(),
randomUUID(),
false,
- 10_000,
- colocationEnabled()
+ 10_000
);
}
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 d9247e862fd..c08e94a027b 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
@@ -76,7 +76,6 @@ import org.apache.ignite.internal.cluster.management.topology.api.LogicalNode;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologyEventListener;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologyService;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologySnapshot;
-import org.apache.ignite.internal.components.SystemPropertiesNodeProperties;
import org.apache.ignite.internal.configuration.RaftGroupOptionsConfigHelper;
import org.apache.ignite.internal.configuration.SystemDistributedConfiguration;
import org.apache.ignite.internal.configuration.SystemLocalConfiguration;
@@ -624,7 +623,6 @@ public class ItTxTestCluster {
lowWatermark,
executor,
new NoOpFailureManager(),
- new SystemPropertiesNodeProperties(),
new TestMetricManager()
);
}
@@ -1315,7 +1313,6 @@ public class ItTxTestCluster {
lowWatermark,
executor,
new NoOpFailureManager(),
- new SystemPropertiesNodeProperties(),
new TestMetricManager()
);
diff --git a/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryManagerTest.java b/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryManagerTest.java
index bba833c7b9c..588067773a5 100644
--- a/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryManagerTest.java
+++ b/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryManagerTest.java
@@ -61,7 +61,6 @@ import org.apache.ignite.internal.partition.replicator.network.disaster.LocalPar
import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.partitiondistribution.Assignments;
import org.apache.ignite.internal.placementdriver.ReplicaMeta;
-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.schema.Column;
@@ -137,7 +136,7 @@ public class ItDisasterRecoveryManagerTest extends ClusterPerTestIntegrationTest
));
}
- private static String findPrimaryNodeName(IgniteImpl ignite, ReplicationGroupId replicationGroupId) {
+ private static String findPrimaryNodeName(IgniteImpl ignite, ZonePartitionId replicationGroupId) {
assertThat(awaitPrimaryReplicaForNow(ignite, replicationGroupId), willCompleteSuccessfully());
CompletableFuture<ReplicaMeta> primary = ignite.placementDriver().getPrimaryReplica(replicationGroupId, ignite.clock().now());
@@ -147,7 +146,7 @@ public class ItDisasterRecoveryManagerTest extends ClusterPerTestIntegrationTest
return primary.join().getLeaseholder();
}
- private Ignite findPrimaryIgniteNode(IgniteImpl ignite, ReplicationGroupId replicationGroupId) {
+ private Ignite findPrimaryIgniteNode(IgniteImpl ignite, ZonePartitionId replicationGroupId) {
return cluster.runningNodes()
.filter(node -> node.name().equals(findPrimaryNodeName(ignite, replicationGroupId)))
.findFirst()
@@ -442,7 +441,7 @@ public class ItDisasterRecoveryManagerTest extends ClusterPerTestIntegrationTest
return unwrapIgniteImpl(nodeToCleanup);
}
- private static CompletableFuture<ReplicaMeta> awaitPrimaryReplicaForNow(IgniteImpl node, ReplicationGroupId replicationGroupId) {
+ private static CompletableFuture<ReplicaMeta> awaitPrimaryReplicaForNow(IgniteImpl node, ZonePartitionId replicationGroupId) {
return node.placementDriver().awaitPrimaryReplica(replicationGroupId, node.clock().now(), 60, SECONDS);
}
diff --git a/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/tx/ItTransactionMetricsTest.java b/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/tx/ItTransactionMetricsTest.java
index 20775b8705a..b3f4499b17a 100644
--- a/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/tx/ItTransactionMetricsTest.java
+++ b/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/tx/ItTransactionMetricsTest.java
@@ -20,7 +20,6 @@ package org.apache.ignite.internal.tx;
import static java.util.stream.Collectors.toSet;
import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
import static org.apache.ignite.internal.TestWrappers.unwrapTableImpl;
-import static org.apache.ignite.internal.lang.IgniteSystemProperties.colocationEnabled;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.assertThrows;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -38,8 +37,6 @@ import org.apache.ignite.internal.app.IgniteImpl;
import org.apache.ignite.internal.metrics.LongMetric;
import org.apache.ignite.internal.metrics.MetricSet;
import org.apache.ignite.internal.placementdriver.ReplicaMeta;
-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.table.NodeUtils;
import org.apache.ignite.internal.table.TableImpl;
@@ -66,7 +63,7 @@ public class ItTransactionMetricsTest extends ClusterPerClassIntegrationTest {
}
@BeforeAll
- void createTable() throws Exception {
+ void createTable() {
sql("CREATE TABLE " + TABLE_NAME + " (id INT PRIMARY KEY, val VARCHAR)");
}
@@ -332,9 +329,7 @@ public class ItTransactionMetricsTest extends ClusterPerClassIntegrationTest {
keyValueView(0).put(tx, key, "value");
- ReplicationGroupId replicationGroupId = colocationEnabled()
- ? new ZonePartitionId(table.zoneId(), partitionId)
- : new TablePartitionId(table.tableId(), partitionId);
+ ZonePartitionId replicationGroupId = new ZonePartitionId(table.zoneId(), partitionId);
ReplicaMeta leaseholder = NodeUtils.leaseholder(unwrapIgniteImpl(node(0)), replicationGroupId);
diff --git a/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/tx/readonly/ItReadOnlyTxInPastTest.java b/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/tx/readonly/ItReadOnlyTxInPastTest.java
index 45a67a023cb..ca0e7eabca4 100644
--- a/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/tx/readonly/ItReadOnlyTxInPastTest.java
+++ b/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/tx/readonly/ItReadOnlyTxInPastTest.java
@@ -20,7 +20,6 @@ package org.apache.ignite.internal.tx.readonly;
import static org.apache.ignite.internal.AssignmentsTestUtils.awaitAssignmentsStabilization;
import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
import static org.apache.ignite.internal.distributionzones.DistributionZonesTestUtil.getDefaultZone;
-import static org.apache.ignite.internal.lang.IgniteSystemProperties.colocationEnabled;
import static org.apache.ignite.internal.sql.engine.util.SqlTestUtils.executeUpdate;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
@@ -66,11 +65,9 @@ class ItReadOnlyTxInPastTest extends ClusterPerTestIntegrationTest {
void explicitReadOnlyTxDoesNotLookBeforeTableCreation() throws Exception {
Ignite node = cluster.node(0);
- if (colocationEnabled()) {
- // Generally it's required to await default zone dataNodesAutoAdjustScaleUp timeout in order to treat zone as ready one.
- // In order to eliminate awaiting interval, default zone scaleUp is altered to be immediate.
- setDefaultZoneAutoAdjustScaleUpTimeoutToImmediate();
- }
+ // Generally it's required to await default zone dataNodesAutoAdjustScaleUp timeout in order to treat zone as ready one.
+ // In order to eliminate awaiting interval, default zone scaleUp is altered to be immediate.
+ setDefaultZoneAutoAdjustScaleUpTimeoutToImmediate();
// In case of empty assignments, SQL engine will throw "Mandatory nodes were excluded from mapping: []".
// In order to eliminate this, assignments stabilization is needed, otherwise test may fail. Not related to colocation.
diff --git a/modules/transactions/src/integrationTest/java/org/apache/ignite/tx/distributed/ItTransactionPrimaryChangeTest.java b/modules/transactions/src/integrationTest/java/org/apache/ignite/tx/distributed/ItTransactionPrimaryChangeTest.java
index 7d14c08d5cd..29e248e9ed1 100644
--- a/modules/transactions/src/integrationTest/java/org/apache/ignite/tx/distributed/ItTransactionPrimaryChangeTest.java
+++ b/modules/transactions/src/integrationTest/java/org/apache/ignite/tx/distributed/ItTransactionPrimaryChangeTest.java
@@ -22,7 +22,6 @@ import static org.apache.ignite.internal.TestDefaultProfilesNames.DEFAULT_AIPERS
import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
import static org.apache.ignite.internal.TestWrappers.unwrapInternalTransaction;
import static org.apache.ignite.internal.TestWrappers.unwrapTableImpl;
-import static org.apache.ignite.internal.lang.IgniteSystemProperties.colocationEnabled;
import static org.apache.ignite.internal.sql.engine.util.SqlTestUtils.executeUpdate;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.apache.ignite.internal.tx.test.ItTransactionTestUtils.waitAndGetPrimaryReplica;
@@ -37,7 +36,6 @@ import org.apache.ignite.internal.ClusterPerTestIntegrationTest;
import org.apache.ignite.internal.TestWrappers;
import org.apache.ignite.internal.app.IgniteImpl;
import org.apache.ignite.internal.partition.replicator.network.command.UpdateCommand;
-import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.table.NodeUtils;
import org.apache.ignite.internal.table.TableImpl;
@@ -70,7 +68,7 @@ public class ItTransactionPrimaryChangeTest extends ClusterPerTestIntegrationTes
+ "}";
@BeforeEach
- public void setup() throws Exception {
+ public void setup() {
String zoneSql = "create zone test_zone (partitions 1, replicas 3) storage profiles ['" + DEFAULT_AIPERSIST_PROFILE_NAME + "']";
String sql = "create table " + TABLE_NAME + " (key int primary key, val varchar(20)) zone TEST_ZONE";
@@ -111,9 +109,7 @@ public class ItTransactionPrimaryChangeTest extends ClusterPerTestIntegrationTes
int partId = 0;
- var replicationGrp = colocationEnabled()
- ? new ZonePartitionId(tbl.zoneId(), partId)
- : new TablePartitionId(tbl.tableId(), partId);
+ var replicationGrp = new ZonePartitionId(tbl.zoneId(), partId);
String leaseholder = waitAndGetPrimaryReplica(unwrapIgniteImpl(node(0)), replicationGrp).getLeaseholder();
diff --git a/modules/transactions/src/integrationTest/java/org/apache/ignite/tx/distributed/ItTransactionRecoveryTest.java b/modules/transactions/src/integrationTest/java/org/apache/ignite/tx/distributed/ItTransactionRecoveryTest.java
index 56bdf31dd4e..ab88e3c864b 100644
--- a/modules/transactions/src/integrationTest/java/org/apache/ignite/tx/distributed/ItTransactionRecoveryTest.java
+++ b/modules/transactions/src/integrationTest/java/org/apache/ignite/tx/distributed/ItTransactionRecoveryTest.java
@@ -22,7 +22,6 @@ import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
import static org.apache.ignite.internal.TestWrappers.unwrapTableImpl;
import static org.apache.ignite.internal.catalog.CatalogService.DEFAULT_STORAGE_PROFILE;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
-import static org.apache.ignite.internal.lang.IgniteSystemProperties.colocationEnabled;
import static org.apache.ignite.internal.sql.engine.util.SqlTestUtils.executeUpdate;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.assertThrowsWithCode;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.bypassingThreadAssertions;
@@ -72,8 +71,6 @@ import org.apache.ignite.internal.network.NetworkMessage;
import org.apache.ignite.internal.partition.replicator.network.replication.ReadWriteSingleRowReplicaRequest;
import org.apache.ignite.internal.placementdriver.message.PlacementDriverMessagesFactory;
import org.apache.ignite.internal.placementdriver.message.StopLeaseProlongationMessage;
-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.message.ErrorTimestampAwareReplicaResponse;
import org.apache.ignite.internal.replicator.message.TimestampAwareReplicaResponse;
@@ -156,11 +153,9 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
public void testMultipleAbandonedTxsAreAborted() throws Exception {
TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));
- var tblReplicationGrp = colocationEnabled()
- ? new ZonePartitionId(tbl.zoneId(), PART_ID)
- : new TablePartitionId(tbl.tableId(), PART_ID);
+ var partitionGroupId = new ZonePartitionId(tbl.zoneId(), PART_ID);
- String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);
+ String leaseholder = waitAndGetLeaseholder(node(0), partitionGroupId);
IgniteImpl commitPartNode = findNodeByName(leaseholder);
@@ -214,11 +209,9 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
public void testMultipleRecoveryRequestsIssued() throws Exception {
TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));
- var tblReplicationGrp = colocationEnabled()
- ? new ZonePartitionId(tbl.zoneId(), PART_ID)
- : new TablePartitionId(tbl.tableId(), PART_ID);
+ var partitionGroupId = new ZonePartitionId(tbl.zoneId(), PART_ID);
- String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);
+ String leaseholder = waitAndGetLeaseholder(node(0), partitionGroupId);
IgniteImpl commitPartNode = findNodeByName(leaseholder);
@@ -262,11 +255,9 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
public void testAbandonedTxIsAborted() throws Exception {
TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));
- var tblReplicationGrp = colocationEnabled()
- ? new ZonePartitionId(tbl.zoneId(), PART_ID)
- : new TablePartitionId(tbl.tableId(), PART_ID);
+ var partitionGroupId = new ZonePartitionId(tbl.zoneId(), PART_ID);
- String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);
+ String leaseholder = waitAndGetLeaseholder(node(0), partitionGroupId);
IgniteImpl commitPartNode = findNodeByName(leaseholder);
@@ -307,11 +298,9 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
public void testAbandonedTxWithCoarseLockIsAborted() throws Exception {
TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));
- var tblReplicationGrp = colocationEnabled()
- ? new ZonePartitionId(tbl.zoneId(), PART_ID)
- : new TablePartitionId(tbl.tableId(), PART_ID);
+ var partitionGroupId = new ZonePartitionId(tbl.zoneId(), PART_ID);
- String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);
+ String leaseholder = waitAndGetLeaseholder(node(0), partitionGroupId);
IgniteImpl commitPartNode = findNodeByName(leaseholder);
@@ -344,11 +333,9 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
public void testWriteIntentRecoverNoCoordinator() throws Exception {
TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));
- var tblReplicationGrp = colocationEnabled()
- ? new ZonePartitionId(tbl.zoneId(), PART_ID)
- : new TablePartitionId(tbl.tableId(), PART_ID);
+ var partitionGroupId = new ZonePartitionId(tbl.zoneId(), PART_ID);
- String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);
+ String leaseholder = waitAndGetLeaseholder(node(0), partitionGroupId);
IgniteImpl commitPartNode = findNodeByName(leaseholder);
@@ -392,11 +379,9 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
public void testWriteIntentNoRecovery() throws Exception {
TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));
- var tblReplicationGrp = colocationEnabled()
- ? new ZonePartitionId(tbl.zoneId(), PART_ID)
- : new TablePartitionId(tbl.tableId(), PART_ID);
+ var partitionGroupId = new ZonePartitionId(tbl.zoneId(), PART_ID);
- String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);
+ String leaseholder = waitAndGetLeaseholder(node(0), partitionGroupId);
IgniteImpl commitPartNode = findNodeByName(leaseholder);
@@ -439,11 +424,9 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
public void testWriteIntentRecoveryAndLockConflict() throws Exception {
TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));
- var tblReplicationGrp = colocationEnabled()
- ? new ZonePartitionId(tbl.zoneId(), PART_ID)
- : new TablePartitionId(tbl.tableId(), PART_ID);
+ var partitionGroupId = new ZonePartitionId(tbl.zoneId(), PART_ID);
- String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);
+ String leaseholder = waitAndGetLeaseholder(node(0), partitionGroupId);
IgniteImpl commitPartNode = findNodeByName(leaseholder);
@@ -505,11 +488,9 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
public void testSendCommitAndDie() throws Exception {
TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));
- var tblReplicationGrp = colocationEnabled()
- ? new ZonePartitionId(tbl.zoneId(), PART_ID)
- : new TablePartitionId(tbl.tableId(), PART_ID);
+ var partitionGroupId = new ZonePartitionId(tbl.zoneId(), PART_ID);
- String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);
+ String leaseholder = waitAndGetLeaseholder(node(0), partitionGroupId);
IgniteImpl commitPartNode = findNodeByName(leaseholder);
@@ -576,11 +557,9 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
public void testCommitAndDieRecoveryFirst() throws Exception {
TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));
- var tblReplicationGrp = colocationEnabled()
- ? new ZonePartitionId(tbl.zoneId(), PART_ID)
- : new TablePartitionId(tbl.tableId(), PART_ID);
+ var partitionGroupId = new ZonePartitionId(tbl.zoneId(), PART_ID);
- String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);
+ String leaseholder = waitAndGetLeaseholder(node(0), partitionGroupId);
IgniteImpl commitPartNode = findNodeByName(leaseholder);
@@ -653,11 +632,9 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
public void testRecoveryIsTriggeredOnce() throws Exception {
TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));
- var tblReplicationGrp = colocationEnabled()
- ? new ZonePartitionId(tbl.zoneId(), PART_ID)
- : new TablePartitionId(tbl.tableId(), PART_ID);
+ var partitionGroupId = new ZonePartitionId(tbl.zoneId(), PART_ID);
- String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);
+ String leaseholder = waitAndGetLeaseholder(node(0), partitionGroupId);
IgniteImpl commitPartNode = findNodeByName(leaseholder);
@@ -732,11 +709,9 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
public void testFinishAlreadyFinishedTx() throws Exception {
TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));
- var tblReplicationGrp = colocationEnabled()
- ? new ZonePartitionId(tbl.zoneId(), PART_ID)
- : new TablePartitionId(tbl.tableId(), PART_ID);
+ var partitionGroupId = new ZonePartitionId(tbl.zoneId(), PART_ID);
- String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);
+ String leaseholder = waitAndGetLeaseholder(node(0), partitionGroupId);
IgniteImpl commitPartNode = findNodeByName(leaseholder);
@@ -756,7 +731,7 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
IgniteImpl txCrdNode2 = unwrapIgniteImpl(node(0));
- ReplicationGroupId commitPartition = ((InternalTransaction) rwTx1).commitPartition();
+ ZonePartitionId commitPartition = ((InternalTransaction) rwTx1).commitPartition();
CompletableFuture<Void> finish2 = txCrdNode2.txManager().finish(
HybridTimestampTracker.atomicTracker(null),
commitPartition,
@@ -774,11 +749,9 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
public void testPrimaryFailureRightAfterCommitMsg() throws Exception {
TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));
- var tblReplicationGrp = colocationEnabled()
- ? new ZonePartitionId(tbl.zoneId(), PART_ID)
- : new TablePartitionId(tbl.tableId(), PART_ID);
+ var partitionGroupId = new ZonePartitionId(tbl.zoneId(), PART_ID);
- String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);
+ String leaseholder = waitAndGetLeaseholder(node(0), partitionGroupId);
IgniteImpl commitPartNode = findNodeByName(leaseholder);
@@ -815,9 +788,9 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
assertThat(commitMsgSentFut, willCompleteSuccessfully());
- cancelLease(commitPartNode, tblReplicationGrp);
+ cancelLease(commitPartNode, partitionGroupId);
- waitAndGetLeaseholder(txCrdNode, tblReplicationGrp);
+ waitAndGetLeaseholder(txCrdNode, partitionGroupId);
cancelLeaseFuture.complete(null);
@@ -836,11 +809,9 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
public void testPrimaryFailureWhileInflightInProgress() throws Exception {
TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));
- var tblReplicationGrp = colocationEnabled()
- ? new ZonePartitionId(tbl.zoneId(), PART_ID)
- : new TablePartitionId(tbl.tableId(), PART_ID);
+ var partitionGroupId = new ZonePartitionId(tbl.zoneId(), PART_ID);
- String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);
+ String leaseholder = waitAndGetLeaseholder(node(0), partitionGroupId);
IgniteImpl commitPartNode = findNodeByName(leaseholder);
@@ -880,11 +851,9 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
public void testPrimaryFailureWhileInflightInProgressAfterFirstResponse() throws Exception {
TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));
- var tblReplicationGrp = colocationEnabled()
- ? new ZonePartitionId(tbl.zoneId(), PART_ID)
- : new TablePartitionId(tbl.tableId(), PART_ID);
+ var partitionGroupId = new ZonePartitionId(tbl.zoneId(), PART_ID);
- String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);
+ String leaseholder = waitAndGetLeaseholder(node(0), partitionGroupId);
IgniteImpl commitPartNode = findNodeByName(leaseholder);
@@ -919,7 +888,7 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
assertThat(firstResponseSent, willCompleteSuccessfully());
- cancelLease(commitPartNode, tblReplicationGrp);
+ cancelLease(commitPartNode, partitionGroupId);
TransactionException txEx = assertWillThrow(commitFut, TransactionException.class, 30, SECONDS);
assertThat(txEx.getCause(), instanceOf(MismatchingTransactionOutcomeInternalException.class));
@@ -937,11 +906,9 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
preloadData(tbl, 10);
- var tblReplicationGrp = colocationEnabled()
- ? new ZonePartitionId(tbl.zoneId(), PART_ID)
- : new TablePartitionId(tbl.tableId(), PART_ID);
+ var partitionGroupId = new ZonePartitionId(tbl.zoneId(), PART_ID);
- String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);
+ String leaseholder = waitAndGetLeaseholder(node(0), partitionGroupId);
IgniteImpl commitPartNode = findNodeByName(leaseholder);
@@ -992,11 +959,9 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
public void testCursorCleanup(boolean readOnly) throws Exception {
TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));
- var tblReplicationGrp = colocationEnabled()
- ? new ZonePartitionId(tbl.zoneId(), PART_ID)
- : new TablePartitionId(tbl.tableId(), PART_ID);
+ var partitionGroupId = new ZonePartitionId(tbl.zoneId(), PART_ID);
- String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);
+ String leaseholder = waitAndGetLeaseholder(node(0), partitionGroupId);
IgniteImpl targetNode = findNodeByName(leaseholder);
@@ -1078,9 +1043,7 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
if (tx.isReadOnly()) {
String primary = waitAndGetLeaseholder(
node(0),
- colocationEnabled()
- ? new ZonePartitionId(tbl.zoneId(), PART_ID)
- : new TablePartitionId(tbl.tableId(), PART_ID)
+ new ZonePartitionId(tbl.zoneId(), PART_ID)
);
InternalClusterNode primaryNode = ClusterNodeImpl.fromPublicClusterNode(
@@ -1114,11 +1077,9 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
public void testCursorsClosedAfterTxClose() throws Exception {
TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));
- var tblReplicationGrp = colocationEnabled()
- ? new ZonePartitionId(tbl.zoneId(), PART_ID)
- : new TablePartitionId(tbl.tableId(), PART_ID);
+ var partitionGroupId = new ZonePartitionId(tbl.zoneId(), PART_ID);
- String leaseholder = waitAndGetPrimaryReplica(unwrapIgniteImpl(node(0)), tblReplicationGrp).getLeaseholder();
+ String leaseholder = waitAndGetPrimaryReplica(unwrapIgniteImpl(node(0)), partitionGroupId).getLeaseholder();
IgniteImpl txExecNode = findNodeByName(leaseholder);
@@ -1192,10 +1153,8 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
private static @Nullable TxMeta txStoredMeta(IgniteImpl node, UUID txId) {
InternalTable internalTable = unwrapTableImpl(node.tables().table(TABLE_NAME)).internalTable();
- return colocationEnabled()
- ? bypassingThreadAssertions(
- () -> node.partitionReplicaLifecycleManager().txStatePartitionStorage(internalTable.zoneId(), 0).get(txId))
- : bypassingThreadAssertions(() -> internalTable.txStateStorage().getPartitionStorage(0).get(txId));
+ return bypassingThreadAssertions(
+ () -> node.partitionReplicaLifecycleManager().txStatePartitionStorage(internalTable.zoneId(), 0).get(txId));
}
/**
@@ -1303,11 +1262,11 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
return findNode(1, initialNodes(), n -> !leaseholder.equals(n.name()));
}
- private static String waitAndGetLeaseholder(Ignite node, ReplicationGroupId tblReplicationGrp) {
- return waitAndGetPrimaryReplica(unwrapIgniteImpl(node), tblReplicationGrp).getLeaseholder();
+ private static String waitAndGetLeaseholder(Ignite node, ZonePartitionId zoneReplicationGroupId) {
+ return waitAndGetPrimaryReplica(unwrapIgniteImpl(node), zoneReplicationGroupId).getLeaseholder();
}
- private void cancelLease(IgniteImpl leaseholder, ReplicationGroupId groupId) {
+ private void cancelLease(IgniteImpl leaseholder, ZonePartitionId groupId) {
StopLeaseProlongationMessage msg = PLACEMENT_DRIVER_MESSAGES_FACTORY
.stopLeaseProlongationMessage()
.groupId(groupId)
diff --git a/modules/transactions/src/integrationTest/java/org/apache/ignite/tx/distributed/ItTxResourcesVacuumTest.java b/modules/transactions/src/integrationTest/java/org/apache/ignite/tx/distributed/ItTxResourcesVacuumTest.java
index 701adb460db..d7059dd174f 100644
--- a/modules/transactions/src/integrationTest/java/org/apache/ignite/tx/distributed/ItTxResourcesVacuumTest.java
+++ b/modules/transactions/src/integrationTest/java/org/apache/ignite/tx/distributed/ItTxResourcesVacuumTest.java
@@ -20,7 +20,6 @@ package org.apache.ignite.tx.distributed;
import static java.util.stream.Collectors.toSet;
import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
import static org.apache.ignite.internal.catalog.CatalogService.DEFAULT_STORAGE_PROFILE;
-import static org.apache.ignite.internal.lang.IgniteSystemProperties.colocationEnabled;
import static org.apache.ignite.internal.sql.engine.util.SqlTestUtils.executeUpdate;
import static org.apache.ignite.internal.table.NodeUtils.transferPrimary;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.runAsync;
@@ -33,7 +32,6 @@ import static org.apache.ignite.internal.tx.test.ItTransactionTestUtils.findTupl
import static org.apache.ignite.internal.tx.test.ItTransactionTestUtils.partitionAssignment;
import static org.apache.ignite.internal.tx.test.ItTransactionTestUtils.partitionIdForTuple;
import static org.apache.ignite.internal.tx.test.ItTransactionTestUtils.table;
-import static org.apache.ignite.internal.tx.test.ItTransactionTestUtils.tableId;
import static org.apache.ignite.internal.tx.test.ItTransactionTestUtils.txId;
import static org.apache.ignite.internal.tx.test.ItTransactionTestUtils.waitAndGetPrimaryReplica;
import static org.apache.ignite.internal.tx.test.ItTransactionTestUtils.zoneId;
@@ -62,8 +60,6 @@ import org.apache.ignite.internal.app.IgniteImpl;
import org.apache.ignite.internal.configuration.SystemDistributedConfiguration;
import org.apache.ignite.internal.configuration.SystemDistributedExtensionConfiguration;
import org.apache.ignite.internal.placementdriver.ReplicaMeta;
-import org.apache.ignite.internal.replicator.PartitionGroupId;
-import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.table.InternalTable;
import org.apache.ignite.internal.testframework.SystemPropertiesExtension;
@@ -211,9 +207,7 @@ public class ItTxResourcesVacuumTest extends ClusterPerTestIntegrationTest {
Set<String> nodes = partitionAssignment(
node,
- colocationEnabled()
- ? new ZonePartitionId(zoneId(node, TABLE_NAME), partId)
- : new TablePartitionId(tableId(node, TABLE_NAME), partId)
+ new ZonePartitionId(zoneId(node, TABLE_NAME), partId)
);
view.upsert(tx, tuple);
@@ -306,9 +300,7 @@ public class ItTxResourcesVacuumTest extends ClusterPerTestIntegrationTest {
int partId = partitionIdForTuple(anyNode(), TABLE_NAME, tuple, null);
- PartitionGroupId groupId = colocationEnabled()
- ? new ZonePartitionId(zoneId(anyNode(), TABLE_NAME), partId)
- : new TablePartitionId(tableId(anyNode(), TABLE_NAME), partId);
+ ZonePartitionId groupId = new ZonePartitionId(zoneId(anyNode(), TABLE_NAME), partId);
ReplicaMeta replicaMeta = waitAndGetPrimaryReplica(anyNode(), groupId);
assertNotNull(replicaMeta);
@@ -393,18 +385,14 @@ public class ItTxResourcesVacuumTest extends ClusterPerTestIntegrationTest {
int commitPartId = partitionIdForTuple(node, TABLE_NAME, tuple0, tx);
- PartitionGroupId commitPartGrpId = colocationEnabled()
- ? new ZonePartitionId(zoneId(node, TABLE_NAME), commitPartId)
- : new TablePartitionId(tableId(node, TABLE_NAME), commitPartId);
+ ZonePartitionId commitPartGrpId = new ZonePartitionId(zoneId(node, TABLE_NAME), commitPartId);
ReplicaMeta replicaMeta = waitAndGetPrimaryReplica(node, commitPartGrpId);
IgniteImpl commitPartitionLeaseholder = findNode(n -> n.id().equals(replicaMeta.getLeaseholderId()));
Set<String> commitPartNodes = partitionAssignment(
node,
- colocationEnabled()
- ? new ZonePartitionId(zoneId(node, TABLE_NAME), commitPartId)
- : new TablePartitionId(tableId(node, TABLE_NAME), commitPartId)
+ new ZonePartitionId(zoneId(node, TABLE_NAME), commitPartId)
);
log.info("Test: Commit partition [part={}, leaseholder={}, hostingNodes={}].", commitPartGrpId, commitPartitionLeaseholder.name(),
@@ -506,18 +494,14 @@ public class ItTxResourcesVacuumTest extends ClusterPerTestIntegrationTest {
int commitPartId = partitionIdForTuple(node, TABLE_NAME, tuple, tx);
- PartitionGroupId commitPartGrpId = colocationEnabled()
- ? new ZonePartitionId(zoneId(node, TABLE_NAME), commitPartId)
- : new TablePartitionId(tableId(node, TABLE_NAME), commitPartId);
+ ZonePartitionId commitPartGrpId = new ZonePartitionId(zoneId(node, TABLE_NAME), commitPartId);
ReplicaMeta replicaMeta = waitAndGetPrimaryReplica(node, commitPartGrpId);
IgniteImpl commitPartitionLeaseholder = findNode(n -> n.id().equals(replicaMeta.getLeaseholderId()));
Set<String> commitPartNodes = partitionAssignment(
node,
- colocationEnabled()
- ? new ZonePartitionId(zoneId(node, TABLE_NAME), commitPartId)
- : new TablePartitionId(tableId(node, TABLE_NAME), commitPartId)
+ new ZonePartitionId(zoneId(node, TABLE_NAME), commitPartId)
);
log.info("Test: Commit partition [leaseholder={}, hostingNodes={}].", commitPartitionLeaseholder.name(), commitPartNodes);
@@ -604,18 +588,14 @@ public class ItTxResourcesVacuumTest extends ClusterPerTestIntegrationTest {
int commitPartId = partitionIdForTuple(node, TABLE_NAME, tuple, tx);
- PartitionGroupId commitPartGrpId = colocationEnabled()
- ? new ZonePartitionId(zoneId(node, TABLE_NAME), commitPartId)
- : new TablePartitionId(tableId(node, TABLE_NAME), commitPartId);
+ ZonePartitionId commitPartGrpId = new ZonePartitionId(zoneId(node, TABLE_NAME), commitPartId);
ReplicaMeta replicaMeta = waitAndGetPrimaryReplica(node, commitPartGrpId);
IgniteImpl commitPartitionLeaseholder = findNode(n -> n.id().equals(replicaMeta.getLeaseholderId()));
Set<String> commitPartNodes = partitionAssignment(
node,
- colocationEnabled()
- ? new ZonePartitionId(zoneId(node, TABLE_NAME), commitPartId)
- : new TablePartitionId(tableId(node, TABLE_NAME), commitPartId)
+ new ZonePartitionId(zoneId(node, TABLE_NAME), commitPartId)
);
log.info("Test: Commit partition [part={}, leaseholder={}, hostingNodes={}].", commitPartGrpId, commitPartitionLeaseholder.name(),
@@ -707,9 +687,7 @@ public class ItTxResourcesVacuumTest extends ClusterPerTestIntegrationTest {
Set<String> commitPartitionNodes = partitionAssignment(
commitPartitionLeaseholder,
- colocationEnabled()
- ? new ZonePartitionId(zoneId(commitPartitionLeaseholder, TABLE_NAME), commitPartId)
- : new TablePartitionId(tableId(commitPartitionLeaseholder, TABLE_NAME), commitPartId)
+ new ZonePartitionId(zoneId(commitPartitionLeaseholder, TABLE_NAME), commitPartId)
);
// Choose some node that doesn't host the partition as a tx coordinator.
@@ -1058,9 +1036,7 @@ public class ItTxResourcesVacuumTest extends ClusterPerTestIntegrationTest {
if (checkCpPrimaryOnly) {
IgniteImpl node = anyNode();
- PartitionGroupId partitionGroupId = colocationEnabled()
- ? new ZonePartitionId(zoneId(node, tableName), partId)
- : new TablePartitionId(tableId(node, tableName), partId);
+ ZonePartitionId partitionGroupId = new ZonePartitionId(zoneId(node, tableName), partId);
CompletableFuture<ReplicaMeta> replicaFut = node.placementDriver().getPrimaryReplica(partitionGroupId, node.clock().now());
assertThat(replicaFut, willCompleteSuccessfully());
@@ -1132,10 +1108,9 @@ public class ItTxResourcesVacuumTest extends ClusterPerTestIntegrationTest {
return runInExecutor(txStateStorageExecutor, () -> {
InternalTable internalTable = table(node, tableName).internalTable();
- TxStatePartitionStorage txStatePartitionStorage =
- colocationEnabled()
- ? node.partitionReplicaLifecycleManager().txStatePartitionStorage(internalTable.zoneId(), partId)
- : internalTable.txStateStorage().getPartitionStorage(partId);
+ TxStatePartitionStorage txStatePartitionStorage = node
+ .partitionReplicaLifecycleManager()
+ .txStatePartitionStorage(internalTable.zoneId(), partId);
assertNotNull(txStatePartitionStorage);
diff --git a/modules/transactions/src/jmh/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistryBenchmark.java b/modules/transactions/src/jmh/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistryBenchmark.java
index 51da08d4e1b..15ee814979a 100644
--- a/modules/transactions/src/jmh/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistryBenchmark.java
+++ b/modules/transactions/src/jmh/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistryBenchmark.java
@@ -26,8 +26,7 @@ import java.util.List;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.hlc.HybridTimestamp;
-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.tx.InternalTransaction;
import org.apache.ignite.internal.tx.PendingTxPartitionEnlistment;
import org.apache.ignite.internal.tx.TxState;
@@ -121,7 +120,7 @@ public class TransactionExpirationRegistryBenchmark {
}
@Override
- public PendingTxPartitionEnlistment enlistedPartition(ReplicationGroupId replicationGroupId) {
+ public PendingTxPartitionEnlistment enlistedPartition(ZonePartitionId replicationGroupId) {
return null;
}
@@ -131,18 +130,18 @@ public class TransactionExpirationRegistryBenchmark {
}
@Override
- public boolean assignCommitPartition(ReplicationGroupId commitPartitionId) {
+ public boolean assignCommitPartition(ZonePartitionId commitPartitionId) {
return false;
}
@Override
- public TablePartitionId commitPartition() {
+ public ZonePartitionId commitPartition() {
return null;
}
@Override
public void enlist(
- ReplicationGroupId replicationGroupId,
+ ZonePartitionId replicationGroupId,
int tableId,
String primaryNodeConsistentId,
long consistencyToken
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/InternalTransaction.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/InternalTransaction.java
index e9b420e8206..0e3b42aa1b3 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/InternalTransaction.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/InternalTransaction.java
@@ -20,7 +20,7 @@ package org.apache.ignite.internal.tx;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.hlc.HybridTimestamp;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.tx.Transaction;
import org.jetbrains.annotations.Nullable;
@@ -44,7 +44,7 @@ public interface InternalTransaction extends Transaction {
* @param replicationGroupId Replication group ID.
* @return Enlisted partition information.
*/
- PendingTxPartitionEnlistment enlistedPartition(ReplicationGroupId replicationGroupId);
+ PendingTxPartitionEnlistment enlistedPartition(ZonePartitionId replicationGroupId);
/**
* Returns a transaction state.
@@ -59,14 +59,14 @@ public interface InternalTransaction extends Transaction {
* @param commitPartitionId Commit partition group id.
* @return True if the partition was assigned as committed, false otherwise.
*/
- boolean assignCommitPartition(ReplicationGroupId commitPartitionId);
+ boolean assignCommitPartition(ZonePartitionId commitPartitionId);
/**
* Gets a partition id that stores the transaction state.
*
* @return Partition id.
*/
- ReplicationGroupId commitPartition();
+ ZonePartitionId commitPartition();
/**
* Enlists a partition group into a transaction.
@@ -77,7 +77,7 @@ public interface InternalTransaction extends Transaction {
* @param consistencyToken Consistency token to enlist for given replication group.
*/
void enlist(
- ReplicationGroupId replicationGroupId,
+ ZonePartitionId replicationGroupId,
int tableId,
String primaryNodeConsistentId,
long consistencyToken
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxManager.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxManager.java
index 290e1437783..a6fadd212a6 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxManager.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxManager.java
@@ -26,7 +26,6 @@ import java.util.function.Function;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.hlc.HybridTimestampTracker;
import org.apache.ignite.internal.manager.IgniteComponent;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.tx.impl.EnlistedPartitionGroup;
import org.apache.ignite.internal.tx.metrics.ResourceVacuumMetrics;
@@ -183,11 +182,11 @@ public interface TxManager extends IgniteComponent {
*/
CompletableFuture<Void> finish(
HybridTimestampTracker timestampTracker,
- @Nullable ReplicationGroupId commitPartition,
+ @Nullable ZonePartitionId commitPartition,
boolean commitIntent,
boolean timeoutExceeded,
boolean recovery,
- Map<ReplicationGroupId, PendingTxPartitionEnlistment> enlistedGroups,
+ Map<ZonePartitionId, PendingTxPartitionEnlistment> enlistedGroups,
UUID txId
);
@@ -204,8 +203,8 @@ public interface TxManager extends IgniteComponent {
* @return Completable future of Void.
*/
CompletableFuture<Void> cleanup(
- @Nullable ReplicationGroupId commitPartitionId,
- Map<ReplicationGroupId, PartitionEnlistment> enlistedPartitions,
+ @Nullable ZonePartitionId commitPartitionId,
+ Map<ZonePartitionId, PartitionEnlistment> enlistedPartitions,
boolean commit,
@Nullable HybridTimestamp commitTimestamp,
UUID txId
@@ -224,7 +223,7 @@ public interface TxManager extends IgniteComponent {
* @return Completable future of Void.
*/
CompletableFuture<Void> cleanup(
- @Nullable ReplicationGroupId commitPartitionId,
+ @Nullable ZonePartitionId commitPartitionId,
Collection<EnlistedPartitionGroup> enlistedPartitions,
boolean commit,
@Nullable HybridTimestamp commitTimestamp,
@@ -239,7 +238,7 @@ public interface TxManager extends IgniteComponent {
* @param txId Transaction id.
* @return Completable future of Void.
*/
- CompletableFuture<Void> cleanup(ReplicationGroupId commitPartitionId, String node, UUID txId);
+ CompletableFuture<Void> cleanup(ZonePartitionId commitPartitionId, String node, UUID txId);
/**
* Locally vacuums no longer needed transactional resources, like txnState both persistent and volatile.
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxMetaSerializer.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxMetaSerializer.java
index 2663acbd278..561186a4202 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxMetaSerializer.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxMetaSerializer.java
@@ -22,9 +22,6 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import org.apache.ignite.internal.hlc.HybridTimestamp;
-import org.apache.ignite.internal.replicator.PartitionGroupId;
-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.tx.impl.EnlistedPartitionGroup;
import org.apache.ignite.internal.util.io.IgniteDataInput;
@@ -45,18 +42,15 @@ public class TxMetaSerializer extends VersionedSerializer<TxMeta> {
@Override
protected void writeExternalData(TxMeta meta, IgniteDataOutput out) throws IOException {
- boolean hasAnyZonePartitionIds = hasAnyZonePartitionIds(meta);
- if (hasAnyZonePartitionIds) {
- assert hasNoTablePartitionIds(meta) : "Both table-based and zone-based partition IDs: " + meta.enlistedPartitions();
- }
-
+ // Write this boolean flag (which is always true) due to compatibility reasons.
+ boolean hasAnyZonePartitionIds = true;
out.writeBoolean(hasAnyZonePartitionIds);
out.writeVarInt(meta.txState().id());
out.writeVarInt(meta.enlistedPartitions().size());
for (EnlistedPartitionGroup enlistedPartitionGroup : meta.enlistedPartitions()) {
- PartitionGroupId partitionGroupId = (PartitionGroupId) enlistedPartitionGroup.groupId();
+ ZonePartitionId partitionGroupId = enlistedPartitionGroup.groupId();
out.writeVarInt(partitionGroupId.objectId());
out.writeVarInt(partitionGroupId.partitionId());
@@ -66,34 +60,28 @@ public class TxMetaSerializer extends VersionedSerializer<TxMeta> {
HybridTimestamp.write(meta.commitTimestamp(), out);
}
- private static boolean hasAnyZonePartitionIds(TxMeta meta) {
- return meta.enlistedPartitions().stream()
- .anyMatch(partition -> partition.groupId() instanceof ZonePartitionId);
- }
-
- private static boolean hasNoTablePartitionIds(TxMeta meta) {
- return meta.enlistedPartitions().stream()
- .noneMatch(partition -> partition.groupId() instanceof TablePartitionId);
- }
-
@Override
protected TxMeta readExternalData(byte protoVer, IgniteDataInput in) throws IOException {
- boolean usesZonePartitionIds;
+ // V1 version is not supported anymore.
+ if (protoVer == 1) {
+ throw new IllegalArgumentException("Transaction meta v1 is not supported.");
+ }
+
if (protoVer >= 2) {
- usesZonePartitionIds = in.readBoolean();
- } else {
- usesZonePartitionIds = false;
+ boolean usesZonePartitionIds = in.readBoolean();
+ assert usesZonePartitionIds : "Transaction meta cannot be deserialized. V1 is not supported anymore.";
}
TxState state = TxState.fromId(in.readVarIntAsInt());
- List<EnlistedPartitionGroup> enlistedPartitions = readEnlistedPartitions(in, protoVer, usesZonePartitionIds);
+ List<EnlistedPartitionGroup> enlistedPartitions = readEnlistedPartitions(in, protoVer);
HybridTimestamp commitTimestamp = HybridTimestamp.readNullableFrom(in);
return new TxMeta(state, enlistedPartitions, commitTimestamp);
}
- private static List<EnlistedPartitionGroup> readEnlistedPartitions(IgniteDataInput in, byte protoVer, boolean usesZonePartitionIds)
- throws IOException {
+ private static List<EnlistedPartitionGroup> readEnlistedPartitions(IgniteDataInput in, byte protoVer) throws IOException {
+ assert protoVer >= 2 : "Unsupported protocol version [ver=" + protoVer + "].";
+
int length = in.readVarIntAsInt();
List<EnlistedPartitionGroup> enlistedPartitions = new ArrayList<>(length);
@@ -101,25 +89,15 @@ public class TxMetaSerializer extends VersionedSerializer<TxMeta> {
int objectId = in.readVarIntAsInt();
int partitionId = in.readVarIntAsInt();
- Set<Integer> tableIds;
- if (protoVer >= 2) {
- tableIds = readVarIntSet(in);
- } else {
- // In version 1 only table-based partitions were supported, so tableIds will be a singleton.
- tableIds = Set.of(objectId);
- }
+ Set<Integer> tableIds = readVarIntSet(in);
- enlistedPartitions.add(new EnlistedPartitionGroup(replicationGroupId(objectId, partitionId, usesZonePartitionIds), tableIds));
+ enlistedPartitions.add(
+ new EnlistedPartitionGroup(
+ new ZonePartitionId(objectId, partitionId),
+ tableIds
+ ));
}
return enlistedPartitions;
}
-
- private static ReplicationGroupId replicationGroupId(int objectId, int partitionId, boolean usesZonePartitionIds) {
- if (usesZonePartitionIds) {
- return new ZonePartitionId(objectId, partitionId);
- } else {
- return new TablePartitionId(objectId, partitionId);
- }
- }
}
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxStateMeta.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxStateMeta.java
index 1647b4ed65d..8c02d1d88c4 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxStateMeta.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxStateMeta.java
@@ -18,7 +18,7 @@
package org.apache.ignite.internal.tx;
import static java.util.Objects.requireNonNull;
-import static org.apache.ignite.internal.replicator.message.ReplicaMessageUtils.toReplicationGroupIdMessage;
+import static org.apache.ignite.internal.replicator.message.ReplicaMessageUtils.toZonePartitionIdMessage;
import static org.apache.ignite.internal.tx.TxState.ABANDONED;
import static org.apache.ignite.internal.tx.TxState.FINISHING;
import static org.apache.ignite.internal.tx.TxState.checkTransitionCorrectness;
@@ -26,7 +26,7 @@ import static org.apache.ignite.internal.util.FastTimestamps.coarseCurrentTimeMi
import java.util.UUID;
import org.apache.ignite.internal.hlc.HybridTimestamp;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.replicator.message.ReplicaMessagesFactory;
import org.apache.ignite.internal.tostring.IgniteToStringExclude;
import org.apache.ignite.internal.tostring.S;
@@ -45,7 +45,7 @@ public class TxStateMeta implements TransactionMeta {
private final @Nullable UUID txCoordinatorId;
/** ID of the replication group that manages a transaction state. */
- private final @Nullable ReplicationGroupId commitPartitionId;
+ private final @Nullable ZonePartitionId commitPartitionId;
private final @Nullable HybridTimestamp commitTimestamp;
@@ -78,8 +78,7 @@ public class TxStateMeta implements TransactionMeta {
public TxStateMeta(
TxState txState,
@Nullable UUID txCoordinatorId,
- // TODO Ignite-22522 Use partition instead.
- @Nullable ReplicationGroupId commitPartitionId,
+ @Nullable ZonePartitionId commitPartitionId,
@Nullable HybridTimestamp commitTimestamp,
@Nullable InternalTransaction tx,
@Nullable Boolean isFinishedDueToTimeout
@@ -103,7 +102,7 @@ public class TxStateMeta implements TransactionMeta {
public TxStateMeta(
TxState txState,
@Nullable UUID txCoordinatorId,
- @Nullable ReplicationGroupId commitPartitionId,
+ @Nullable ZonePartitionId commitPartitionId,
@Nullable HybridTimestamp commitTimestamp,
@Nullable InternalTransaction tx,
@Nullable Long initialVacuumObservationTimestamp,
@@ -165,7 +164,7 @@ public class TxStateMeta implements TransactionMeta {
return txCoordinatorId;
}
- public @Nullable ReplicationGroupId commitPartitionId() {
+ public @Nullable ZonePartitionId commitPartitionId() {
return commitPartitionId;
}
@@ -199,7 +198,7 @@ public class TxStateMeta implements TransactionMeta {
.txState(txState)
.txCoordinatorId(txCoordinatorId)
.commitPartitionId(
- commitPartitionId == null ? null : toReplicationGroupIdMessage(replicaMessagesFactory, commitPartitionId)
+ commitPartitionId == null ? null : toZonePartitionIdMessage(replicaMessagesFactory, commitPartitionId)
)
.commitTimestamp(commitTimestamp)
.initialVacuumObservationTimestamp(initialVacuumObservationTimestamp)
@@ -250,7 +249,7 @@ public class TxStateMeta implements TransactionMeta {
public static class TxStateMetaBuilder {
protected TxState txState;
protected @Nullable UUID txCoordinatorId;
- protected @Nullable ReplicationGroupId commitPartitionId;
+ protected @Nullable ZonePartitionId commitPartitionId;
private @Nullable HybridTimestamp commitTimestamp;
private @Nullable Long initialVacuumObservationTimestamp;
private @Nullable Long cleanupCompletionTimestamp;
@@ -301,7 +300,7 @@ public class TxStateMeta implements TransactionMeta {
* @param commitPartitionId Commit partition replication group id.
* @return Builder.
*/
- public TxStateMetaBuilder commitPartitionId(@Nullable ReplicationGroupId commitPartitionId) {
+ public TxStateMetaBuilder commitPartitionId(@Nullable ZonePartitionId commitPartitionId) {
if (commitPartitionId != null) {
this.commitPartitionId = commitPartitionId;
}
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxStateMetaAbandoned.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxStateMetaAbandoned.java
index 99fcd39d9ca..df7e8f2fcfe 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxStateMetaAbandoned.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxStateMetaAbandoned.java
@@ -17,11 +17,11 @@
package org.apache.ignite.internal.tx;
-import static org.apache.ignite.internal.replicator.message.ReplicaMessageUtils.toReplicationGroupIdMessage;
+import static org.apache.ignite.internal.replicator.message.ReplicaMessageUtils.toZonePartitionIdMessage;
import static org.apache.ignite.internal.tx.TxState.ABANDONED;
import java.util.UUID;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.replicator.message.ReplicaMessagesFactory;
import org.apache.ignite.internal.tostring.S;
import org.apache.ignite.internal.tx.message.TxMessagesFactory;
@@ -47,7 +47,7 @@ public class TxStateMetaAbandoned extends TxStateMeta {
*/
public TxStateMetaAbandoned(
UUID txCoordinatorId,
- ReplicationGroupId commitPartitionId,
+ ZonePartitionId commitPartitionId,
@Nullable String txLabel
) {
super(ABANDONED, txCoordinatorId, commitPartitionId, null, null, null, null, null, txLabel);
@@ -69,13 +69,13 @@ public class TxStateMetaAbandoned extends TxStateMeta {
ReplicaMessagesFactory replicaMessagesFactory,
TxMessagesFactory txMessagesFactory
) {
- ReplicationGroupId commitPartitionId = commitPartitionId();
+ ZonePartitionId commitPartitionId = commitPartitionId();
return txMessagesFactory.txStateMetaAbandonedMessage()
.txState(txState())
.txCoordinatorId(txCoordinatorId())
.commitPartitionId(
- commitPartitionId == null ? null : toReplicationGroupIdMessage(replicaMessagesFactory, commitPartitionId)
+ commitPartitionId == null ? null : toZonePartitionIdMessage(replicaMessagesFactory, commitPartitionId)
)
.commitTimestamp(commitTimestamp())
.initialVacuumObservationTimestamp(initialVacuumObservationTimestamp())
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxStateMetaFinishing.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxStateMetaFinishing.java
index 30738e04811..d5aee32faf6 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxStateMetaFinishing.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxStateMetaFinishing.java
@@ -20,7 +20,7 @@ package org.apache.ignite.internal.tx;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.hlc.HybridTimestamp;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.replicator.message.ReplicaMessagesFactory;
import org.apache.ignite.internal.tx.message.TxMessagesFactory;
import org.apache.ignite.internal.tx.message.TxStateMetaMessage;
@@ -46,7 +46,7 @@ public class TxStateMetaFinishing extends TxStateMeta {
*/
public TxStateMetaFinishing(
@Nullable UUID txCoordinatorId,
- @Nullable ReplicationGroupId commitPartitionId,
+ @Nullable ZonePartitionId commitPartitionId,
@Nullable Boolean isFinishingDueToTimeout,
@Nullable String txLabel
) {
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/EnlistedPartitionGroup.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/EnlistedPartitionGroup.java
index 9ec0e1c5731..1226525b640 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/EnlistedPartitionGroup.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/EnlistedPartitionGroup.java
@@ -19,8 +19,7 @@ package org.apache.ignite.internal.tx.impl;
import java.util.Objects;
import java.util.Set;
-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.tostring.IgniteToStringInclude;
import org.apache.ignite.internal.tostring.S;
@@ -28,28 +27,21 @@ import org.apache.ignite.internal.tostring.S;
* Partition enlistment information together with partition group ID.
*/
public class EnlistedPartitionGroup {
- private final ReplicationGroupId groupId;
+ private final ZonePartitionId groupId;
@IgniteToStringInclude
private final Set<Integer> tableIds;
/** Constructor. */
- // TODO IGNITE-22522 Convert ReplicationGroupId to ZonePartitionId.
- public EnlistedPartitionGroup(ReplicationGroupId groupId, Set<Integer> tableIds) {
+ public EnlistedPartitionGroup(ZonePartitionId groupId, Set<Integer> tableIds) {
this.groupId = groupId;
this.tableIds = Set.copyOf(tableIds);
}
- /** Constructor. */
- // TODO IGNITE-22522 Remove.
- public EnlistedPartitionGroup(TablePartitionId groupId) {
- this(groupId, Set.of(groupId.tableId()));
- }
-
/**
* Returns replication group ID of the partition.
*/
- public ReplicationGroupId groupId() {
+ public ZonePartitionId groupId() {
return groupId;
}
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/OrphanDetector.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/OrphanDetector.java
index 2fb71270023..f6b942a0eab 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/OrphanDetector.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/OrphanDetector.java
@@ -39,7 +39,7 @@ import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.network.InternalClusterNode;
import org.apache.ignite.internal.network.TopologyService;
import org.apache.ignite.internal.replicator.ReplicaService;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.replicator.message.ReplicaMessagesFactory;
import org.apache.ignite.internal.tx.LockManager;
import org.apache.ignite.internal.tx.TxStateMeta;
@@ -196,7 +196,7 @@ public class OrphanDetector {
* @param cmpPartGrp Replication group of commit partition.
* @param txId Transaction id.
*/
- private void sendTxRecoveryMessage(ReplicationGroupId cmpPartGrp, UUID txId) {
+ private void sendTxRecoveryMessage(ZonePartitionId cmpPartGrp, UUID txId) {
placementDriverHelper.awaitPrimaryReplicaWithExceptionHandling(cmpPartGrp)
.thenCompose(replicaMeta -> {
InternalClusterNode commitPartPrimaryNode = topologyService.getByConsistentId(replicaMeta.getLeaseholder());
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PersistentTxStateVacuumizer.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PersistentTxStateVacuumizer.java
index c7b35b045eb..6ec051c40fb 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PersistentTxStateVacuumizer.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PersistentTxStateVacuumizer.java
@@ -18,7 +18,7 @@
package org.apache.ignite.internal.tx.impl;
import static java.util.stream.Collectors.toSet;
-import static org.apache.ignite.internal.replicator.message.ReplicaMessageUtils.toReplicationGroupIdMessage;
+import static org.apache.ignite.internal.replicator.message.ReplicaMessageUtils.toZonePartitionIdMessage;
import static org.apache.ignite.internal.util.CompletableFutures.allOf;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.ExceptionUtils.hasCause;
@@ -44,7 +44,7 @@ import org.apache.ignite.internal.network.InternalClusterNode;
import org.apache.ignite.internal.placementdriver.PlacementDriver;
import org.apache.ignite.internal.raft.GroupOverloadedException;
import org.apache.ignite.internal.replicator.ReplicaService;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.replicator.exception.AwaitReplicaTimeoutException;
import org.apache.ignite.internal.replicator.exception.PrimaryReplicaMissException;
import org.apache.ignite.internal.replicator.message.ReplicaMessagesFactory;
@@ -102,9 +102,7 @@ public class PersistentTxStateVacuumizer {
* @return A future, result is the set of successfully processed txn states and count of persistent states that were vacuumized.
*/
public CompletableFuture<PersistentTxStateVacuumResult> vacuumPersistentTxStates(
- // TODO https://issues.apache.org/jira/browse/IGNITE-22522
- // Should be changed to ZonePartitionId.
- Map<ReplicationGroupId, Set<VacuumizableTx>> txIds
+ Map<ZonePartitionId, Set<VacuumizableTx>> txIds
) {
Set<UUID> successful = ConcurrentHashMap.newKeySet();
List<CompletableFuture<?>> futures = new ArrayList<>();
@@ -139,9 +137,7 @@ public class PersistentTxStateVacuumizer {
VacuumTxStateReplicaRequest request = TX_MESSAGES_FACTORY.vacuumTxStateReplicaRequest()
.enlistmentConsistencyToken(replicaMeta.getStartTime().longValue())
- // TODO https://issues.apache.org/jira/browse/IGNITE-22522
- // Should be changed to ZonePartitionId.
- .groupId(toReplicationGroupIdMessage(REPLICA_MESSAGES_FACTORY, commitPartitionId))
+ .groupId(toZonePartitionIdMessage(REPLICA_MESSAGES_FACTORY, commitPartitionId))
.transactionIds(filteredTxIds)
.build();
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PlacementDriverHelper.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PlacementDriverHelper.java
index d4a83cff41e..d9cc060254a 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PlacementDriverHelper.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PlacementDriverHelper.java
@@ -40,7 +40,7 @@ import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.placementdriver.PlacementDriver;
import org.apache.ignite.internal.placementdriver.ReplicaMeta;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.tx.TransactionException;
/**
@@ -76,7 +76,7 @@ public class PlacementDriverHelper {
* @return Future that completes with node id that is a primary for the provided partition, or completes with exception if no primary
* appeared during the await timeout.
*/
- public CompletableFuture<ReplicaMeta> awaitPrimaryReplicaWithExceptionHandling(ReplicationGroupId partitionId) {
+ public CompletableFuture<ReplicaMeta> awaitPrimaryReplicaWithExceptionHandling(ZonePartitionId partitionId) {
return awaitPrimaryReplicaWithExceptionHandling(partitionId, AWAIT_PRIMARY_REPLICA_TIMEOUT, SECONDS);
}
@@ -90,7 +90,7 @@ public class PlacementDriverHelper {
* appeared during the await timeout.
*/
public CompletableFuture<ReplicaMeta> awaitPrimaryReplicaWithExceptionHandling(
- ReplicationGroupId partitionId,
+ ZonePartitionId partitionId,
long timeout,
TimeUnit timeUnit
) {
@@ -100,7 +100,7 @@ public class PlacementDriverHelper {
}
private CompletableFuture<ReplicaMeta> awaitPrimaryReplicaWithExceptionHandling(
- ReplicationGroupId partitionId,
+ ZonePartitionId partitionId,
HybridTimestamp timestamp,
long timeout,
TimeUnit timeUnit
@@ -112,7 +112,7 @@ public class PlacementDriverHelper {
throw withCause(TransactionException::new, REPLICA_UNAVAILABLE_ERR,
"Failed to get the primary replica"
- + " [tablePartitionId=" + partitionId + ", awaitTimestamp=" + timestamp + ']', e);
+ + " [zonePartitionId=" + partitionId + ", awaitTimestamp=" + timestamp + ']', e);
}
return primaryReplica;
@@ -126,7 +126,7 @@ public class PlacementDriverHelper {
* @return A future that completes with a map of node to the partitions the node is primary for and a collection of partitions that we
* failed to find the primary for.
*/
- public CompletableFuture<PartitionData> findPrimaryReplicas(Collection<ReplicationGroupId> partitions) {
+ public CompletableFuture<PartitionData> findPrimaryReplicas(Collection<ZonePartitionId> partitions) {
// Please note that we are using `get primary replica` instead of `await primary replica`.
// This method is faster, yet we still have the correctness:
// If the primary replica has not changed, get will return a valid value and we'll send an unlock request to this node.
@@ -141,8 +141,8 @@ public class PlacementDriverHelper {
* @param partitions A collection of partitions.
* @return A future that completes with a map of node to the partitions the node is primary for.
*/
- public CompletableFuture<Map<String, Set<ReplicationGroupId>>> awaitPrimaryReplicas(Collection<ReplicationGroupId> partitions) {
- BiFunction<ReplicationGroupId, HybridTimestamp, CompletableFuture<ReplicaMeta>> action = (groupId, timestamp)
+ public CompletableFuture<Map<String, Set<ZonePartitionId>>> awaitPrimaryReplicas(Collection<ZonePartitionId> partitions) {
+ BiFunction<ZonePartitionId, HybridTimestamp, CompletableFuture<ReplicaMeta>> action = (groupId, timestamp)
-> awaitPrimaryReplicaWithExceptionHandling(groupId, timestamp, AWAIT_PRIMARY_REPLICA_TIMEOUT, SECONDS);
return computePrimaryReplicas(partitions, action)
.thenApply(partitionData -> partitionData.partitionsByNode);
@@ -156,8 +156,8 @@ public class PlacementDriverHelper {
* failed to find the primary for.
*/
private CompletableFuture<PartitionData> computePrimaryReplicas(
- Collection<ReplicationGroupId> partitions,
- BiFunction<ReplicationGroupId, HybridTimestamp, CompletableFuture<ReplicaMeta>> placementFunction
+ Collection<ZonePartitionId> partitions,
+ BiFunction<ZonePartitionId, HybridTimestamp, CompletableFuture<ReplicaMeta>> placementFunction
) {
if (partitions == null || partitions.isEmpty()) {
return completedFuture(new PartitionData(emptyMap(), emptySet()));
@@ -165,23 +165,23 @@ public class PlacementDriverHelper {
HybridTimestamp timestamp = clockService.now();
- Map<ReplicationGroupId, CompletableFuture<ReplicaMeta>> primaryReplicaFutures = new HashMap<>();
+ Map<ZonePartitionId, CompletableFuture<ReplicaMeta>> primaryReplicaFutures = new HashMap<>();
- for (ReplicationGroupId partitionId : partitions) {
+ for (ZonePartitionId partitionId : partitions) {
primaryReplicaFutures.put(partitionId, placementFunction.apply(partitionId, timestamp));
}
return allOf(primaryReplicaFutures.values().toArray(new CompletableFuture<?>[0]))
.thenApply(v -> {
- Map<String, Set<ReplicationGroupId>> partitionsByNode = new HashMap<>();
+ Map<String, Set<ZonePartitionId>> partitionsByNode = new HashMap<>();
- Set<ReplicationGroupId> partitionsWithoutPrimary = new HashSet<>();
+ Set<ZonePartitionId> partitionsWithoutPrimary = new HashSet<>();
- for (Entry<ReplicationGroupId, CompletableFuture<ReplicaMeta>> entry : primaryReplicaFutures.entrySet()) {
+ for (Entry<ZonePartitionId, CompletableFuture<ReplicaMeta>> entry : primaryReplicaFutures.entrySet()) {
// Safe to call join, the future has already finished.
ReplicaMeta meta = entry.getValue().join();
- ReplicationGroupId partition = entry.getKey();
+ ZonePartitionId partition = entry.getKey();
if (meta != null && meta.getLeaseholder() != null) {
partitionsByNode.computeIfAbsent(meta.getLeaseholder(), s -> new HashSet<>())
@@ -198,11 +198,11 @@ public class PlacementDriverHelper {
* The result of retrieving primary replicas for a collection of partitions.
*/
public static class PartitionData {
- final Map<String, Set<ReplicationGroupId>> partitionsByNode;
+ final Map<String, Set<ZonePartitionId>> partitionsByNode;
- final Set<ReplicationGroupId> partitionsWithoutPrimary;
+ final Set<ZonePartitionId> partitionsWithoutPrimary;
- PartitionData(Map<String, Set<ReplicationGroupId>> partitionsByNode, Set<ReplicationGroupId> partitionsWithoutPrimary) {
+ PartitionData(Map<String, Set<ZonePartitionId>> partitionsByNode, Set<ZonePartitionId> partitionsWithoutPrimary) {
this.partitionsByNode = partitionsByNode;
this.partitionsWithoutPrimary = partitionsWithoutPrimary;
}
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PrimaryReplicaExpiredException.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PrimaryReplicaExpiredException.java
index 92130f03f70..fc951e706ba 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PrimaryReplicaExpiredException.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PrimaryReplicaExpiredException.java
@@ -22,7 +22,7 @@ import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_PRIMARY_REPLICA
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.lang.IgniteInternalException;
import org.apache.ignite.internal.placementdriver.ReplicaMeta;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.tx.RetriableTransactionException;
import org.jetbrains.annotations.Nullable;
@@ -37,7 +37,7 @@ public class PrimaryReplicaExpiredException extends IgniteInternalException impl
* @param currentPrimaryReplica Current primary replica
*/
public PrimaryReplicaExpiredException(
- ReplicationGroupId groupId,
+ ZonePartitionId groupId,
long expectedEnlistmentConsistencyToken,
@Nullable HybridTimestamp commitTimestamp,
@Nullable ReplicaMeta currentPrimaryReplica
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PublicApiThreadingTransaction.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PublicApiThreadingTransaction.java
index 6dd9486b682..bc1d5734c7f 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PublicApiThreadingTransaction.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PublicApiThreadingTransaction.java
@@ -25,7 +25,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.function.Supplier;
import org.apache.ignite.internal.hlc.HybridTimestamp;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.thread.PublicApiThreading;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.tx.PendingTxPartitionEnlistment;
@@ -85,7 +85,7 @@ public class PublicApiThreadingTransaction implements InternalTransaction, Wrapp
}
@Override
- public PendingTxPartitionEnlistment enlistedPartition(ReplicationGroupId replicationGroupId) {
+ public PendingTxPartitionEnlistment enlistedPartition(ZonePartitionId replicationGroupId) {
return transaction.enlistedPartition(replicationGroupId);
}
@@ -95,18 +95,18 @@ public class PublicApiThreadingTransaction implements InternalTransaction, Wrapp
}
@Override
- public boolean assignCommitPartition(ReplicationGroupId commitPartitionId) {
+ public boolean assignCommitPartition(ZonePartitionId commitPartitionId) {
return transaction.assignCommitPartition(commitPartitionId);
}
@Override
- public ReplicationGroupId commitPartition() {
+ public ZonePartitionId commitPartition() {
return transaction.commitPartition();
}
@Override
public void enlist(
- ReplicationGroupId replicationGroupId,
+ ZonePartitionId replicationGroupId,
int tableId,
String primaryNodeConsistentId,
long consistencyToken
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadOnlyImplicitTransactionImpl.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadOnlyImplicitTransactionImpl.java
index 0833cd3be94..c152eae6d40 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadOnlyImplicitTransactionImpl.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadOnlyImplicitTransactionImpl.java
@@ -23,8 +23,7 @@ import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.hlc.HybridTimestampTracker;
-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.tx.InternalTransaction;
import org.apache.ignite.internal.tx.PendingTxPartitionEnlistment;
import org.apache.ignite.internal.tx.TxState;
@@ -109,7 +108,7 @@ public class ReadOnlyImplicitTransactionImpl implements InternalTransaction {
@Override
public void enlist(
- ReplicationGroupId replicationGroupId,
+ ZonePartitionId replicationGroupId,
int tableId,
String primaryNodeConsistentId,
long consistencyToken
@@ -118,17 +117,17 @@ public class ReadOnlyImplicitTransactionImpl implements InternalTransaction {
}
@Override
- public PendingTxPartitionEnlistment enlistedPartition(ReplicationGroupId replicationGroupId) {
+ public PendingTxPartitionEnlistment enlistedPartition(ZonePartitionId replicationGroupId) {
return null;
}
@Override
- public boolean assignCommitPartition(ReplicationGroupId commitPartitionId) {
+ public boolean assignCommitPartition(ZonePartitionId commitPartitionId) {
return true;
}
@Override
- public TablePartitionId commitPartition() {
+ public ZonePartitionId commitPartition() {
return null;
}
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadOnlyTransactionImpl.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadOnlyTransactionImpl.java
index 5c85ffefbe8..77b639e5a3c 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadOnlyTransactionImpl.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadOnlyTransactionImpl.java
@@ -26,8 +26,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.hlc.HybridTimestampTracker;
-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.tx.PendingTxPartitionEnlistment;
/**
@@ -85,7 +84,7 @@ public class ReadOnlyTransactionImpl extends IgniteAbstractTransactionImpl {
@Override
public void enlist(
- ReplicationGroupId replicationGroupId,
+ ZonePartitionId replicationGroupId,
int tableId,
String primaryNodeConsistentId,
long consistencyToken
@@ -94,17 +93,17 @@ public class ReadOnlyTransactionImpl extends IgniteAbstractTransactionImpl {
}
@Override
- public PendingTxPartitionEnlistment enlistedPartition(ReplicationGroupId replicationGroupId) {
+ public PendingTxPartitionEnlistment enlistedPartition(ZonePartitionId replicationGroupId) {
return null;
}
@Override
- public boolean assignCommitPartition(ReplicationGroupId commitPartitionId) {
+ public boolean assignCommitPartition(ZonePartitionId commitPartitionId) {
return true;
}
@Override
- public TablePartitionId commitPartition() {
+ public ZonePartitionId commitPartition() {
return null;
}
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadWriteTransactionImpl.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadWriteTransactionImpl.java
index 1e7fe583582..6e58b3dd4f3 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadWriteTransactionImpl.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadWriteTransactionImpl.java
@@ -32,8 +32,6 @@ import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.hlc.HybridTimestampTracker;
-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.tx.PendingTxPartitionEnlistment;
import org.apache.ignite.internal.tx.TransactionIds;
@@ -45,17 +43,15 @@ import org.jetbrains.annotations.Nullable;
* The read-write implementation of an internal transaction.
*/
public class ReadWriteTransactionImpl extends IgniteAbstractTransactionImpl {
- private final boolean colocationEnabled;
-
/** Commit partition updater. */
- private static final AtomicReferenceFieldUpdater<ReadWriteTransactionImpl, ReplicationGroupId> COMMIT_PART_UPDATER =
- AtomicReferenceFieldUpdater.newUpdater(ReadWriteTransactionImpl.class, ReplicationGroupId.class, "commitPart");
+ private static final AtomicReferenceFieldUpdater<ReadWriteTransactionImpl, ZonePartitionId> COMMIT_PART_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(ReadWriteTransactionImpl.class, ZonePartitionId.class, "commitPart");
/** Enlisted partitions: partition id -> partition info. */
- private final Map<ReplicationGroupId, PendingTxPartitionEnlistment> enlisted = new ConcurrentHashMap<>();
+ private final Map<ZonePartitionId, PendingTxPartitionEnlistment> enlisted = new ConcurrentHashMap<>();
/** A partition which stores the transaction state. {@code null} before first enlistment. */
- private volatile @Nullable ReplicationGroupId commitPart;
+ private volatile @Nullable ZonePartitionId commitPart;
/** The lock protects the transaction topology from concurrent modification during finishing. */
private final ReentrantReadWriteLock enlistPartitionLock = new ReentrantReadWriteLock();
@@ -84,46 +80,37 @@ public class ReadWriteTransactionImpl extends IgniteAbstractTransactionImpl {
UUID id,
UUID txCoordinatorId,
boolean implicit,
- long timeout,
- boolean colocationEnabled
+ long timeout
) {
super(txManager, observableTsTracker, id, txCoordinatorId, implicit, timeout);
-
- this.colocationEnabled = colocationEnabled;
}
/** {@inheritDoc} */
@Override
- public boolean assignCommitPartition(ReplicationGroupId commitPartitionId) {
- assertReplicationGroupType(commitPartitionId);
-
+ public boolean assignCommitPartition(ZonePartitionId commitPartitionId) {
return COMMIT_PART_UPDATER.compareAndSet(this, null, commitPartitionId);
}
/** {@inheritDoc} */
@Override
- public ReplicationGroupId commitPartition() {
+ public ZonePartitionId commitPartition() {
return commitPart;
}
/** {@inheritDoc} */
@Override
- public PendingTxPartitionEnlistment enlistedPartition(ReplicationGroupId partGroupId) {
- assertReplicationGroupType(partGroupId);
-
+ public PendingTxPartitionEnlistment enlistedPartition(ZonePartitionId partGroupId) {
return enlisted.get(partGroupId);
}
/** {@inheritDoc} */
@Override
public void enlist(
- ReplicationGroupId replicationGroupId,
+ ZonePartitionId replicationGroupId,
int tableId,
String primaryNodeConsistentId,
long consistencyToken
) {
- assertReplicationGroupType(replicationGroupId);
-
// No need to wait for lock if commit is in progress.
if (!enlistPartitionLock.readLock().tryLock()) {
failEnlist();
@@ -144,11 +131,6 @@ public class ReadWriteTransactionImpl extends IgniteAbstractTransactionImpl {
}
}
- private void assertReplicationGroupType(ReplicationGroupId replicationGroupId) {
- assert (colocationEnabled ? replicationGroupId instanceof ZonePartitionId : replicationGroupId instanceof TablePartitionId)
- : "Invalid replication group type: " + replicationGroupId.getClass();
- }
-
/**
* Fails the operation.
*/
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/RemoteReadWriteTransaction.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/RemoteReadWriteTransaction.java
index c72e58b282f..a73e5bfffce 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/RemoteReadWriteTransaction.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/RemoteReadWriteTransaction.java
@@ -24,7 +24,6 @@ import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.network.InternalClusterNode;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.tostring.IgniteToStringExclude;
import org.apache.ignite.internal.tostring.S;
@@ -100,12 +99,12 @@ public abstract class RemoteReadWriteTransaction implements InternalTransaction
}
@Override
- public PendingTxPartitionEnlistment enlistedPartition(ReplicationGroupId replicationGroupId) {
+ public PendingTxPartitionEnlistment enlistedPartition(ZonePartitionId replicationGroupId) {
return enlistment;
}
@Override
- public boolean assignCommitPartition(ReplicationGroupId replicationGroupId) {
+ public boolean assignCommitPartition(ZonePartitionId replicationGroupId) {
return false;
}
@@ -115,7 +114,7 @@ public abstract class RemoteReadWriteTransaction implements InternalTransaction
}
@Override
- public void enlist(ReplicationGroupId replicationGroupId, int tableId, String primaryNodeConsistentId, long consistencyToken) {
+ public void enlist(ZonePartitionId replicationGroupId, int tableId, String primaryNodeConsistentId, long consistencyToken) {
// Validate primary replica.
if (!localNodeConsistentId.equals(primaryNodeConsistentId)) {
throw new TransactionException(REPLICA_MISS_ERR, format("The primary replica has changed [txId={}, "
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TransactionInflights.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TransactionInflights.java
index c6c79c1c89c..338383a45f6 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TransactionInflights.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TransactionInflights.java
@@ -36,7 +36,7 @@ import java.util.function.Function;
import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.placementdriver.PlacementDriver;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.tx.MismatchingTransactionOutcomeInternalException;
import org.apache.ignite.internal.tx.PendingTxPartitionEnlistment;
import org.apache.ignite.internal.tx.TransactionResult;
@@ -224,7 +224,7 @@ public class TransactionInflights {
txCtxMap.keySet().removeAll(txIds);
}
- void cancelWaitingInflights(ReplicationGroupId groupId) {
+ void cancelWaitingInflights(ZonePartitionId groupId) {
for (Map.Entry<UUID, TxContext> ctxEntry : txCtxMap.entrySet()) {
if (ctxEntry.getValue() instanceof ReadWriteTxContext) {
ReadWriteTxContext txContext = (ReadWriteTxContext) ctxEntry.getValue();
@@ -254,7 +254,7 @@ public class TransactionInflights {
});
}
- ReadWriteTxContext lockTxForNewUpdates(UUID txId, Map<ReplicationGroupId, PendingTxPartitionEnlistment> enlistedGroups) {
+ ReadWriteTxContext lockTxForNewUpdates(UUID txId, Map<ZonePartitionId, PendingTxPartitionEnlistment> enlistedGroups) {
return (ReadWriteTxContext) txCtxMap.compute(txId, (uuid, tuple0) -> {
if (tuple0 == null) {
tuple0 = new ReadWriteTxContext(placementDriver, clockService, true); // No writes enlisted, can go with unlock only.
@@ -291,7 +291,7 @@ public class TransactionInflights {
abstract void onInflightRemoved(@Nullable Throwable t);
- abstract void finishTx(@Nullable Map<ReplicationGroupId, PendingTxPartitionEnlistment> enlistedGroups);
+ abstract void finishTx(@Nullable Map<ZonePartitionId, PendingTxPartitionEnlistment> enlistedGroups);
abstract boolean isTxFinishing();
@@ -318,7 +318,7 @@ public class TransactionInflights {
}
@Override
- public void finishTx(@Nullable Map<ReplicationGroupId, PendingTxPartitionEnlistment> enlistedGroups) {
+ public void finishTx(@Nullable Map<ZonePartitionId, PendingTxPartitionEnlistment> enlistedGroups) {
markedFinished = true;
}
@@ -343,7 +343,7 @@ public class TransactionInflights {
private final PlacementDriver placementDriver;
private final boolean noWrites;
private volatile CompletableFuture<Void> finishInProgressFuture = null;
- private volatile Map<ReplicationGroupId, PendingTxPartitionEnlistment> enlistedGroups;
+ private volatile Map<ZonePartitionId, PendingTxPartitionEnlistment> enlistedGroups;
private final ClockService clockService;
private ReadWriteTxContext(PlacementDriver placementDriver, ClockService clockService) {
@@ -423,7 +423,7 @@ public class TransactionInflights {
int cntr = 0;
- for (Map.Entry<ReplicationGroupId, PendingTxPartitionEnlistment> e : enlistedGroups.entrySet()) {
+ for (Map.Entry<ZonePartitionId, PendingTxPartitionEnlistment> e : enlistedGroups.entrySet()) {
futures[cntr++] = placementDriver.getPrimaryReplica(e.getKey(), now)
.thenApply(replicaMeta -> {
long enlistmentConsistencyToken = e.getValue().consistencyToken();
@@ -455,7 +455,7 @@ public class TransactionInflights {
return waitRepFut;
}
- void cancelWaitingInflights(ReplicationGroupId groupId, long enlistmentConsistencyToken) {
+ void cancelWaitingInflights(ZonePartitionId groupId, long enlistmentConsistencyToken) {
waitRepFut.completeExceptionally(new PrimaryReplicaExpiredException(groupId, enlistmentConsistencyToken, null, null));
}
@@ -471,7 +471,7 @@ public class TransactionInflights {
}
@Override
- public void finishTx(Map<ReplicationGroupId, PendingTxPartitionEnlistment> enlistedGroups) {
+ public void finishTx(Map<ZonePartitionId, PendingTxPartitionEnlistment> enlistedGroups) {
this.enlistedGroups = enlistedGroups;
finishInProgressFuture = new CompletableFuture<>();
}
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxCleanupRequestHandler.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxCleanupRequestHandler.java
index a4a995afc62..93248505b12 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxCleanupRequestHandler.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxCleanupRequestHandler.java
@@ -18,7 +18,7 @@
package org.apache.ignite.internal.tx.impl;
import static java.util.concurrent.CompletableFuture.allOf;
-import static org.apache.ignite.internal.replicator.message.ReplicaMessageUtils.toReplicationGroupIdMessage;
+import static org.apache.ignite.internal.replicator.message.ReplicaMessageUtils.toZonePartitionIdMessage;
import static org.apache.ignite.internal.tx.impl.TxCleanupExceptionUtils.writeIntentSwitchFailureShouldBeLogged;
import java.util.ArrayList;
@@ -41,9 +41,9 @@ import org.apache.ignite.internal.network.ChannelType;
import org.apache.ignite.internal.network.InternalClusterNode;
import org.apache.ignite.internal.network.MessagingService;
import org.apache.ignite.internal.network.NetworkMessage;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.replicator.message.ReplicaMessagesFactory;
-import org.apache.ignite.internal.replicator.message.ReplicationGroupIdMessage;
+import org.apache.ignite.internal.replicator.message.ZonePartitionIdMessage;
import org.apache.ignite.internal.tx.LockManager;
import org.apache.ignite.internal.tx.message.CleanupReplicatedInfo;
import org.apache.ignite.internal.tx.message.CleanupReplicatedInfoMessage;
@@ -234,7 +234,7 @@ public class TxCleanupRequestHandler {
* @param groups Replication groups.
* @param sender Cleanup request sender, needed to send cleanup replicated response.
*/
- private void trackPartitions(UUID txId, Set<ReplicationGroupId> groups, InternalClusterNode sender) {
+ private void trackPartitions(UUID txId, Set<ZonePartitionId> groups, InternalClusterNode sender) {
writeIntentsReplicated.put(txId, new CleanupContext(sender, groups, groups));
}
@@ -258,7 +258,7 @@ public class TxCleanupRequestHandler {
*/
void writeIntentSwitchReplicated(WriteIntentSwitchReplicatedInfo info) {
CleanupContext cleanupContext = writeIntentsReplicated.computeIfPresent(info.txId(), (uuid, context) -> {
- Set<ReplicationGroupId> partitions = new HashSet<>(context.partitions);
+ Set<ZonePartitionId> partitions = new HashSet<>(context.partitions);
partitions.remove(info.partitionId());
return new CleanupContext(context.sender, partitions, context.initialPartitions);
@@ -279,18 +279,18 @@ public class TxCleanupRequestHandler {
* @param sender Cleanup request sender.
* @param partitions Partitions that we received replication confirmation for.
*/
- private void sendCleanupReplicatedResponse(UUID txId, InternalClusterNode sender, Collection<ReplicationGroupId> partitions) {
+ private void sendCleanupReplicatedResponse(UUID txId, InternalClusterNode sender, Collection<ZonePartitionId> partitions) {
messagingService.send(sender, ChannelType.DEFAULT, prepareResponse(new CleanupReplicatedInfo(txId, partitions)));
}
private static class CleanupContext {
private final InternalClusterNode sender;
- private final Set<ReplicationGroupId> partitions;
+ private final Set<ZonePartitionId> partitions;
- private final Set<ReplicationGroupId> initialPartitions;
+ private final Set<ZonePartitionId> initialPartitions;
- public CleanupContext(InternalClusterNode sender, Set<ReplicationGroupId> partitions, Set<ReplicationGroupId> initialPartitions) {
+ public CleanupContext(InternalClusterNode sender, Set<ZonePartitionId> partitions, Set<ZonePartitionId> initialPartitions) {
this.sender = sender;
this.partitions = partitions;
this.initialPartitions = initialPartitions;
@@ -298,11 +298,11 @@ public class TxCleanupRequestHandler {
}
private static CleanupReplicatedInfoMessage toCleanupReplicatedInfoMessage(CleanupReplicatedInfo info) {
- Collection<ReplicationGroupId> partitions = info.partitions();
- List<ReplicationGroupIdMessage> partitionMessages = new ArrayList<>(partitions.size());
+ Collection<ZonePartitionId> partitions = info.partitions();
+ List<ZonePartitionIdMessage> partitionMessages = new ArrayList<>(partitions.size());
- for (ReplicationGroupId partition : partitions) {
- partitionMessages.add(toReplicationGroupIdMessage(REPLICA_MESSAGES_FACTORY, partition));
+ for (ZonePartitionId partition : partitions) {
+ partitionMessages.add(toZonePartitionIdMessage(REPLICA_MESSAGES_FACTORY, partition));
}
return TX_MESSAGES_FACTORY.cleanupReplicatedInfoMessage()
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxCleanupRequestSender.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxCleanupRequestSender.java
index 0a65cd62fb7..2435112208f 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxCleanupRequestSender.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxCleanupRequestSender.java
@@ -39,8 +39,8 @@ import java.util.concurrent.ConcurrentMap;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.ReplicatorRecoverableExceptions;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.tx.PartitionEnlistment;
import org.apache.ignite.internal.tx.TransactionMeta;
import org.apache.ignite.internal.tx.TxState;
@@ -121,7 +121,7 @@ public class TxCleanupRequestSender {
}
}
- private void markTxnCleanupReplicated(UUID txId, TxState state, ReplicationGroupId commitPartitionId) {
+ private void markTxnCleanupReplicated(UUID txId, TxState state, ZonePartitionId commitPartitionId) {
long cleanupCompletionTimestamp = System.currentTimeMillis();
TxStateMeta txStateMeta = txStateVolatileStorage.state(txId);
@@ -159,7 +159,7 @@ public class TxCleanupRequestSender {
* @param txId Transaction id.
* @return Completable future of Void.
*/
- public CompletableFuture<Void> cleanup(ReplicationGroupId commitPartitionId, String node, UUID txId) {
+ public CompletableFuture<Void> cleanup(ZonePartitionId commitPartitionId, String node, UUID txId) {
return sendCleanupMessageWithRetries(commitPartitionId, false, null, txId, node, null);
}
@@ -174,8 +174,8 @@ public class TxCleanupRequestSender {
* @return Completable future of Void.
*/
public CompletableFuture<Void> cleanup(
- @Nullable ReplicationGroupId commitPartitionId,
- Map<ReplicationGroupId, PartitionEnlistment> enlistedPartitions,
+ @Nullable ZonePartitionId commitPartitionId,
+ Map<ZonePartitionId, PartitionEnlistment> enlistedPartitions,
boolean commit,
@Nullable HybridTimestamp commitTimestamp,
UUID txId
@@ -211,13 +211,13 @@ public class TxCleanupRequestSender {
* @return Completable future of Void.
*/
public CompletableFuture<Void> cleanup(
- @Nullable ReplicationGroupId commitPartitionId,
+ @Nullable ZonePartitionId commitPartitionId,
Collection<EnlistedPartitionGroup> partitions,
boolean commit,
@Nullable HybridTimestamp commitTimestamp,
UUID txId
) {
- Map<ReplicationGroupId, EnlistedPartitionGroup> partitionIds = partitions.stream()
+ Map<ZonePartitionId, EnlistedPartitionGroup> partitionIds = partitions.stream()
.collect(toMap(EnlistedPartitionGroup::groupId, identity()));
if (commitPartitionId != null) {
@@ -248,16 +248,16 @@ public class TxCleanupRequestSender {
}
private static Map<String, List<EnlistedPartitionGroup>> toPartitionInfosByPrimaryName(
- Map<String, Set<ReplicationGroupId>> partitionsByNode,
- Map<ReplicationGroupId, EnlistedPartitionGroup> partitionIds
+ Map<String, Set<ZonePartitionId>> partitionsByNode,
+ Map<ZonePartitionId, EnlistedPartitionGroup> partitionIds
) {
return partitionsByNode.entrySet().stream()
.collect(toMap(Entry::getKey, entry -> toPartitionInfos(entry.getValue(), partitionIds)));
}
private static List<EnlistedPartitionGroup> toPartitionInfos(
- Set<ReplicationGroupId> groupIds,
- Map<ReplicationGroupId, EnlistedPartitionGroup> partitionIds
+ Set<ZonePartitionId> groupIds,
+ Map<ZonePartitionId, EnlistedPartitionGroup> partitionIds
) {
return groupIds.stream()
.map(partitionIds::get)
@@ -265,13 +265,13 @@ public class TxCleanupRequestSender {
}
private void cleanupPartitionsWithoutPrimary(
- @Nullable ReplicationGroupId commitPartitionId,
+ @Nullable ZonePartitionId commitPartitionId,
boolean commit,
@Nullable HybridTimestamp commitTimestamp,
UUID txId,
List<EnlistedPartitionGroup> partitionsWithoutPrimary
) {
- Map<ReplicationGroupId, EnlistedPartitionGroup> partitionIds = partitionsWithoutPrimary.stream()
+ Map<ZonePartitionId, EnlistedPartitionGroup> partitionIds = partitionsWithoutPrimary.stream()
.collect(toMap(EnlistedPartitionGroup::groupId, identity()));
// For the partitions without primary, we need to wait until a new primary is found.
@@ -287,7 +287,7 @@ public class TxCleanupRequestSender {
}
private CompletableFuture<Void> cleanupPartitions(
- @Nullable ReplicationGroupId commitPartitionId,
+ @Nullable ZonePartitionId commitPartitionId,
Map<String, List<EnlistedPartitionGroup>> partitionsByNode,
boolean commit,
@Nullable HybridTimestamp commitTimestamp,
@@ -307,7 +307,7 @@ public class TxCleanupRequestSender {
}
private CompletableFuture<Void> sendCleanupMessageWithRetries(
- @Nullable ReplicationGroupId commitPartitionId,
+ @Nullable ZonePartitionId commitPartitionId,
boolean commit,
@Nullable HybridTimestamp commitTimestamp,
UUID txId,
@@ -357,19 +357,19 @@ public class TxCleanupRequestSender {
}
private static class CleanupContext {
- private final ReplicationGroupId commitPartitionId;
+ private final ZonePartitionId commitPartitionId;
/**
* The partitions the we have not received write intent replication confirmation for.
*/
- private final Set<ReplicationGroupId> partitions;
+ private final Set<ZonePartitionId> partitions;
/**
* The state of the transaction.
*/
private final TxState txState;
- private CleanupContext(ReplicationGroupId commitPartitionId, Set<ReplicationGroupId> partitions, TxState txState) {
+ private CleanupContext(ZonePartitionId commitPartitionId, Set<ZonePartitionId> partitions, TxState txState) {
this.commitPartitionId = commitPartitionId;
this.partitions = partitions;
this.txState = txState;
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java
index bd419a01a34..6b97fa4c3e4 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java
@@ -60,8 +60,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.LongSupplier;
-import org.apache.ignite.internal.components.NodeProperties;
-import org.apache.ignite.internal.components.SystemPropertiesNodeProperties;
import org.apache.ignite.internal.configuration.SystemDistributedConfiguration;
import org.apache.ignite.internal.configuration.SystemPropertyView;
import org.apache.ignite.internal.event.EventListener;
@@ -88,9 +86,7 @@ import org.apache.ignite.internal.placementdriver.PlacementDriver;
import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEvent;
import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEventParameters;
import org.apache.ignite.internal.replicator.ReplicaService;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.ReplicatorRecoverableExceptions;
-import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.replicator.message.ErrorReplicaResponse;
import org.apache.ignite.internal.replicator.message.ReplicaMessageGroup;
@@ -225,8 +221,6 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
private final FailureProcessor failureProcessor;
- private final NodeProperties nodeProperties;
-
private final TransactionsViewProvider txViewProvider = new TransactionsViewProvider();
private volatile PersistentTxStateVacuumizer persistentTxStateVacuumizer;
@@ -300,7 +294,6 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
lowWatermark,
commonScheduler,
new FailureManager(new NoOpFailureHandler()),
- new SystemPropertiesNodeProperties(),
metricManager
);
}
@@ -344,7 +337,6 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
LowWatermark lowWatermark,
ScheduledExecutorService commonScheduler,
FailureProcessor failureProcessor,
- NodeProperties nodeProperties,
MetricManager metricManager
) {
this.txConfig = txConfig;
@@ -365,7 +357,6 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
this.replicaService = replicaService;
this.commonScheduler = commonScheduler;
this.failureProcessor = failureProcessor;
- this.nodeProperties = nodeProperties;
this.metricsManager = metricManager;
placementDriverHelper = new PlacementDriverHelper(placementDriver, clockService);
@@ -406,16 +397,12 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
private CompletableFuture<Boolean> primaryReplicaEventListener(
PrimaryReplicaEventParameters eventParameters,
- Consumer<ReplicationGroupId> action
+ Consumer<ZonePartitionId> action
) {
- assertReplicationGroupType(eventParameters.groupId());
-
- // TODO: https://issues.apache.org/jira/browse/IGNITE-22522 - remove check for TablePartitionId.
- if (!(eventParameters.groupId() instanceof TablePartitionId) && !(eventParameters.groupId() instanceof ZonePartitionId)) {
- return falseCompletedFuture();
- }
+ assert eventParameters.groupId() instanceof ZonePartitionId :
+ "Invalid replication group type: " + eventParameters.groupId().getClass();
- action.accept(eventParameters.groupId());
+ action.accept((ZonePartitionId) eventParameters.groupId());
return falseCompletedFuture();
}
@@ -471,8 +458,8 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
HybridTimestampTracker timestampTracker,
HybridTimestamp beginTimestamp,
boolean implicit,
- InternalTxOptions options) {
-
+ InternalTxOptions options
+ ) {
UUID txId = transactionIdGenerator.transactionIdFor(beginTimestamp, options.priority());
long timeout = getTimeoutOrDefault(options, txConfig.readWriteTimeoutMillis().value());
@@ -483,8 +470,7 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
txId,
localNodeId,
implicit,
- timeout,
- nodeProperties.colocationEnabled()
+ timeout
);
// Implicit transactions are finished as soon as their operation/query is finished, they cannot be abandoned, so there is
@@ -620,22 +606,15 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
@Override
public CompletableFuture<Void> finish(
HybridTimestampTracker observableTimestampTracker,
- @Nullable ReplicationGroupId commitPartition,
+ @Nullable ZonePartitionId commitPartition,
boolean commitIntent,
boolean timeout,
boolean recovery,
- Map<ReplicationGroupId, PendingTxPartitionEnlistment> enlistedGroups,
+ Map<ZonePartitionId, PendingTxPartitionEnlistment> enlistedGroups,
UUID txId
) {
LOG.debug("Finish [commit={}, txId={}, groups={}, commitPartId={}].", commitIntent, txId, enlistedGroups, commitPartition);
- if (commitPartition != null) {
- assertReplicationGroupType(commitPartition);
- }
- for (ReplicationGroupId replicationGroupId : enlistedGroups.keySet()) {
- assertReplicationGroupType(replicationGroupId);
- }
-
assert enlistedGroups != null;
if (enlistedGroups.isEmpty()) {
@@ -708,12 +687,6 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
});
}
- private void assertReplicationGroupType(ReplicationGroupId replicationGroupId) {
- assert (nodeProperties.colocationEnabled() ? replicationGroupId instanceof ZonePartitionId
- : replicationGroupId instanceof TablePartitionId)
- : "Invalid replication group type: " + replicationGroupId.getClass();
- }
-
private static CompletableFuture<Void> checkTxOutcome(boolean commit, UUID txId, TransactionMeta stateMeta) {
if ((stateMeta.txState() == COMMITTED) == commit) {
return nullCompletedFuture();
@@ -727,9 +700,9 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
private CompletableFuture<Void> prepareFinish(
HybridTimestampTracker observableTimestampTracker,
- @Nullable ReplicationGroupId commitPartition,
+ @Nullable ZonePartitionId commitPartition,
boolean commit,
- Map<ReplicationGroupId, PendingTxPartitionEnlistment> enlistedGroups,
+ Map<ZonePartitionId, PendingTxPartitionEnlistment> enlistedGroups,
UUID txId,
CompletableFuture<TransactionMeta> txFinishFuture,
boolean unlock
@@ -745,7 +718,7 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
(unused, throwable) -> {
boolean verifiedCommit = throwable == null && commit;
- Map<ReplicationGroupId, PartitionEnlistment> groups = enlistedGroups.entrySet().stream()
+ Map<ZonePartitionId, PartitionEnlistment> groups = enlistedGroups.entrySet().stream()
.collect(toMap(Entry::getKey, Entry::getValue));
if (unlock) {
@@ -798,9 +771,9 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
*/
private CompletableFuture<Void> durableFinish(
HybridTimestampTracker observableTimestampTracker,
- ReplicationGroupId commitPartition,
+ ZonePartitionId commitPartition,
boolean commit,
- Map<ReplicationGroupId, PartitionEnlistment> enlistedPartitions,
+ Map<ZonePartitionId, PartitionEnlistment> enlistedPartitions,
UUID txId,
HybridTimestamp commitTimestamp,
CompletableFuture<TransactionMeta> txFinishFuture
@@ -865,11 +838,11 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
private CompletableFuture<Void> sendFinishRequest(
HybridTimestampTracker observableTimestampTracker,
- ReplicationGroupId commitPartition,
+ ZonePartitionId commitPartition,
String primaryConsistentId,
Long enlistmentConsistencyToken,
boolean commit,
- Map<ReplicationGroupId, PartitionEnlistment> enlistedPartitions,
+ Map<ZonePartitionId, PartitionEnlistment> enlistedPartitions,
UUID txId,
HybridTimestamp commitTimestamp,
CompletableFuture<TransactionMeta> txFinishFuture
@@ -1107,22 +1080,18 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
@Override
public CompletableFuture<Void> cleanup(
- @Nullable ReplicationGroupId commitPartitionId,
- Map<ReplicationGroupId, PartitionEnlistment> enlistedPartitions,
+ @Nullable ZonePartitionId commitPartitionId,
+ Map<ZonePartitionId, PartitionEnlistment> enlistedPartitions,
boolean commit,
@Nullable HybridTimestamp commitTimestamp,
UUID txId
) {
- for (ReplicationGroupId replicationGroupId : enlistedPartitions.keySet()) {
- assertReplicationGroupType(replicationGroupId);
- }
-
return txCleanupRequestSender.cleanup(commitPartitionId, enlistedPartitions, commit, commitTimestamp, txId);
}
@Override
public CompletableFuture<Void> cleanup(
- ReplicationGroupId commitPartitionId,
+ ZonePartitionId commitPartitionId,
Collection<EnlistedPartitionGroup> enlistedPartitions,
boolean commit,
@Nullable HybridTimestamp commitTimestamp,
@@ -1132,9 +1101,7 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
}
@Override
- public CompletableFuture<Void> cleanup(ReplicationGroupId commitPartitionId, String node, UUID txId) {
- assertReplicationGroupType(commitPartitionId);
-
+ public CompletableFuture<Void> cleanup(ZonePartitionId commitPartitionId, String node, UUID txId) {
return txCleanupRequestSender.cleanup(commitPartitionId, node, txId);
}
@@ -1235,14 +1202,14 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
* @return Verification future.
*/
private CompletableFuture<Void> verifyCommitTimestamp(
- Map<ReplicationGroupId, PendingTxPartitionEnlistment> enlistedGroups,
+ Map<ZonePartitionId, PendingTxPartitionEnlistment> enlistedGroups,
HybridTimestamp commitTimestamp
) {
var verificationFutures = new CompletableFuture[enlistedGroups.size()];
int cnt = -1;
- for (Map.Entry<ReplicationGroupId, PendingTxPartitionEnlistment> enlistedGroup : enlistedGroups.entrySet()) {
- ReplicationGroupId groupId = enlistedGroup.getKey();
+ for (Map.Entry<ZonePartitionId, PendingTxPartitionEnlistment> enlistedGroup : enlistedGroups.entrySet()) {
+ ZonePartitionId groupId = enlistedGroup.getKey();
long expectedEnlistmentConsistencyToken = enlistedGroup.getValue().consistencyToken();
verificationFutures[++cnt] = placementDriver.getPrimaryReplica(groupId, commitTimestamp)
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxMessageSender.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxMessageSender.java
index 9d1c3945dcf..203b9dc8490 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxMessageSender.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxMessageSender.java
@@ -17,7 +17,7 @@
package org.apache.ignite.internal.tx.impl;
-import static org.apache.ignite.internal.replicator.message.ReplicaMessageUtils.toReplicationGroupIdMessage;
+import static org.apache.ignite.internal.replicator.message.ReplicaMessageUtils.toZonePartitionIdMessage;
import java.util.ArrayList;
import java.util.Collection;
@@ -32,10 +32,10 @@ import org.apache.ignite.internal.network.InternalClusterNode;
import org.apache.ignite.internal.network.MessagingService;
import org.apache.ignite.internal.network.NetworkMessage;
import org.apache.ignite.internal.replicator.ReplicaService;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.replicator.message.ReplicaMessagesFactory;
import org.apache.ignite.internal.replicator.message.ReplicaResponse;
-import org.apache.ignite.internal.replicator.message.ReplicationGroupIdMessage;
+import org.apache.ignite.internal.replicator.message.ZonePartitionIdMessage;
import org.apache.ignite.internal.tx.PartitionEnlistment;
import org.apache.ignite.internal.tx.TransactionMeta;
import org.apache.ignite.internal.tx.TransactionResult;
@@ -103,7 +103,7 @@ public class TxMessageSender {
return replicaService.invoke(
primaryConsistentId,
TX_MESSAGES_FACTORY.writeIntentSwitchReplicaRequest()
- .groupId(toReplicationGroupIdMessage(REPLICA_MESSAGES_FACTORY, partition.groupId()))
+ .groupId(toZonePartitionIdMessage(REPLICA_MESSAGES_FACTORY, partition.groupId()))
.tableIds(partition.tableIds())
.timestamp(clockService.now())
.txId(txId)
@@ -156,14 +156,14 @@ public class TxMessageSender {
*/
public CompletableFuture<TransactionResult> finish(
String primaryConsistentId,
- ReplicationGroupId commitPartition,
- Map<ReplicationGroupId, PartitionEnlistment> enlistedPartitions,
+ ZonePartitionId commitPartition,
+ Map<ZonePartitionId, PartitionEnlistment> enlistedPartitions,
UUID txId,
Long consistencyToken,
boolean commit,
@Nullable HybridTimestamp commitTimestamp
) {
- ReplicationGroupIdMessage commitPartitionIdMessage = toReplicationGroupIdMessage(REPLICA_MESSAGES_FACTORY, commitPartition);
+ ZonePartitionIdMessage commitPartitionIdMessage = toZonePartitionIdMessage(REPLICA_MESSAGES_FACTORY, commitPartition);
return replicaService.invoke(
primaryConsistentId,
@@ -171,7 +171,7 @@ public class TxMessageSender {
.txId(txId)
.commitPartitionId(commitPartitionIdMessage)
.timestamp(clockService.now())
- .groupId(toReplicationGroupIdMessage(REPLICA_MESSAGES_FACTORY, commitPartition))
+ .groupId(toZonePartitionIdMessage(REPLICA_MESSAGES_FACTORY, commitPartition))
.groups(toEnlistedPartitionMessagesByGroupId(enlistedPartitions))
.commit(commit)
.commitTimestamp(commitTimestamp)
@@ -192,13 +192,13 @@ public class TxMessageSender {
public CompletableFuture<TransactionMeta> resolveTxStateFromCommitPartition(
String primaryConsistentId,
UUID txId,
- ReplicationGroupId commitGrpId,
+ ZonePartitionId commitGrpId,
Long consistencyToken
) {
return replicaService.invoke(
primaryConsistentId,
TX_MESSAGES_FACTORY.txStateCommitPartitionRequest()
- .groupId(toReplicationGroupIdMessage(REPLICA_MESSAGES_FACTORY, commitGrpId))
+ .groupId(toZonePartitionIdMessage(REPLICA_MESSAGES_FACTORY, commitGrpId))
.txId(txId)
.enlistmentConsistencyToken(consistencyToken)
.build()
@@ -239,11 +239,11 @@ public class TxMessageSender {
* @param replicationGroupId Replication group ID corresponding to a partition.
* @return Completable future of ReplicaResponse.
*/
- public CompletableFuture<ReplicaResponse> sendRecoveryCleanup(String primaryConsistentId, ReplicationGroupId replicationGroupId) {
+ public CompletableFuture<ReplicaResponse> sendRecoveryCleanup(String primaryConsistentId, ZonePartitionId replicationGroupId) {
return replicaService.invoke(
primaryConsistentId,
TX_MESSAGES_FACTORY.txCleanupRecoveryRequest()
- .groupId(toReplicationGroupIdMessage(REPLICA_MESSAGES_FACTORY, replicationGroupId))
+ .groupId(toZonePartitionIdMessage(REPLICA_MESSAGES_FACTORY, replicationGroupId))
.build()
);
}
@@ -264,7 +264,7 @@ public class TxMessageSender {
for (EnlistedPartitionGroup partition : enlistedPartitionGroups) {
messages.add(
TX_MESSAGES_FACTORY.enlistedPartitionGroupMessage()
- .groupId(toReplicationGroupIdMessage(REPLICA_MESSAGES_FACTORY, partition.groupId()))
+ .groupId(toZonePartitionIdMessage(REPLICA_MESSAGES_FACTORY, partition.groupId()))
.tableIds(partition.tableIds())
.build()
);
@@ -273,16 +273,16 @@ public class TxMessageSender {
return messages;
}
- private static Map<ReplicationGroupIdMessage, PartitionEnlistmentMessage> toEnlistedPartitionMessagesByGroupId(
- Map<ReplicationGroupId, PartitionEnlistment> idEnlistedPartitions
+ private static Map<ZonePartitionIdMessage, PartitionEnlistmentMessage> toEnlistedPartitionMessagesByGroupId(
+ Map<ZonePartitionId, PartitionEnlistment> idEnlistedPartitions
) {
- var messages = new HashMap<ReplicationGroupIdMessage, PartitionEnlistmentMessage>(idEnlistedPartitions.size());
+ var messages = new HashMap<ZonePartitionIdMessage, PartitionEnlistmentMessage>(idEnlistedPartitions.size());
- for (Map.Entry<ReplicationGroupId, PartitionEnlistment> e : idEnlistedPartitions.entrySet()) {
+ for (Map.Entry<ZonePartitionId, PartitionEnlistment> e : idEnlistedPartitions.entrySet()) {
PartitionEnlistment enlistedPartition = e.getValue();
messages.put(
- toReplicationGroupIdMessage(REPLICA_MESSAGES_FACTORY, e.getKey()),
+ toZonePartitionIdMessage(REPLICA_MESSAGES_FACTORY, e.getKey()),
TX_MESSAGES_FACTORY.partitionEnlistmentMessage()
.primaryConsistentId(enlistedPartition.primaryNodeConsistentId())
.tableIds(enlistedPartition.tableIds())
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/VolatileTxStateMetaStorage.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/VolatileTxStateMetaStorage.java
index ae78f93ced9..80893844248 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/VolatileTxStateMetaStorage.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/VolatileTxStateMetaStorage.java
@@ -32,7 +32,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.tx.TxState;
import org.apache.ignite.internal.tx.TxStateMeta;
@@ -148,14 +148,10 @@ public class VolatileTxStateMetaStorage {
public CompletableFuture<Void> vacuum(
long vacuumObservationTimestamp,
long txnResourceTtl,
- // TODO https://issues.apache.org/jira/browse/IGNITE-22522
- // Should be changed to ZonePartitionId.
- Function<Map<ReplicationGroupId, Set<VacuumizableTx>>, CompletableFuture<PersistentTxStateVacuumResult>> persistentVacuumOp,
+ Function<Map<ZonePartitionId, Set<VacuumizableTx>>, CompletableFuture<PersistentTxStateVacuumResult>> persistentVacuumOp,
ResourceVacuumMetrics resourceVacuumMetrics
) {
- // TODO https://issues.apache.org/jira/browse/IGNITE-22522
- // Should be changed to ZonePartitionId.
- Map<ReplicationGroupId, Set<VacuumizableTx>> txIds = new HashMap<>();
+ Map<ZonePartitionId, Set<VacuumizableTx>> txIds = new HashMap<>();
Map<UUID, Long> cleanupCompletionTimestamps = new HashMap<>();
var skippedForFurtherProcessingUnfinishedTxnsCount = new AtomicInteger();
@@ -182,8 +178,6 @@ public class VolatileTxStateMetaStorage {
return null;
} else {
- // TODO https://issues.apache.org/jira/browse/IGNITE-22522
- // Should be changed to ZonePartitionId.
Set<VacuumizableTx> ids = txIds.computeIfAbsent(meta0.commitPartitionId(), k -> new HashSet<>());
ids.add(new VacuumizableTx(txId, cleanupCompletionTimestamp));
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/CleanupReplicatedInfo.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/CleanupReplicatedInfo.java
index 5cd2e6c0b23..1eca2a6b25d 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/CleanupReplicatedInfo.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/CleanupReplicatedInfo.java
@@ -17,23 +17,19 @@
package org.apache.ignite.internal.tx.message;
-import java.io.Serializable;
import java.util.Collection;
import java.util.UUID;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
/**
* The result of a replicated cleanup request.
*/
-public class CleanupReplicatedInfo implements Serializable {
-
- private static final long serialVersionUID = -975001033274630774L;
-
+public class CleanupReplicatedInfo {
private final UUID txId;
- private final Collection<ReplicationGroupId> partitions;
+ private final Collection<ZonePartitionId> partitions;
- public CleanupReplicatedInfo(UUID txId, Collection<ReplicationGroupId> partitions) {
+ public CleanupReplicatedInfo(UUID txId, Collection<ZonePartitionId> partitions) {
this.txId = txId;
this.partitions = partitions;
}
@@ -42,7 +38,7 @@ public class CleanupReplicatedInfo implements Serializable {
return txId;
}
- public Collection<ReplicationGroupId> partitions() {
+ public Collection<ZonePartitionId> partitions() {
return partitions;
}
}
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/CleanupReplicatedInfoMessage.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/CleanupReplicatedInfoMessage.java
index 76025e233c1..c7e4d213735 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/CleanupReplicatedInfoMessage.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/CleanupReplicatedInfoMessage.java
@@ -22,8 +22,8 @@ import java.util.List;
import java.util.UUID;
import org.apache.ignite.internal.network.NetworkMessage;
import org.apache.ignite.internal.network.annotations.Transferable;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
-import org.apache.ignite.internal.replicator.message.ReplicationGroupIdMessage;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.replicator.message.ZonePartitionIdMessage;
/** Message for transferring a {@link CleanupReplicatedInfo}. */
@Transferable(TxMessageGroup.CLEANUP_REPLICATED_INFO_MESSAGE)
@@ -32,14 +32,14 @@ public interface CleanupReplicatedInfoMessage extends NetworkMessage {
UUID txId();
/** Partitions. */
- List<ReplicationGroupIdMessage> partitions();
+ List<ZonePartitionIdMessage> partitions();
/** Converts to {@link CleanupReplicatedInfo}. */
default CleanupReplicatedInfo asCleanupReplicatedInfo() {
- List<ReplicationGroupIdMessage> partitionMessages = partitions();
- List<ReplicationGroupId> partitions = new ArrayList<>(partitionMessages.size());
+ List<ZonePartitionIdMessage> partitionMessages = partitions();
+ List<ZonePartitionId> partitions = new ArrayList<>(partitionMessages.size());
- for (ReplicationGroupIdMessage partitionMessage : partitionMessages) {
+ for (ZonePartitionIdMessage partitionMessage : partitionMessages) {
partitions.add(partitionMessage.asReplicationGroupId());
}
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/EnlistedPartitionGroupMessage.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/EnlistedPartitionGroupMessage.java
index 929c2535822..ff9e11e51b8 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/EnlistedPartitionGroupMessage.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/EnlistedPartitionGroupMessage.java
@@ -20,7 +20,7 @@ package org.apache.ignite.internal.tx.message;
import java.util.Set;
import org.apache.ignite.internal.network.NetworkMessage;
import org.apache.ignite.internal.network.annotations.Transferable;
-import org.apache.ignite.internal.replicator.message.ReplicationGroupIdMessage;
+import org.apache.ignite.internal.replicator.message.ZonePartitionIdMessage;
import org.apache.ignite.internal.tx.impl.EnlistedPartitionGroup;
/**
@@ -31,7 +31,7 @@ public interface EnlistedPartitionGroupMessage extends NetworkMessage {
/**
* Replication group ID of the partition.
*/
- ReplicationGroupIdMessage groupId();
+ ZonePartitionIdMessage groupId();
/**
* IDs of tables for which the partition is enlisted.
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TableWriteIntentSwitchReplicaRequest.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TableWriteIntentSwitchReplicaRequest.java
index 77c919b798c..12145799fe1 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TableWriteIntentSwitchReplicaRequest.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TableWriteIntentSwitchReplicaRequest.java
@@ -25,8 +25,7 @@ import org.apache.ignite.internal.replicator.message.TableAware;
* or removes them (TxState.ABORTED).
*
* <p>Never sent by users of ReplicaService; it's rather an internal request used to handle {@link WriteIntentSwitchReplicaRequest}s
- * in per-zone case (invoked on table replicas directly). Never used in per-table case.
- * // TODO: IGNITE-22522 remove the above mention of per-table case.
+ * in per-zone case (invoked on table replicas directly).
*/
@Transferable(TxMessageGroup.TABLE_WRITE_INTENT_SWITCH_REQUEST)
public interface TableWriteIntentSwitchReplicaRequest extends WriteIntentSwitchReplicaRequestBase, TableAware {
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxFinishReplicaRequest.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxFinishReplicaRequest.java
index 2dbcd92cdbd..6da908a7ef8 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxFinishReplicaRequest.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxFinishReplicaRequest.java
@@ -25,8 +25,8 @@ import java.util.UUID;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.network.annotations.Transferable;
import org.apache.ignite.internal.replicator.message.PrimaryReplicaRequest;
-import org.apache.ignite.internal.replicator.message.ReplicationGroupIdMessage;
import org.apache.ignite.internal.replicator.message.TimestampAware;
+import org.apache.ignite.internal.replicator.message.ZonePartitionIdMessage;
import org.jetbrains.annotations.Nullable;
/**
@@ -52,7 +52,7 @@ public interface TxFinishReplicaRequest extends PrimaryReplicaRequest, Timestamp
*
* @return Commit partition id.
*/
- ReplicationGroupIdMessage commitPartitionId();
+ ZonePartitionIdMessage commitPartitionId();
/**
* Returns {@code True} if a commit request.
@@ -65,7 +65,7 @@ public interface TxFinishReplicaRequest extends PrimaryReplicaRequest, Timestamp
@Nullable HybridTimestamp commitTimestamp();
/** Enlisted partition groups aggregated by expected primary replica nodes. */
- Map<ReplicationGroupIdMessage, PartitionEnlistmentMessage> groups();
+ Map<ZonePartitionIdMessage, PartitionEnlistmentMessage> groups();
/** IDs of tables enlisted in the transaction. */
default Set<Integer> tableIds() {
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxStateMetaAbandonedMessage.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxStateMetaAbandonedMessage.java
index 04db81f3c96..6df66a7d6c1 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxStateMetaAbandonedMessage.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxStateMetaAbandonedMessage.java
@@ -18,7 +18,7 @@
package org.apache.ignite.internal.tx.message;
import org.apache.ignite.internal.network.annotations.Transferable;
-import org.apache.ignite.internal.replicator.message.ReplicationGroupIdMessage;
+import org.apache.ignite.internal.replicator.message.ZonePartitionIdMessage;
import org.apache.ignite.internal.tx.TransactionMeta;
import org.apache.ignite.internal.tx.TxStateMetaAbandoned;
@@ -30,7 +30,7 @@ public interface TxStateMetaAbandonedMessage extends TxStateMetaMessage {
/** Converts to {@link TxStateMetaAbandoned}. */
default TxStateMetaAbandoned asTxStateMetaAbandoned() {
- ReplicationGroupIdMessage commitPartitionId = commitPartitionId();
+ ZonePartitionIdMessage commitPartitionId = commitPartitionId();
return new TxStateMetaAbandoned(
txCoordinatorId(),
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxStateMetaFinishingMessage.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxStateMetaFinishingMessage.java
index f50e931c2a0..687889ed5d2 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxStateMetaFinishingMessage.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxStateMetaFinishingMessage.java
@@ -18,7 +18,7 @@
package org.apache.ignite.internal.tx.message;
import org.apache.ignite.internal.network.annotations.Transferable;
-import org.apache.ignite.internal.replicator.message.ReplicationGroupIdMessage;
+import org.apache.ignite.internal.replicator.message.ZonePartitionIdMessage;
import org.apache.ignite.internal.tx.TransactionMeta;
import org.apache.ignite.internal.tx.TxStateMetaFinishing;
@@ -31,7 +31,7 @@ import org.apache.ignite.internal.tx.TxStateMetaFinishing;
public interface TxStateMetaFinishingMessage extends TxStateMetaMessage {
/** Converts to {@link TxStateMetaFinishing}. */
default TxStateMetaFinishing asTxStateMetaFinishing() {
- ReplicationGroupIdMessage commitPartitionId = commitPartitionId();
+ ZonePartitionIdMessage commitPartitionId = commitPartitionId();
return new TxStateMetaFinishing(
txCoordinatorId(),
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxStateMetaMessage.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxStateMetaMessage.java
index 80334d1d8fb..42837d0b41d 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxStateMetaMessage.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxStateMetaMessage.java
@@ -19,7 +19,7 @@ package org.apache.ignite.internal.tx.message;
import java.util.UUID;
import org.apache.ignite.internal.network.annotations.Transferable;
-import org.apache.ignite.internal.replicator.message.ReplicationGroupIdMessage;
+import org.apache.ignite.internal.replicator.message.ZonePartitionIdMessage;
import org.apache.ignite.internal.tx.TransactionMeta;
import org.apache.ignite.internal.tx.TxStateMeta;
import org.jetbrains.annotations.Nullable;
@@ -31,7 +31,7 @@ public interface TxStateMetaMessage extends TransactionMetaMessage {
@Nullable UUID txCoordinatorId();
/** ID of the replication group that manages a transaction state. */
- @Nullable ReplicationGroupIdMessage commitPartitionId();
+ @Nullable ZonePartitionIdMessage commitPartitionId();
/** Initial vacuum observation timestamp. */
@Nullable Long initialVacuumObservationTimestamp();
@@ -45,7 +45,7 @@ public interface TxStateMetaMessage extends TransactionMetaMessage {
/** Converts to {@link TxStateMeta}. */
default TxStateMeta asTxStateMeta() {
- ReplicationGroupIdMessage commitPartitionId = commitPartitionId();
+ ZonePartitionIdMessage commitPartitionId = commitPartitionId();
return new TxStateMeta(
txState(),
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/WriteIntentSwitchReplicatedInfo.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/WriteIntentSwitchReplicatedInfo.java
index 476cdf47fef..b8efd0f09ba 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/WriteIntentSwitchReplicatedInfo.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/WriteIntentSwitchReplicatedInfo.java
@@ -17,23 +17,19 @@
package org.apache.ignite.internal.tx.message;
-import java.io.Serializable;
import java.util.UUID;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.tostring.S;
/**
* The result of a replicated write intent switch request.
*/
-public class WriteIntentSwitchReplicatedInfo implements Serializable {
-
- private static final long serialVersionUID = 8130171618503063413L;
-
+public class WriteIntentSwitchReplicatedInfo {
private final UUID txId;
- private final ReplicationGroupId partitionId;
+ private final ZonePartitionId partitionId;
- public WriteIntentSwitchReplicatedInfo(UUID txId, ReplicationGroupId partitionId) {
+ public WriteIntentSwitchReplicatedInfo(UUID txId, ZonePartitionId partitionId) {
this.txId = txId;
this.partitionId = partitionId;
}
@@ -42,7 +38,7 @@ public class WriteIntentSwitchReplicatedInfo implements Serializable {
return txId;
}
- public ReplicationGroupId partitionId() {
+ public ZonePartitionId partitionId() {
return partitionId;
}
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateMetaRocksDbPartitionStorage.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateMetaRocksDbPartitionStorage.java
index 13614760e93..d94c4e7403c 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateMetaRocksDbPartitionStorage.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateMetaRocksDbPartitionStorage.java
@@ -19,7 +19,7 @@ package org.apache.ignite.internal.tx.storage.state.rocksdb;
import static org.apache.ignite.internal.rocksdb.RocksUtils.incrementPrefix;
import static org.apache.ignite.internal.tx.storage.state.rocksdb.TxStateRocksDbSharedStorage.BYTE_ORDER;
-import static org.apache.ignite.internal.tx.storage.state.rocksdb.TxStateRocksDbStorage.TABLE_OR_ZONE_ID_SIZE_BYTES;
+import static org.apache.ignite.internal.tx.storage.state.rocksdb.TxStateRocksDbStorage.ZONE_ID_SIZE_BYTES;
import java.nio.ByteBuffer;
import java.util.List;
@@ -36,11 +36,11 @@ import org.rocksdb.WriteBatch;
* A wrapper around a RocksDB column family to store TX storage meta information.
*/
class TxStateMetaRocksDbPartitionStorage {
- /** Prefix of the meta keys including table/zone ID. Consists of a 1-byte prefix, and tableId/zoneId (4 bytes), in Big Endian. */
- static final int TABLE_OR_ZONE_PREFIX_SIZE_BYTES = 1 + TABLE_OR_ZONE_ID_SIZE_BYTES;
+ /** Prefix of the meta keys including zone ID. Consists of a 1-byte prefix, and zoneId (4 bytes), in Big Endian. */
+ static final int ZONE_PREFIX_SIZE_BYTES = 1 + ZONE_ID_SIZE_BYTES;
- /** Key length for the payload. Consists of a 1-byte prefix, tableId/zoneId (4 bytes) and partitionId (2 bytes), in Big Endian. */
- private static final int KEY_SIZE_BYTES = TABLE_OR_ZONE_PREFIX_SIZE_BYTES + Short.BYTES;
+ /** Key length for the payload. Consists of a 1-byte prefix, zoneId (4 bytes) and partitionId (2 bytes), in Big Endian. */
+ private static final int KEY_SIZE_BYTES = ZONE_PREFIX_SIZE_BYTES + Short.BYTES;
/**
* Prefix to store meta information, such as last applied index and term.
@@ -64,7 +64,7 @@ class TxStateMetaRocksDbPartitionStorage {
private final ColumnFamily columnFamily;
- private final int tableOrZoneId;
+ private final int zoneId;
private final int partitionId;
@@ -85,10 +85,10 @@ class TxStateMetaRocksDbPartitionStorage {
@Nullable
private volatile LeaseInfo leaseInfo;
- TxStateMetaRocksDbPartitionStorage(ColumnFamily columnFamily, int tableOrZoneId, int partitionId) {
+ TxStateMetaRocksDbPartitionStorage(ColumnFamily columnFamily, int zoneId, int partitionId) {
this.columnFamily = columnFamily;
this.partitionId = partitionId;
- this.tableOrZoneId = tableOrZoneId;
+ this.zoneId = zoneId;
lastAppliedKey = createKey(LAST_APPLIED_PREFIX);
confKey = createKey(CONF_PREFIX);
@@ -100,16 +100,16 @@ class TxStateMetaRocksDbPartitionStorage {
return ByteBuffer.allocate(KEY_SIZE_BYTES)
.order(BYTE_ORDER)
.put(prefix)
- .putInt(tableOrZoneId)
+ .putInt(zoneId)
.putShort((short) partitionId)
.array();
}
- private static byte[] createKeyPrefixForTableOrZone(byte prefix, int tableOrZoneId) {
- return ByteBuffer.allocate(TABLE_OR_ZONE_PREFIX_SIZE_BYTES)
+ private static byte[] createKeyPrefixForZone(byte prefix, int zoneId) {
+ return ByteBuffer.allocate(ZONE_PREFIX_SIZE_BYTES)
.order(BYTE_ORDER)
.put(prefix)
- .putInt(tableOrZoneId)
+ .putInt(zoneId)
.array();
}
@@ -132,33 +132,6 @@ class TxStateMetaRocksDbPartitionStorage {
}
}
- /**
- * Special method to be used with TX storages using a legacy format. Such storages saved last applied index and term inside a
- * different column family.
- */
- // TODO: remove this method after the colocation track migration, see https://issues.apache.org/jira/browse/IGNITE-22522.
- void startInCompatibilityMode(long lastAppliedIndex, long lastAppliedTerm) throws RocksDBException {
- byte[] lastAppliedBytes = columnFamily.get(lastAppliedKey);
-
- if (lastAppliedBytes != null) {
- ByteBuffer buf = ByteBuffer.wrap(lastAppliedBytes).order(BYTE_ORDER);
-
- this.lastAppliedIndex = buf.getLong();
- this.lastAppliedTerm = buf.getLong();
- } else {
- this.lastAppliedIndex = lastAppliedIndex;
- this.lastAppliedTerm = lastAppliedTerm;
- }
-
- config = columnFamily.get(confKey);
-
- byte[] leaseBytes = columnFamily.get(leaseInfoKey);
-
- if (leaseBytes != null) {
- leaseInfo = VersionedSerialization.fromBytes(leaseBytes, LeaseInfoSerializer.INSTANCE);
- }
- }
-
long lastAppliedIndex() {
return lastAppliedIndex;
}
@@ -222,9 +195,9 @@ class TxStateMetaRocksDbPartitionStorage {
leaseInfo = null;
}
- static void clearForTableOrZone(WriteBatch writeBatch, ColumnFamilyHandle cf, int tableOrZoneId) throws RocksDBException {
+ static void clearForZone(WriteBatch writeBatch, ColumnFamilyHandle cf, int zoneId) throws RocksDBException {
for (byte prefixByte : List.of(LAST_APPLIED_PREFIX, CONF_PREFIX, LEASE_INFO_PREFIX, SNAPSHOT_INFO_PREFIX)) {
- byte[] start = createKeyPrefixForTableOrZone(prefixByte, tableOrZoneId);
+ byte[] start = createKeyPrefixForZone(prefixByte, zoneId);
byte[] end = incrementPrefix(start);
writeBatch.deleteRange(cf, start, end);
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateRocksDbPartitionStorage.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateRocksDbPartitionStorage.java
index 61210110308..87485c3773b 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateRocksDbPartitionStorage.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateRocksDbPartitionStorage.java
@@ -21,7 +21,7 @@ import static java.util.Objects.requireNonNull;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
import static org.apache.ignite.internal.storage.util.StorageUtils.transitionToDestroyedState;
import static org.apache.ignite.internal.tx.storage.state.rocksdb.TxStateRocksDbSharedStorage.BYTE_ORDER;
-import static org.apache.ignite.internal.tx.storage.state.rocksdb.TxStateRocksDbStorage.TABLE_OR_ZONE_PREFIX_SIZE_BYTES;
+import static org.apache.ignite.internal.tx.storage.state.rocksdb.TxStateRocksDbStorage.ZONE_PREFIX_SIZE_BYTES;
import static org.apache.ignite.internal.util.ByteUtils.bytesToLong;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_STATE_STORAGE_ERR;
@@ -65,7 +65,7 @@ import org.rocksdb.WriteBatch;
*/
public class TxStateRocksDbPartitionStorage implements TxStatePartitionStorage {
/** Prefix length for the payload. Consists of tableId/zoneId (4 bytes) and partitionId (2 bytes), both in Big Endian. */
- public static final int PREFIX_SIZE_BYTES = TABLE_OR_ZONE_PREFIX_SIZE_BYTES + Short.BYTES;
+ public static final int PREFIX_SIZE_BYTES = ZONE_PREFIX_SIZE_BYTES + Short.BYTES;
/** Size of the key in the storage. Consists of {@link #PREFIX_SIZE_BYTES} and a UUID (2x {@link Long#BYTES}. */
private static final int FULL_KEY_SIZE_BYES = PREFIX_SIZE_BYTES + 2 * Long.BYTES;
@@ -85,8 +85,8 @@ public class TxStateRocksDbPartitionStorage implements TxStatePartitionStorage {
/** Shared TX state storage. */
private final TxStateRocksDbSharedStorage sharedStorage;
- /** Table/zone ID. */
- private final int tableOrZoneId;
+ /** Zone ID. */
+ private final int zoneId;
/** Current state of the storage. */
private final AtomicReference<StorageState> state = new AtomicReference<>(StorageState.RUNNABLE);
@@ -103,12 +103,12 @@ public class TxStateRocksDbPartitionStorage implements TxStatePartitionStorage {
TxStateRocksDbPartitionStorage(int partitionId, TxStateRocksDbStorage parentStorage) {
this.partitionId = partitionId;
this.sharedStorage = parentStorage.sharedStorage;
- this.tableOrZoneId = parentStorage.id;
+ this.zoneId = parentStorage.id;
this.dataColumnFamily = parentStorage.sharedStorage.txStateColumnFamily();
this.metaStorage = new TxStateMetaRocksDbPartitionStorage(
parentStorage.sharedStorage.txStateMetaColumnFamily(),
- tableOrZoneId,
+ zoneId,
partitionId
);
}
@@ -126,19 +126,7 @@ public class TxStateRocksDbPartitionStorage implements TxStatePartitionStorage {
public void start() {
busy(() -> {
try {
- // Read last applied index and term. This is only actual for older storage versions, where this information was not saved
- // in the meta storage, but rather in the data storage.
- // TODO: remove this after the colocation track migration, see https://issues.apache.org/jira/browse/IGNITE-22522.
- byte[] indexAndTermBytes = readLastAppliedIndexAndTerm();
-
- if (indexAndTermBytes == null) {
- metaStorage.start();
- } else {
- long lastAppliedIndex = bytesToLong(indexAndTermBytes);
- long lastAppliedTerm = bytesToLong(indexAndTermBytes, Long.BYTES);
-
- metaStorage.startInCompatibilityMode(lastAppliedIndex, lastAppliedTerm);
- }
+ metaStorage.start();
} catch (RocksDBException e) {
throw new TxStateStorageException(
TX_STATE_STORAGE_ERR,
@@ -287,7 +275,7 @@ public class TxStateRocksDbPartitionStorage implements TxStatePartitionStorage {
// This lower bound is the lowest possible key that goes after "lastAppliedIndexAndTermKey".
byte[] lowerBound = ByteBuffer.allocate(PREFIX_SIZE_BYTES + 1).order(BYTE_ORDER)
- .putInt(tableOrZoneId)
+ .putInt(zoneId)
.putShort(shortPartitionId(partitionId))
.put((byte) 0)
.array();
@@ -369,15 +357,6 @@ public class TxStateRocksDbPartitionStorage implements TxStatePartitionStorage {
updateData(writeBatch -> null, lastAppliedIndex, lastAppliedTerm);
}
- private byte @Nullable [] readLastAppliedIndexAndTerm() throws RocksDBException {
- byte[] lastAppliedIndexAndTermKey = ByteBuffer.allocate(PREFIX_SIZE_BYTES).order(BYTE_ORDER)
- .putInt(tableOrZoneId)
- .putShort(shortPartitionId(partitionId))
- .array();
-
- return dataColumnFamily.get(lastAppliedIndexAndTermKey);
- }
-
@Override
public void destroy() {
transitionToDestroyedState(state);
@@ -395,21 +374,21 @@ public class TxStateRocksDbPartitionStorage implements TxStatePartitionStorage {
private byte[] partitionStartPrefix() {
return ByteBuffer.allocate(PREFIX_SIZE_BYTES).order(BYTE_ORDER)
- .putInt(tableOrZoneId)
+ .putInt(zoneId)
.putShort(shortPartitionId(partitionId))
.array();
}
private byte[] partitionEndPrefix() {
return ByteBuffer.allocate(PREFIX_SIZE_BYTES).order(BYTE_ORDER)
- .putInt(tableOrZoneId)
+ .putInt(zoneId)
.putShort(shortPartitionId(partitionId + 1))
.array();
}
private byte[] txIdToKey(UUID txId) {
return ByteBuffer.allocate(FULL_KEY_SIZE_BYES).order(BYTE_ORDER)
- .putInt(tableOrZoneId)
+ .putInt(zoneId)
.putShort(shortPartitionId(partitionId))
.putLong(txId.getMostSignificantBits())
.putLong(txId.getLeastSignificantBits())
@@ -667,7 +646,7 @@ public class TxStateRocksDbPartitionStorage implements TxStatePartitionStorage {
}
private String createStorageInfo() {
- return "tableOrZone=" + tableOrZoneId + ", partitionId=" + partitionId;
+ return "zone=" + zoneId + ", partitionId=" + partitionId;
}
private <V> V busy(Supplier<V> supplier) {
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateRocksDbSharedStorage.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateRocksDbSharedStorage.java
index de82e766835..378a0335fef 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateRocksDbSharedStorage.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateRocksDbSharedStorage.java
@@ -64,9 +64,9 @@ import org.rocksdb.WriteBatch;
import org.rocksdb.WriteOptions;
/**
- * Shared RocksDB storage instance to be used in {@link TxStateRocksDbStorage}. Exists to make createTable/createZone operations faster,
+ * Shared RocksDB storage instance to be used in {@link TxStateRocksDbStorage}. Exists to make createZone operations faster,
* as well as reducing the amount of resources that would otherwise be used by multiple RocksDB instances, if they existed
- * on per-table/per-zone basis.
+ * on per-zone basis.
*/
public class TxStateRocksDbSharedStorage implements IgniteComponent {
static {
@@ -295,34 +295,34 @@ public class TxStateRocksDbSharedStorage implements IgniteComponent {
/**
* Destroys tx state storage for table or zone by its ID.
*
- * @param tableOrZoneId ID of the table or zone.
+ * @param zoneId ID of the zone.
*/
- public void destroyStorage(int tableOrZoneId) {
- byte[] dataStart = ByteBuffer.allocate(TxStateRocksDbStorage.TABLE_OR_ZONE_PREFIX_SIZE_BYTES)
+ public void destroyStorage(int zoneId) {
+ byte[] dataStart = ByteBuffer.allocate(TxStateRocksDbStorage.ZONE_PREFIX_SIZE_BYTES)
.order(BYTE_ORDER)
- .putInt(tableOrZoneId)
+ .putInt(zoneId)
.array();
byte[] dataEnd = incrementPrefix(dataStart);
try (WriteBatch writeBatch = new WriteBatch()) {
writeBatch.deleteRange(txStateColumnFamily.handle(), dataStart, dataEnd);
- TxStateMetaRocksDbPartitionStorage.clearForTableOrZone(writeBatch, txStateMetaColumnFamily().handle(), tableOrZoneId);
+ TxStateMetaRocksDbPartitionStorage.clearForZone(writeBatch, txStateMetaColumnFamily().handle(), zoneId);
db.write(writeOptions, writeBatch);
} catch (Exception e) {
- throw new TxStateStorageException("Failed to destroy the transaction state storage [tableOrZoneId={}]", e, tableOrZoneId);
+ throw new TxStateStorageException("Failed to destroy the transaction state storage [zoneId={}]", e, zoneId);
}
}
/**
- * Returns IDs of tables/zones for which there are tx state partition storages on disk. Those were created and flushed to disk; either
+ * Returns IDs of zones for which there are tx state partition storages on disk. Those were created and flushed to disk; either
* destruction was not started for them, or it failed.
*
* <p>This method should only be called when the tx state storage is not accessed otherwise (so no storages in it can appear or
* be destroyed in parallel with this call).
*/
- public Set<Integer> tableOrZoneIdsOnDisk() {
+ public Set<Integer> zoneIdsOnDisk() {
Set<Integer> ids = new HashSet<>();
byte[] lastAppliedGlobalPrefix = {TxStateMetaRocksDbPartitionStorage.LAST_APPLIED_PREFIX};
@@ -336,8 +336,8 @@ public class TxStateRocksDbSharedStorage implements IgniteComponent {
while (it.isValid()) {
byte[] key = it.key();
- int tableOrZoneId = ByteUtils.bytesToInt(key, lastAppliedGlobalPrefix.length);
- ids.add(tableOrZoneId);
+ int zoneId = ByteUtils.bytesToInt(key, lastAppliedGlobalPrefix.length);
+ ids.add(zoneId);
it.next();
}
@@ -346,7 +346,7 @@ public class TxStateRocksDbSharedStorage implements IgniteComponent {
// the iteration space.
it.status();
} catch (RocksDBException e) {
- throw new TxStateStorageException(INTERNAL_ERR, "Cannot get table/zone IDs", e);
+ throw new TxStateStorageException(INTERNAL_ERR, "Cannot get zone IDs", e);
}
return unmodifiableSet(ids);
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateRocksDbStorage.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateRocksDbStorage.java
index a6da379dcd5..1c5d39cde68 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateRocksDbStorage.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateRocksDbStorage.java
@@ -35,10 +35,10 @@ import org.jetbrains.annotations.Nullable;
* RocksDb implementation of {@link TxStateStorage}.
*/
public class TxStateRocksDbStorage implements TxStateStorage {
- static final int TABLE_OR_ZONE_ID_SIZE_BYTES = Integer.BYTES;
+ static final int ZONE_ID_SIZE_BYTES = Integer.BYTES;
- /** Prefix length for the payload within a table/zone. Consists of tableId/zoneId (4 bytes) in Big Endian. */
- static final int TABLE_OR_ZONE_PREFIX_SIZE_BYTES = TABLE_OR_ZONE_ID_SIZE_BYTES;
+ /** Prefix length for the payload within a zone. Consists of zoneId (4 bytes) in Big Endian. */
+ static final int ZONE_PREFIX_SIZE_BYTES = ZONE_ID_SIZE_BYTES;
/** Partition storages. */
private final AtomicReferenceArray<TxStateRocksDbPartitionStorage> storages;
@@ -49,7 +49,7 @@ public class TxStateRocksDbStorage implements TxStateStorage {
/** Busy lock to stop synchronously. */
private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
- /** Table/zone ID. */
+ /** Zone ID. */
final int id;
final TxStateRocksDbSharedStorage sharedStorage;
@@ -57,7 +57,7 @@ public class TxStateRocksDbStorage implements TxStateStorage {
/**
* Constructor.
*
- * @param id Table/zone ID.
+ * @param id zone ID.
* @param partitions Count of partitions.
*/
public TxStateRocksDbStorage(
@@ -80,7 +80,7 @@ public class TxStateRocksDbStorage implements TxStateStorage {
if (partitionId < 0 || partitionId >= storages.length()) {
throw new IllegalArgumentException(S.toString(
"Unable to access partition with id outside of configured range",
- "tableId/zoneId", id, false,
+ "zoneId", id, false,
"partitionId", partitionId, false,
"partitions", storages.length(), false
));
@@ -156,7 +156,7 @@ public class TxStateRocksDbStorage implements TxStateStorage {
reverse(resources);
closeAll(resources);
} catch (Exception e) {
- throw new TxStateStorageException("Failed to stop transaction state storage [tableOrZoneId={}]", e, id);
+ throw new TxStateStorageException("Failed to stop transaction state storage [zoneId={}]", e, id);
}
}
diff --git a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/TxCleanupTest.java b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/TxCleanupTest.java
index 94468a7cc77..fbbf820a26d 100644
--- a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/TxCleanupTest.java
+++ b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/TxCleanupTest.java
@@ -56,7 +56,6 @@ import org.apache.ignite.internal.network.TopologyService;
import org.apache.ignite.internal.placementdriver.PlacementDriver;
import org.apache.ignite.internal.placementdriver.TestReplicaMetaImpl;
import org.apache.ignite.internal.replicator.ReplicaService;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.testframework.IgniteAbstractTest;
import org.apache.ignite.internal.tx.configuration.TransactionConfiguration;
@@ -92,7 +91,7 @@ public class TxCleanupTest extends IgniteAbstractTest {
private static final int TABLE_ID_2 = 20;
private static final int TABLE_ID_3 = 30;
- private static final Map<ReplicationGroupId, PartitionEnlistment> PARTITIONS = Map.of(
+ private static final Map<ZonePartitionId, PartitionEnlistment> PARTITIONS = Map.of(
ZONE_PARTITION_ID_1, new PartitionEnlistment(LOCAL_NODE.name(), Set.of(TABLE_ID_1)),
ZONE_PARTITION_ID_2, new PartitionEnlistment(LOCAL_NODE.name(), Set.of(TABLE_ID_2)),
ZONE_PARTITION_ID_3, new PartitionEnlistment(LOCAL_NODE.name(), Set.of(TABLE_ID_3))
diff --git a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/TxManagerTest.java b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/TxManagerTest.java
index ccbc5b4f477..82808c30793 100644
--- a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/TxManagerTest.java
+++ b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/TxManagerTest.java
@@ -84,7 +84,6 @@ import org.apache.ignite.internal.placementdriver.PlacementDriver;
import org.apache.ignite.internal.placementdriver.ReplicaMeta;
import org.apache.ignite.internal.placementdriver.TestReplicaMetaImpl;
import org.apache.ignite.internal.replicator.ReplicaService;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.replicator.exception.PrimaryReplicaMissException;
import org.apache.ignite.internal.testframework.ExecutorServiceExtension;
@@ -243,7 +242,7 @@ public class TxManagerTest extends IgniteAbstractTest {
InternalTransaction tx = txManager.beginExplicitRw(hybridTimestampTracker, InternalTxOptions.defaults());
- ReplicationGroupId partitionIdForEnlistment = new ZonePartitionId(1, 0);
+ ZonePartitionId partitionIdForEnlistment = new ZonePartitionId(1, 0);
tx.assignCommitPartition(partitionIdForEnlistment);
tx.enlist(partitionIdForEnlistment, 10, REMOTE_NODE.name(), 1L);
@@ -340,7 +339,7 @@ public class TxManagerTest extends IgniteAbstractTest {
InternalTransaction tx = txManager.beginExplicitRw(hybridTimestampTracker, InternalTxOptions.defaults());
- ReplicationGroupId replicationGroupId = new ZonePartitionId(1, 0);
+ ZonePartitionId replicationGroupId = new ZonePartitionId(1, 0);
tx.enlist(replicationGroupId, 10, REMOTE_NODE.name(), 1L);
tx.assignCommitPartition(replicationGroupId);
@@ -362,7 +361,7 @@ public class TxManagerTest extends IgniteAbstractTest {
InternalTransaction tx = txManager.beginExplicitRw(hybridTimestampTracker, InternalTxOptions.defaults());
- ReplicationGroupId replicationGroupId = new ZonePartitionId(1, 0);
+ ZonePartitionId replicationGroupId = new ZonePartitionId(1, 0);
tx.enlist(replicationGroupId, 10, REMOTE_NODE.name(), 1L);
tx.assignCommitPartition(replicationGroupId);
@@ -391,7 +390,7 @@ public class TxManagerTest extends IgniteAbstractTest {
InternalTransaction tx = txManager.beginExplicitRw(hybridTimestampTracker, InternalTxOptions.defaults());
- ReplicationGroupId replicationGroupId = new ZonePartitionId(1, 0);
+ ZonePartitionId replicationGroupId = new ZonePartitionId(1, 0);
transactionInflights.track(tx.id());
tx.enlist(replicationGroupId, 10, REMOTE_NODE.name(), 1L);
@@ -420,7 +419,7 @@ public class TxManagerTest extends IgniteAbstractTest {
InternalTransaction tx = txManager.beginExplicitRw(hybridTimestampTracker, InternalTxOptions.defaults());
- ReplicationGroupId replicationGroupId = new ZonePartitionId(1, 0);
+ ZonePartitionId replicationGroupId = new ZonePartitionId(1, 0);
transactionInflights.track(tx.id());
tx.enlist(replicationGroupId, 10, REMOTE_NODE.name(), 1L);
@@ -662,11 +661,11 @@ public class TxManagerTest extends IgniteAbstractTest {
String nodeName = "SomeNode";
- ReplicationGroupId partitionIdForEnlistment1 = new ZonePartitionId(1, 0);
+ ZonePartitionId partitionIdForEnlistment1 = new ZonePartitionId(1, 0);
tx.enlist(partitionIdForEnlistment1, 10, nodeName, 1L);
tx.assignCommitPartition(partitionIdForEnlistment1);
- ReplicationGroupId partitionIdForEnlistment2 = new ZonePartitionId(2, 0);
+ ZonePartitionId partitionIdForEnlistment2 = new ZonePartitionId(2, 0);
tx.enlist(partitionIdForEnlistment2, 20, nodeName, 1L);
when(placementDriver.getPrimaryReplica(eq(partitionIdForEnlistment1), any()))
@@ -827,7 +826,7 @@ public class TxManagerTest extends IgniteAbstractTest {
private InternalTransaction prepareTransaction() {
InternalTransaction tx = txManager.beginExplicitRw(hybridTimestampTracker, InternalTxOptions.defaults());
- ReplicationGroupId replicationGroupId = new ZonePartitionId(1, 0);
+ ZonePartitionId replicationGroupId = new ZonePartitionId(1, 0);
tx.enlist(replicationGroupId, 10, REMOTE_NODE.name(), 1L);
tx.assignCommitPartition(replicationGroupId);
diff --git a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/TxMetaSerializerTest.java b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/TxMetaSerializerTest.java
index 562a74e1840..276d416d460 100644
--- a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/TxMetaSerializerTest.java
+++ b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/TxMetaSerializerTest.java
@@ -21,12 +21,12 @@ import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import java.util.Base64;
import java.util.List;
import java.util.Set;
import org.apache.ignite.internal.hlc.HybridTimestamp;
-import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.tx.impl.EnlistedPartitionGroup;
import org.apache.ignite.internal.versioned.VersionedSerialization;
@@ -34,7 +34,6 @@ import org.junit.jupiter.api.Test;
class TxMetaSerializerTest {
private static final String V1_SERIALIZED_BASE64 = "Ae++QwUD6QcQ0Q8a////f///gIAE";
- private static final String V2_WITH_TABLES_SERIALIZED_BASE64 = "Au++QwAFA+kHEALpB9EPGgLRD////3///4CABA==";
private static final String V2_WITH_ZONES_SERIALIZED_BASE64 = "Au++QwEFA+kHEANmZdEPGgPKAckB////f///gIAE";
private final TxMetaSerializer serializer = new TxMetaSerializer();
@@ -44,8 +43,8 @@ class TxMetaSerializerTest {
TxMeta originalMeta = new TxMeta(
TxState.COMMITTED,
List.of(
- new EnlistedPartitionGroup(new TablePartitionId(1000, 15), Set.of(100, 101)),
- new EnlistedPartitionGroup(new TablePartitionId(2000, 25), Set.of(200, 201))
+ new EnlistedPartitionGroup(new ZonePartitionId(1000, 15), Set.of(100, 101)),
+ new EnlistedPartitionGroup(new ZonePartitionId(2000, 25), Set.of(200, 201))
),
HybridTimestamp.MAX_VALUE
);
@@ -61,8 +60,8 @@ class TxMetaSerializerTest {
TxMeta originalMeta = new TxMeta(
TxState.ABANDONED,
List.of(
- new EnlistedPartitionGroup(new TablePartitionId(1000, 15), Set.of(100, 101)),
- new EnlistedPartitionGroup(new TablePartitionId(2000, 25), Set.of(200, 201))
+ new EnlistedPartitionGroup(new ZonePartitionId(1000, 15), Set.of(100, 101)),
+ new EnlistedPartitionGroup(new ZonePartitionId(2000, 25), Set.of(200, 201))
),
null
);
@@ -74,43 +73,10 @@ class TxMetaSerializerTest {
}
@Test
- void v1CanBeDeserialized() {
+ void v1IsNotSupported() {
byte[] bytes = Base64.getDecoder().decode(V1_SERIALIZED_BASE64);
- TxMeta restoredMeta = VersionedSerialization.fromBytes(bytes, serializer);
- assertThat(restoredMeta.txState(), is(TxState.ABANDONED));
- assertThat(restoredMeta.enlistedPartitions(), contains(
- new EnlistedPartitionGroup(new TablePartitionId(1000, 15), Set.of(1000)),
- new EnlistedPartitionGroup(new TablePartitionId(2000, 25), Set.of(2000))
- ));
- assertThat(restoredMeta.commitTimestamp(), is(HybridTimestamp.MAX_VALUE));
- }
-
- @Test
- void v2WithTablesCanBeDeserialized() {
- byte[] bytes = Base64.getDecoder().decode(V2_WITH_TABLES_SERIALIZED_BASE64);
- TxMeta restoredMeta = VersionedSerialization.fromBytes(bytes, serializer);
-
- assertThat(restoredMeta.txState(), is(TxState.ABANDONED));
- assertThat(restoredMeta.enlistedPartitions(), contains(
- new EnlistedPartitionGroup(new TablePartitionId(1000, 15), Set.of(1000)),
- new EnlistedPartitionGroup(new TablePartitionId(2000, 25), Set.of(2000))
- ));
- assertThat(restoredMeta.commitTimestamp(), is(HybridTimestamp.MAX_VALUE));
- }
-
- @SuppressWarnings("unused")
- private String v2TablesSerializedBase64() {
- TxMeta originalMeta = new TxMeta(
- TxState.ABANDONED,
- List.of(
- new EnlistedPartitionGroup(new TablePartitionId(1000, 15), Set.of(1000)),
- new EnlistedPartitionGroup(new TablePartitionId(2000, 25), Set.of(2000))
- ),
- HybridTimestamp.MAX_VALUE
- );
- byte[] v1Bytes = VersionedSerialization.toBytes(originalMeta, serializer);
- return Base64.getEncoder().encodeToString(v1Bytes);
+ assertThrows(IllegalArgumentException.class, () -> VersionedSerialization.fromBytes(bytes, serializer));
}
@Test
diff --git a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/OrphanDetectorTest.java b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/OrphanDetectorTest.java
index ed4cfaf5284..a05db78716b 100644
--- a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/OrphanDetectorTest.java
+++ b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/OrphanDetectorTest.java
@@ -49,7 +49,6 @@ import org.apache.ignite.internal.network.TopologyService;
import org.apache.ignite.internal.placementdriver.PlacementDriver;
import org.apache.ignite.internal.placementdriver.TestReplicaMetaImpl;
import org.apache.ignite.internal.replicator.ReplicaService;
-import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.storage.RowId;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
@@ -173,18 +172,18 @@ public class OrphanDetectorTest extends BaseIgniteAbstractTest {
void testNoTriggerCommittedState() {
UUID orphanTxId = idGenerator.transactionIdFor(clock.now());
- TablePartitionId tpId = new TablePartitionId(1, 0);
+ ZonePartitionId zonePartitionId = new ZonePartitionId(1, 0);
- RowId rowId = new RowId(tpId.partitionId());
+ RowId rowId = new RowId(zonePartitionId.partitionId());
// Coordinator is dead.
when(topologyService.getById(eq(LOCAL_NODE.id()))).thenReturn(null);
- lockManager.acquire(orphanTxId, new LockKey(tpId.tableId(), rowId), LockMode.X);
+ lockManager.acquire(orphanTxId, new LockKey(zonePartitionId.zoneId(), rowId), LockMode.X);
UUID concurrentTxId = idGenerator.transactionIdFor(clock.now());
- TxStateMeta committedState = new TxStateMeta(TxState.COMMITTED, LOCAL_NODE.id(), tpId, clock.now(), null, null);
+ TxStateMeta committedState = new TxStateMeta(TxState.COMMITTED, LOCAL_NODE.id(), zonePartitionId, clock.now(), null, null);
txStateMetaStorage.updateMeta(orphanTxId, stateMeta -> committedState);
@@ -205,18 +204,18 @@ public class OrphanDetectorTest extends BaseIgniteAbstractTest {
void testNoTriggerAbortedState() {
UUID orphanTxId = idGenerator.transactionIdFor(clock.now());
- TablePartitionId tpId = new TablePartitionId(1, 0);
+ ZonePartitionId zonePartitionId = new ZonePartitionId(1, 0);
- RowId rowId = new RowId(tpId.partitionId());
+ RowId rowId = new RowId(zonePartitionId.partitionId());
// Coordinator is dead.
when(topologyService.getById(eq(LOCAL_NODE.id()))).thenReturn(null);
- lockManager.acquire(orphanTxId, new LockKey(tpId.tableId(), rowId), LockMode.X);
+ lockManager.acquire(orphanTxId, new LockKey(zonePartitionId.zoneId(), rowId), LockMode.X);
UUID concurrentTxId = idGenerator.transactionIdFor(clock.now());
- TxStateMeta abortedState = new TxStateMeta(TxState.ABORTED, LOCAL_NODE.id(), tpId, null, null, null);
+ TxStateMeta abortedState = new TxStateMeta(TxState.ABORTED, LOCAL_NODE.id(), zonePartitionId, null, null, null);
txStateMetaStorage.updateMeta(orphanTxId, stateMeta -> abortedState);
@@ -237,15 +236,15 @@ public class OrphanDetectorTest extends BaseIgniteAbstractTest {
void testNoTriggerFinishingState() {
UUID orphanTxId = idGenerator.transactionIdFor(clock.now());
- TablePartitionId tpId = new TablePartitionId(1, 0);
+ ZonePartitionId zonePartitionId = new ZonePartitionId(1, 0);
- RowId rowId = new RowId(tpId.partitionId());
+ RowId rowId = new RowId(zonePartitionId.partitionId());
- lockManager.acquire(orphanTxId, new LockKey(tpId.tableId(), rowId), LockMode.X);
+ lockManager.acquire(orphanTxId, new LockKey(zonePartitionId.zoneId(), rowId), LockMode.X);
UUID concurrentTxId = idGenerator.transactionIdFor(clock.now());
- TxStateMeta finishingState = new TxStateMeta(TxState.FINISHING, LOCAL_NODE.id(), tpId, null, null, null);
+ TxStateMeta finishingState = new TxStateMeta(TxState.FINISHING, LOCAL_NODE.id(), zonePartitionId, null, null, null);
txStateMetaStorage.updateMeta(orphanTxId, stateMeta -> finishingState);
@@ -269,15 +268,15 @@ public class OrphanDetectorTest extends BaseIgniteAbstractTest {
void testNoTriggerCoordinatorAlive() {
UUID orphanTxId = idGenerator.transactionIdFor(clock.now());
- TablePartitionId tpId = new TablePartitionId(1, 0);
+ ZonePartitionId zonePartitionId = new ZonePartitionId(1, 0);
- RowId rowId = new RowId(tpId.partitionId());
+ RowId rowId = new RowId(zonePartitionId.partitionId());
- lockManager.acquire(orphanTxId, new LockKey(tpId.tableId(), rowId), LockMode.X);
+ lockManager.acquire(orphanTxId, new LockKey(zonePartitionId.zoneId(), rowId), LockMode.X);
UUID concurrentTxId = idGenerator.transactionIdFor(clock.now());
- TxStateMeta pendingState = new TxStateMeta(TxState.PENDING, LOCAL_NODE.id(), tpId, null, null, null);
+ TxStateMeta pendingState = new TxStateMeta(TxState.PENDING, LOCAL_NODE.id(), zonePartitionId, null, null, null);
txStateMetaStorage.updateMeta(orphanTxId, stateMeta -> pendingState);
diff --git a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/ReadWriteTransactionImplTest.java b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/ReadWriteTransactionImplTest.java
index 5c67836f986..349cafd4efd 100644
--- a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/ReadWriteTransactionImplTest.java
+++ b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/ReadWriteTransactionImplTest.java
@@ -37,7 +37,6 @@ import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.hlc.HybridTimestampTracker;
import org.apache.ignite.internal.network.ClusterNodeImpl;
import org.apache.ignite.internal.network.InternalClusterNode;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
import org.apache.ignite.internal.tx.TxManager;
@@ -64,7 +63,7 @@ class ReadWriteTransactionImplTest extends BaseIgniteAbstractTest {
private static final int ZONE_ID = 2;
/** Transaction commit partition id. */
- private final ReplicationGroupId txCommitPart = new ZonePartitionId(ZONE_ID, 0);
+ private final ZonePartitionId txCommitPart = new ZonePartitionId(ZONE_ID, 0);
@Mock
private TxManager txManager;
@@ -81,7 +80,7 @@ class ReadWriteTransactionImplTest extends BaseIgniteAbstractTest {
UUID txId = TestTransactionIds.TRANSACTION_ID_GENERATOR.transactionIdFor(beginTs);
var tx = new ReadWriteTransactionImpl(
- txManager, HybridTimestampTracker.atomicTracker(null), txId, CLUSTER_NODE.id(), false, 10_000, true
+ txManager, HybridTimestampTracker.atomicTracker(null), txId, CLUSTER_NODE.id(), false, 10_000
);
assertThat(tx.schemaTimestamp(), is(beginTs));
@@ -114,7 +113,7 @@ class ReadWriteTransactionImplTest extends BaseIgniteAbstractTest {
UUID txId = TestTransactionIds.TRANSACTION_ID_GENERATOR.transactionIdFor(beginTs);
var tx = new ReadWriteTransactionImpl(
- txManager, HybridTimestampTracker.atomicTracker(null), txId, CLUSTER_NODE.id(), false, 10_000, true
+ txManager, HybridTimestampTracker.atomicTracker(null), txId, CLUSTER_NODE.id(), false, 10_000
);
tx.assignCommitPartition(txCommitPart);
diff --git a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistryConcurrentTest.java b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistryConcurrentTest.java
index 42cfe948736..3d4e773727e 100644
--- a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistryConcurrentTest.java
+++ b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistryConcurrentTest.java
@@ -28,8 +28,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.ignite.internal.hlc.HybridTimestamp;
-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.testframework.BaseIgniteAbstractTest;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.tx.PendingTxPartitionEnlistment;
@@ -185,7 +184,7 @@ class TransactionExpirationRegistryConcurrentTest extends BaseIgniteAbstractTest
}
@Override
- public PendingTxPartitionEnlistment enlistedPartition(ReplicationGroupId replicationGroupId) {
+ public PendingTxPartitionEnlistment enlistedPartition(ZonePartitionId replicationGroupId) {
return null;
}
@@ -195,18 +194,17 @@ class TransactionExpirationRegistryConcurrentTest extends BaseIgniteAbstractTest
}
@Override
- public boolean assignCommitPartition(ReplicationGroupId commitPartitionId) {
+ public boolean assignCommitPartition(ZonePartitionId commitPartitionId) {
return false;
}
@Override
- public TablePartitionId commitPartition() {
+ public ZonePartitionId commitPartition() {
return null;
}
@Override
- public void enlist(ReplicationGroupId replicationGroupId, int tableId, String primaryNodeConsistentId, long consistencyToken) {
-
+ public void enlist(ZonePartitionId replicationGroupId, int tableId, String primaryNodeConsistentId, long consistencyToken) {
}
@Override
diff --git a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/storage/state/rocksdb/RocksDbTxStatePartitionStorageTest.java b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/storage/state/rocksdb/RocksDbTxStatePartitionStorageTest.java
index 36642b00e34..92c1a1e5eb6 100644
--- a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/storage/state/rocksdb/RocksDbTxStatePartitionStorageTest.java
+++ b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/storage/state/rocksdb/RocksDbTxStatePartitionStorageTest.java
@@ -61,9 +61,9 @@ public class RocksDbTxStatePartitionStorageTest extends AbstractTxStatePartition
private ExecutorService executor;
@Override
- protected TxStateRocksDbStorage createTableOrZoneStorage() {
+ protected TxStateRocksDbStorage createZoneStorage() {
return new TxStateRocksDbStorage(
- TABLE_ID,
+ ZONE_ID,
3,
sharedStorage
);
@@ -96,7 +96,7 @@ public class RocksDbTxStatePartitionStorageTest extends AbstractTxStatePartition
@Test
void testRestartStorageInProgressOfRebalance() {
- TxStatePartitionStorage storage = tableStorage.getOrCreatePartitionStorage(0);
+ TxStatePartitionStorage storage = txStateStorage.getOrCreatePartitionStorage(0);
List<IgniteBiTuple<UUID, TxMeta>> rows = List.of(
randomTxMetaTuple(1, UUID.randomUUID()),
@@ -110,13 +110,13 @@ public class RocksDbTxStatePartitionStorageTest extends AbstractTxStatePartition
assertThat(storage.flush(), willCompleteSuccessfully());
- tableStorage.close();
+ txStateStorage.close();
- tableStorage = createTableOrZoneStorage();
+ txStateStorage = createZoneStorage();
- tableStorage.start();
+ txStateStorage.start();
- storage = tableStorage.getOrCreatePartitionStorage(0);
+ storage = txStateStorage.getOrCreatePartitionStorage(0);
checkMeta(storage, REBALANCE_IN_PROGRESS, REBALANCE_IN_PROGRESS, GROUP_CONFIGURATION, LEASE_INFO, SNAPSHOT_INFO);
diff --git a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateMetaRocksDbPartitionStorageTest.java b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateMetaRocksDbPartitionStorageTest.java
index 0bf7b92b87d..bb000ffc9a6 100644
--- a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateMetaRocksDbPartitionStorageTest.java
+++ b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateMetaRocksDbPartitionStorageTest.java
@@ -72,8 +72,8 @@ class TxStateMetaRocksDbPartitionStorageTest extends IgniteAbstractTest {
assertThat(sharedStorage.stopAsync(new ComponentContext()), willCompleteSuccessfully());
}
- private TxStateMetaRocksDbPartitionStorage createMetaStorage(int tableId, int partitionId) {
- return new TxStateMetaRocksDbPartitionStorage(sharedStorage.txStateMetaColumnFamily(), tableId, partitionId);
+ private TxStateMetaRocksDbPartitionStorage createMetaStorage(int zoneId, int partitionId) {
+ return new TxStateMetaRocksDbPartitionStorage(sharedStorage.txStateMetaColumnFamily(), zoneId, partitionId);
}
@Test
@@ -134,36 +134,6 @@ class TxStateMetaRocksDbPartitionStorageTest extends IgniteAbstractTest {
assertThat(storage.leaseInfo(), is(leaseInfo));
}
- @Test
- void testStartInCompatibilityMode() throws RocksDBException {
- TxStateMetaRocksDbPartitionStorage storage = createMetaStorage(1, 2);
-
- storage.startInCompatibilityMode(5, 6);
-
- assertThat(storage.lastAppliedIndex(), is(5L));
- assertThat(storage.lastAppliedTerm(), is(6L));
- assertThat(storage.configuration(), is(nullValue()));
- assertThat(storage.leaseInfo(), is(nullValue()));
-
- try (
- var writeBatch = new WriteBatch();
- var options = new WriteOptions()
- ) {
- storage.updateLastApplied(writeBatch, 10, 20);
-
- sharedStorage.db().write(options, writeBatch);
- }
-
- storage = createMetaStorage(1, 2);
-
- storage.startInCompatibilityMode(5, 6);
-
- assertThat(storage.lastAppliedIndex(), is(10L));
- assertThat(storage.lastAppliedTerm(), is(20L));
- assertThat(storage.configuration(), is(nullValue()));
- assertThat(storage.leaseInfo(), is(nullValue()));
- }
-
@Test
void testClear() throws RocksDBException {
TxStateMetaRocksDbPartitionStorage storage = createMetaStorage(1, 2);
diff --git a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateRocksDbSharedStorageTest.java b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateRocksDbSharedStorageTest.java
index 31677831a61..3f995f60867 100644
--- a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateRocksDbSharedStorageTest.java
+++ b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/storage/state/rocksdb/TxStateRocksDbSharedStorageTest.java
@@ -79,7 +79,7 @@ class TxStateRocksDbSharedStorageTest {
@ParameterizedTest
@ValueSource(booleans = {true, false})
- void remembersCreatedTableOrZoneIdsOnDisk(boolean restart) {
+ void remembersCreatedZoneIdsOnDisk(boolean restart) {
createZoneTxStateStorageLeavingTraceOnDisk(1);
createZoneTxStateStorageLeavingTraceOnDisk(3);
@@ -87,7 +87,7 @@ class TxStateRocksDbSharedStorageTest {
flushAndRestartSharedStorage();
}
- assertThat(sharedStorage.tableOrZoneIdsOnDisk(), containsInAnyOrder(1, 3));
+ assertThat(sharedStorage.zoneIdsOnDisk(), containsInAnyOrder(1, 3));
}
private void createZoneTxStateStorageLeavingTraceOnDisk(int zoneId) {
@@ -109,7 +109,7 @@ class TxStateRocksDbSharedStorageTest {
@ParameterizedTest
@ValueSource(booleans = {true, false})
- void tableOrZoneIdsOnDiskGetRemovedOnPersistedDestruction(boolean restart) {
+ void zoneIdsOnDiskGetRemovedOnPersistedDestruction(boolean restart) {
createZoneTxStateStorageLeavingTraceOnDisk(1);
createZoneTxStateStorageLeavingTraceOnDisk(3);
@@ -119,6 +119,6 @@ class TxStateRocksDbSharedStorageTest {
flushAndRestartSharedStorage();
}
- assertThat(sharedStorage.tableOrZoneIdsOnDisk(), contains(3));
+ assertThat(sharedStorage.zoneIdsOnDisk(), contains(3));
}
}
diff --git a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/storage/state/test/TestTxStatePartitionStorageTest.java b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/storage/state/test/TestTxStatePartitionStorageTest.java
index 293cac400fd..26c8040e747 100644
--- a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/storage/state/test/TestTxStatePartitionStorageTest.java
+++ b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/storage/state/test/TestTxStatePartitionStorageTest.java
@@ -25,7 +25,7 @@ import org.apache.ignite.internal.tx.storage.state.TxStateStorage;
* Tx partition storage test for test implementation based on {@link ConcurrentHashMap}.
*/
public class TestTxStatePartitionStorageTest extends AbstractTxStatePartitionStorageTest {
- @Override protected TxStateStorage createTableOrZoneStorage() {
+ @Override protected TxStateStorage createZoneStorage() {
return new TestTxStateStorage();
}
}
diff --git a/modules/transactions/src/testFixtures/java/org/apache/ignite/internal/tx/storage/state/AbstractTxStatePartitionStorageTest.java b/modules/transactions/src/testFixtures/java/org/apache/ignite/internal/tx/storage/state/AbstractTxStatePartitionStorageTest.java
index c7b1c3a1d90..4c37af41498 100644
--- a/modules/transactions/src/testFixtures/java/org/apache/ignite/internal/tx/storage/state/AbstractTxStatePartitionStorageTest.java
+++ b/modules/transactions/src/testFixtures/java/org/apache/ignite/internal/tx/storage/state/AbstractTxStatePartitionStorageTest.java
@@ -53,7 +53,7 @@ import java.util.stream.IntStream;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.lang.IgniteBiTuple;
import org.apache.ignite.internal.lang.IgniteInternalException;
-import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.storage.engine.MvPartitionMeta;
import org.apache.ignite.internal.storage.lease.LeaseInfo;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
@@ -73,29 +73,31 @@ import org.junit.jupiter.api.function.Executable;
public abstract class AbstractTxStatePartitionStorageTest extends BaseIgniteAbstractTest {
protected static final int TABLE_ID = 1;
+ protected static final int ZONE_ID = 2;
+
protected static final byte[] GROUP_CONFIGURATION = {1, 2, 3};
protected static final byte[] SNAPSHOT_INFO = {4, 5, 6};
protected static final LeaseInfo LEASE_INFO = new LeaseInfo(1, UUID.randomUUID(), "node");
- protected TxStateStorage tableStorage;
+ protected TxStateStorage txStateStorage;
/**
* Creates {@link TxStatePartitionStorage} to test.
*/
- protected abstract TxStateStorage createTableOrZoneStorage();
+ protected abstract TxStateStorage createZoneStorage();
@BeforeEach
protected void beforeTest() {
- tableStorage = createTableOrZoneStorage();
+ txStateStorage = createZoneStorage();
- tableStorage.start();
+ txStateStorage.start();
}
@AfterEach
protected void afterTest() throws Exception {
- tableStorage.close();
+ txStateStorage.close();
}
@Test
@@ -115,7 +117,7 @@ public abstract class AbstractTxStatePartitionStorageTest extends BaseIgniteAbst
}
private void testPutGetRemove0(BiConsumer<TxStatePartitionStorage, Set<UUID>> removeOp) {
- TxStatePartitionStorage storage = tableStorage.getOrCreatePartitionStorage(0);
+ TxStatePartitionStorage storage = txStateStorage.getOrCreatePartitionStorage(0);
List<UUID> txIds = new ArrayList<>();
@@ -163,7 +165,7 @@ public abstract class AbstractTxStatePartitionStorageTest extends BaseIgniteAbst
private static List<EnlistedPartitionGroup> generateEnlistedPartitions(int c) {
return IntStream.range(0, c)
- .mapToObj(partitionNumber -> new EnlistedPartitionGroup(new TablePartitionId(TABLE_ID, partitionNumber), Set.of(TABLE_ID)))
+ .mapToObj(partitionNumber -> new EnlistedPartitionGroup(new ZonePartitionId(ZONE_ID, partitionNumber), Set.of(TABLE_ID)))
.collect(toList());
}
@@ -179,7 +181,7 @@ public abstract class AbstractTxStatePartitionStorageTest extends BaseIgniteAbst
@Test
public void testRemoveAll() {
- TxStatePartitionStorage storage = tableStorage.getOrCreatePartitionStorage(0);
+ TxStatePartitionStorage storage = txStateStorage.getOrCreatePartitionStorage(0);
List<UUID> txIds = new ArrayList<>();
@@ -227,7 +229,7 @@ public abstract class AbstractTxStatePartitionStorageTest extends BaseIgniteAbst
@Test
public void testCas() {
- TxStatePartitionStorage storage = tableStorage.getOrCreatePartitionStorage(0);
+ TxStatePartitionStorage storage = txStateStorage.getOrCreatePartitionStorage(0);
UUID txId = UUID.randomUUID();
@@ -270,9 +272,9 @@ public abstract class AbstractTxStatePartitionStorageTest extends BaseIgniteAbst
@Test
public void testScan() {
- TxStatePartitionStorage storage0 = tableStorage.getOrCreatePartitionStorage(0);
- TxStatePartitionStorage storage1 = tableStorage.getOrCreatePartitionStorage(1);
- TxStatePartitionStorage storage2 = tableStorage.getOrCreatePartitionStorage(2);
+ TxStatePartitionStorage storage0 = txStateStorage.getOrCreatePartitionStorage(0);
+ TxStatePartitionStorage storage1 = txStateStorage.getOrCreatePartitionStorage(1);
+ TxStatePartitionStorage storage2 = txStateStorage.getOrCreatePartitionStorage(2);
Map<UUID, TxMeta> txs = new HashMap<>();
@@ -303,8 +305,8 @@ public abstract class AbstractTxStatePartitionStorageTest extends BaseIgniteAbst
@Test
public void testDestroy() {
- TxStatePartitionStorage storage0 = tableStorage.getOrCreatePartitionStorage(0);
- TxStatePartitionStorage storage1 = tableStorage.getOrCreatePartitionStorage(1);
+ TxStatePartitionStorage storage0 = txStateStorage.getOrCreatePartitionStorage(0);
+ TxStatePartitionStorage storage1 = txStateStorage.getOrCreatePartitionStorage(1);
UUID txId0 = UUID.randomUUID();
storage0.putForRebalance(txId0, new TxMeta(TxState.COMMITTED, generateEnlistedPartitions(1), generateTimestamp(txId0)));
@@ -319,7 +321,7 @@ public abstract class AbstractTxStatePartitionStorageTest extends BaseIgniteAbst
@Test
public void scansInOrderDefinedByTxIds() {
- TxStatePartitionStorage partitionStorage = tableStorage.getOrCreatePartitionStorage(0);
+ TxStatePartitionStorage partitionStorage = txStateStorage.getOrCreatePartitionStorage(0);
for (int i = 0; i < 100; i++) {
putRandomTxMetaWithCommandIndex(partitionStorage, i, i);
@@ -340,7 +342,7 @@ public abstract class AbstractTxStatePartitionStorageTest extends BaseIgniteAbst
@Test
public void scanOnlySeesDataExistingAtTheMomentOfCreation() {
- TxStatePartitionStorage partitionStorage = tableStorage.getOrCreatePartitionStorage(0);
+ TxStatePartitionStorage partitionStorage = txStateStorage.getOrCreatePartitionStorage(0);
UUID existingBeforeScan = new UUID(2, 0);
partitionStorage.putForRebalance(existingBeforeScan, randomTxMeta(1, existingBeforeScan));
@@ -361,7 +363,7 @@ public abstract class AbstractTxStatePartitionStorageTest extends BaseIgniteAbst
@Test
void lastAppliedIndexGetterIsConsistentWithSetter() {
- TxStatePartitionStorage partitionStorage = tableStorage.getOrCreatePartitionStorage(0);
+ TxStatePartitionStorage partitionStorage = txStateStorage.getOrCreatePartitionStorage(0);
partitionStorage.lastApplied(10, 2);
@@ -370,7 +372,7 @@ public abstract class AbstractTxStatePartitionStorageTest extends BaseIgniteAbst
@Test
void compareAndSetMakesLastAppliedChangeVisible() {
- TxStatePartitionStorage partitionStorage = tableStorage.getOrCreatePartitionStorage(0);
+ TxStatePartitionStorage partitionStorage = txStateStorage.getOrCreatePartitionStorage(0);
UUID txId = UUID.randomUUID();
partitionStorage.compareAndSet(txId, null, randomTxMeta(1, txId), 10, 2);
@@ -381,7 +383,7 @@ public abstract class AbstractTxStatePartitionStorageTest extends BaseIgniteAbst
@Test
public void testSuccessRebalance() {
- TxStatePartitionStorage storage = tableStorage.getOrCreatePartitionStorage(0);
+ TxStatePartitionStorage storage = txStateStorage.getOrCreatePartitionStorage(0);
MvPartitionMeta partitionMeta = saneMvPartitionMeta(30, 50);
@@ -416,7 +418,7 @@ public abstract class AbstractTxStatePartitionStorageTest extends BaseIgniteAbst
@Test
public void testFailRebalance() {
- TxStatePartitionStorage storage = tableStorage.getOrCreatePartitionStorage(0);
+ TxStatePartitionStorage storage = txStateStorage.getOrCreatePartitionStorage(0);
// Nothing will happen because rebalance has not started.
assertThat(storage.abortRebalance(), willCompleteSuccessfully());
@@ -449,8 +451,8 @@ public abstract class AbstractTxStatePartitionStorageTest extends BaseIgniteAbst
@Test
public void testStartRebalanceForClosedOrDestroyedPartition() {
- TxStatePartitionStorage storage0 = tableStorage.getOrCreatePartitionStorage(0);
- TxStatePartitionStorage storage1 = tableStorage.getOrCreatePartitionStorage(1);
+ TxStatePartitionStorage storage0 = txStateStorage.getOrCreatePartitionStorage(0);
+ TxStatePartitionStorage storage1 = txStateStorage.getOrCreatePartitionStorage(1);
storage0.close();
storage1.destroy();
@@ -461,7 +463,7 @@ public abstract class AbstractTxStatePartitionStorageTest extends BaseIgniteAbst
@Test
void testClear() {
- TxStatePartitionStorage storage = tableStorage.getOrCreatePartitionStorage(0);
+ TxStatePartitionStorage storage = txStateStorage.getOrCreatePartitionStorage(0);
// Cleaning up on empty storage should not generate errors.
assertThat(storage.clear(), willCompleteSuccessfully());
@@ -485,9 +487,9 @@ public abstract class AbstractTxStatePartitionStorageTest extends BaseIgniteAbst
@Test
void testCleanOnClosedDestroyedAndRebalancedStorages() {
- TxStatePartitionStorage storage0 = tableStorage.getOrCreatePartitionStorage(0);
- TxStatePartitionStorage storage1 = tableStorage.getOrCreatePartitionStorage(1);
- TxStatePartitionStorage storage2 = tableStorage.getOrCreatePartitionStorage(2);
+ TxStatePartitionStorage storage0 = txStateStorage.getOrCreatePartitionStorage(0);
+ TxStatePartitionStorage storage1 = txStateStorage.getOrCreatePartitionStorage(1);
+ TxStatePartitionStorage storage2 = txStateStorage.getOrCreatePartitionStorage(2);
storage0.close();
storage1.destroy();
@@ -504,7 +506,7 @@ public abstract class AbstractTxStatePartitionStorageTest extends BaseIgniteAbst
@Test
void testSetAndReadMeta() {
- TxStatePartitionStorage storage = tableStorage.getOrCreatePartitionStorage(0);
+ TxStatePartitionStorage storage = txStateStorage.getOrCreatePartitionStorage(0);
storage.lastApplied(1, 15);
@@ -530,7 +532,7 @@ public abstract class AbstractTxStatePartitionStorageTest extends BaseIgniteAbst
@Test
void testCloseStartedRebalance() {
- TxStatePartitionStorage storage = tableStorage.getOrCreatePartitionStorage(0);
+ TxStatePartitionStorage storage = txStateStorage.getOrCreatePartitionStorage(0);
assertThat(storage.startRebalance(), willCompleteSuccessfully());
@@ -541,7 +543,7 @@ public abstract class AbstractTxStatePartitionStorageTest extends BaseIgniteAbst
@Test
void testDestroyStartedRebalance() {
- TxStatePartitionStorage storage = tableStorage.getOrCreatePartitionStorage(0);
+ TxStatePartitionStorage storage = txStateStorage.getOrCreatePartitionStorage(0);
assertThat(storage.startRebalance(), willCompleteSuccessfully());
diff --git a/modules/transactions/src/testFixtures/java/org/apache/ignite/internal/tx/test/ItTransactionTestUtils.java b/modules/transactions/src/testFixtures/java/org/apache/ignite/internal/tx/test/ItTransactionTestUtils.java
index 32758ecc714..a16ea498f51 100644
--- a/modules/transactions/src/testFixtures/java/org/apache/ignite/internal/tx/test/ItTransactionTestUtils.java
+++ b/modules/transactions/src/testFixtures/java/org/apache/ignite/internal/tx/test/ItTransactionTestUtils.java
@@ -21,7 +21,6 @@ import static java.util.Objects.requireNonNull;
import static java.util.concurrent.TimeUnit.SECONDS;
import static java.util.stream.Collectors.toSet;
import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.stablePartAssignmentsKey;
-import static org.apache.ignite.internal.lang.IgniteSystemProperties.colocationEnabled;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -42,8 +41,6 @@ import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.partitiondistribution.Assignments;
import org.apache.ignite.internal.placementdriver.ReplicaMeta;
import org.apache.ignite.internal.replicator.PartitionGroupId;
-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.schema.BinaryRowEx;
import org.apache.ignite.internal.table.RecordBinaryViewImpl;
@@ -139,9 +136,7 @@ public class ItTransactionTestUtils {
int partId = partitionIdForTuple(node, tableName, t, tx);
partitionIds.add(partId);
- PartitionGroupId grpId = colocationEnabled()
- ? new ZonePartitionId(zoneId(node, tableName), partId)
- : new TablePartitionId(tableId(node, tableName), partId);
+ ZonePartitionId grpId = new ZonePartitionId(zoneId(node, tableName), partId);
if (primary) {
ReplicaMeta replicaMeta = waitAndGetPrimaryReplica(node, grpId);
@@ -220,7 +215,7 @@ public class ItTransactionTestUtils {
* @param replicationGrpId Replication group.
* @return Primary replica meta.
*/
- public static ReplicaMeta waitAndGetPrimaryReplica(IgniteImpl node, ReplicationGroupId replicationGrpId) {
+ public static ReplicaMeta waitAndGetPrimaryReplica(IgniteImpl node, ZonePartitionId replicationGrpId) {
CompletableFuture<ReplicaMeta> primaryReplicaFut = node.placementDriver().awaitPrimaryReplica(
replicationGrpId,
node.clock().now(),