You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vp...@apache.org on 2023/12/12 13:29:35 UTC
(ignite-3) branch main updated: IGNITE-20994 Adjust writeIntentResolution logic in order to initiate the recovery if coordinator is dead (#2918)
This is an automated email from the ASF dual-hosted git repository.
vpyatkov 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 64d3bcb345 IGNITE-20994 Adjust writeIntentResolution logic in order to initiate the recovery if coordinator is dead (#2918)
64d3bcb345 is described below
commit 64d3bcb345a20e1dbcc554e40d8e709111b02110
Author: Denis Chudov <mo...@gmail.com>
AuthorDate: Tue Dec 12 16:29:30 2023 +0300
IGNITE-20994 Adjust writeIntentResolution logic in order to initiate the recovery if coordinator is dead (#2918)
---
.../apache/ignite/client/fakes/FakeTxManager.java | 2 +-
...xDistributedTestSingleNodeNoCleanupMessage.java | 7 +-
.../internal/table/distributed/TableManager.java | 15 +-
.../table/distributed/TableMessageGroup.java | 4 +
.../command/MarkLocksReleasedCommand.java} | 12 +-
.../table/distributed/raft/PartitionListener.java | 71 +++++++--
.../distributed/replicator/LeaderOrTxState.java | 56 -------
.../replicator/PartitionReplicaListener.java | 163 ++++++++++++++-------
.../replicator/TransactionStateResolver.java | 126 ++++++++--------
.../PartitionReplicaListenerDurableUnlockTest.java | 39 ++++-
.../PartitionReplicaListenerIndexLockingTest.java | 4 +-
.../replication/PartitionReplicaListenerTest.java | 32 ++--
.../apache/ignite/distributed/ItTxTestCluster.java | 19 +--
.../ignite/internal/table/TxAbstractTest.java | 5 +-
.../table/impl/DummyInternalTableImpl.java | 4 +-
.../org/apache/ignite/internal/tx/TxManager.java | 3 +-
.../org/apache/ignite/internal/tx/TxState.java | 24 +++
.../ignite/internal/tx/impl/OrphanDetector.java | 18 +--
.../internal/tx/impl/ReadWriteTransactionImpl.java | 3 +
.../ignite/internal/tx/impl/TxManagerImpl.java | 8 +-
.../tx/impl/VolatileTxStateMetaStorage.java | 6 +-
.../tx/message/TxStateCommitPartitionRequest.java | 4 +-
.../org/apache/ignite/internal/tx/TxStateTest.java | 8 +-
23 files changed, 371 insertions(+), 262 deletions(-)
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 dce43dd8d2..cdb5501380 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
@@ -146,7 +146,7 @@ public class FakeTxManager implements TxManager {
}
@Override
- public TxStateMeta updateTxMeta(UUID txId, Function<TxStateMeta, TxStateMeta> updater) {
+ public <T extends TxStateMeta> T updateTxMeta(UUID txId, Function<TxStateMeta, TxStateMeta> updater) {
return null;
}
diff --git a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNodeNoCleanupMessage.java b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNodeNoCleanupMessage.java
index 97201815c5..4a33e063f2 100644
--- a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNodeNoCleanupMessage.java
+++ b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNodeNoCleanupMessage.java
@@ -61,6 +61,7 @@ import org.apache.ignite.internal.tx.storage.state.TxStateStorage;
import org.apache.ignite.internal.util.Lazy;
import org.apache.ignite.internal.util.PendingComparableValuesTracker;
import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterNodeResolver;
import org.apache.ignite.network.ClusterService;
import org.apache.ignite.table.Tuple;
import org.apache.ignite.tx.TransactionException;
@@ -150,7 +151,8 @@ public class ItTxDistributedTestSingleNodeNoCleanupMessage extends ItTxDistribut
ClusterNode localNode,
SchemaSyncService schemaSyncService,
CatalogService catalogService,
- PlacementDriver placementDriver
+ PlacementDriver placementDriver,
+ ClusterNodeResolver clusterNodeResolver
) {
return new PartitionReplicaListener(
mvDataStorage,
@@ -172,7 +174,8 @@ public class ItTxDistributedTestSingleNodeNoCleanupMessage extends ItTxDistribut
localNode,
schemaSyncService,
catalogService,
- placementDriver
+ placementDriver,
+ clusterNodeResolver
) {
@Override
public CompletableFuture<ReplicaResult> invoke(ReplicaRequest request, String senderId) {
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 b5195f854b..fe0a3647b8 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
@@ -25,7 +25,6 @@ import static java.util.concurrent.CompletableFuture.anyOf;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.concurrent.CompletableFuture.failedFuture;
import static java.util.concurrent.CompletableFuture.runAsync;
-import static java.util.stream.Collectors.toList;
import static org.apache.ignite.internal.causality.IncrementalVersionedValue.dependingOn;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.partitionAssignments;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.tableAssignments;
@@ -416,7 +415,8 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
txManager,
clock,
topologyService,
- clusterService.messagingService()
+ clusterService.messagingService(),
+ placementDriver
);
schemaVersions = new SchemaVersionsImpl(schemaSyncService, catalogService, clock);
@@ -654,9 +654,6 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
TablePartitionId replicaGrpId = new TablePartitionId(tableId, partId);
- transactionStateResolver.updateAssignment(replicaGrpId, newConfiguration.peers().stream().map(Peer::consistentId)
- .collect(toList()));
-
var safeTimeTracker = new PendingComparableValuesTracker<HybridTimestamp, Void>(
new HybridTimestamp(1, 0)
);
@@ -887,7 +884,8 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
localNode(),
schemaSyncService,
catalogService,
- placementDriver
+ placementDriver,
+ clusterService.topologyService()
);
}
@@ -1663,11 +1661,6 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
Set<Assignment> stableAssignments = ByteUtils.fromBytes(stableAssignmentsEntry.value());
- transactionStateResolver.updateAssignment(
- replicaGrpId,
- stableAssignments.stream().filter(Assignment::isPeer).map(Assignment::consistentId).collect(toList())
- );
-
// Start a new Raft node and Replica if this node has appeared in the new assignments.
boolean shouldStartLocalServices = pendingAssignments.stream()
.filter(assignment -> localMember.name().equals(assignment.consistentId()))
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableMessageGroup.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableMessageGroup.java
index 56eeaa3e8d..43792511c8 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableMessageGroup.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableMessageGroup.java
@@ -21,6 +21,7 @@ import static org.apache.ignite.internal.table.distributed.TableMessageGroup.GRO
import org.apache.ignite.internal.table.distributed.command.BuildIndexCommand;
import org.apache.ignite.internal.table.distributed.command.FinishTxCommand;
+import org.apache.ignite.internal.table.distributed.command.MarkLocksReleasedCommand;
import org.apache.ignite.internal.table.distributed.command.TablePartitionIdMessage;
import org.apache.ignite.internal.table.distributed.command.TimedBinaryRowMessage;
import org.apache.ignite.internal.table.distributed.command.UpdateAllCommand;
@@ -206,5 +207,8 @@ public interface TableMessageGroup {
/** Message type for {@link TablePartitionIdMessage}. */
short TABLE_PARTITION_ID = 61;
+
+ /** Message type for {@link MarkLocksReleasedCommand}. */
+ short MARK_LOCKS_RELEASED = 62;
}
}
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxStateCommitPartitionRequest.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/MarkLocksReleasedCommand.java
similarity index 71%
copy from modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxStateCommitPartitionRequest.java
copy to modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/MarkLocksReleasedCommand.java
index 76ed5803ad..1f17e51b8c 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxStateCommitPartitionRequest.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/MarkLocksReleasedCommand.java
@@ -15,16 +15,14 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.tx.message;
+package org.apache.ignite.internal.table.distributed.command;
-import java.util.UUID;
-import org.apache.ignite.internal.replicator.message.ReplicaRequest;
+import org.apache.ignite.internal.table.distributed.TableMessageGroup.Commands;
import org.apache.ignite.network.annotations.Transferable;
/**
- * Transaction state request.
+ * Command that is used to replicate the locks released update for tx state storage.
*/
-@Transferable(TxMessageGroup.TX_STATE_COMMIT_PARTITION_REQUEST)
-public interface TxStateCommitPartitionRequest extends ReplicaRequest {
- UUID txId();
+@Transferable(Commands.MARK_LOCKS_RELEASED)
+public interface MarkLocksReleasedCommand extends PartitionCommand {
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java
index 40468ba0d3..54aaff0021 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java
@@ -24,6 +24,7 @@ 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;
import static org.apache.ignite.internal.tx.TxState.PENDING;
+import static org.apache.ignite.internal.tx.TxState.isFinalState;
import static org.apache.ignite.internal.util.CollectionUtils.last;
import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_UNEXPECTED_STATE_ERR;
@@ -61,6 +62,7 @@ import org.apache.ignite.internal.storage.RowId;
import org.apache.ignite.internal.table.distributed.StorageUpdateHandler;
import org.apache.ignite.internal.table.distributed.command.BuildIndexCommand;
import org.apache.ignite.internal.table.distributed.command.FinishTxCommand;
+import org.apache.ignite.internal.table.distributed.command.MarkLocksReleasedCommand;
import org.apache.ignite.internal.table.distributed.command.TablePartitionIdMessage;
import org.apache.ignite.internal.table.distributed.command.UpdateAllCommand;
import org.apache.ignite.internal.table.distributed.command.UpdateCommand;
@@ -205,6 +207,8 @@ public class PartitionListener implements RaftGroupListener, BeforeApplyHandler
handleSafeTimeSyncCommand((SafeTimeSyncCommand) command, commandIndex, commandTerm);
} else if (command instanceof BuildIndexCommand) {
handleBuildIndexCommand((BuildIndexCommand) command, commandIndex, commandTerm);
+ } else if (command instanceof MarkLocksReleasedCommand) {
+ handleMarkLocksReleasedCommand((MarkLocksReleasedCommand) command, commandIndex, commandTerm);
} else {
assert false : "Command was not found [cmd=" + command + ']';
}
@@ -348,21 +352,7 @@ public class PartitionListener implements RaftGroupListener, BeforeApplyHandler
LOG.debug("Finish the transaction txId = {}, state = {}, txStateChangeRes = {}", txId, txMetaToSet, txStateChangeRes);
if (!txStateChangeRes) {
- UUID traceId = UUID.randomUUID();
-
- String errorMsg = format("Fail to finish the transaction txId = {} because of inconsistent state = {},"
- + " expected state = null, state to set = {}",
- txId,
- txMetaBeforeCas,
- txMetaToSet
- );
-
- IgniteInternalException stateChangeException = new IgniteInternalException(traceId, TX_UNEXPECTED_STATE_ERR, errorMsg);
-
- // Exception is explicitly logged because otherwise it can be lost if it did not occur on the leader.
- LOG.error(errorMsg);
-
- throw stateChangeException;
+ onTxStateStorageCasFail(txId, txMetaBeforeCas, txMetaToSet);
}
}
@@ -543,6 +533,57 @@ public class PartitionListener implements RaftGroupListener, BeforeApplyHandler
}
}
+ /**
+ * Handles the {@link MarkLocksReleasedCommand}.
+ *
+ * @param cmd Command.
+ * @param commandIndex Command index.
+ * @param commandTerm Command term.
+ */
+ private void handleMarkLocksReleasedCommand(MarkLocksReleasedCommand cmd, long commandIndex, long commandTerm) {
+ UUID txId = cmd.txId();
+
+ TxMeta txMetaBeforeCas = txStateStorage.get(txId);
+
+ assert txMetaBeforeCas != null && isFinalState(txMetaBeforeCas.txState()) : "Unexpected tx state [txId=" + cmd.txId()
+ + ", state=" + txMetaBeforeCas + "].";
+
+ TxMeta txMetaToSet = new TxMeta(
+ txMetaBeforeCas.txState(),
+ txMetaBeforeCas.enlistedPartitions(),
+ txMetaBeforeCas.commitTimestamp(),
+ true
+ );
+
+ boolean txStateChangeRes = txStateStorage.compareAndSet(
+ txId,
+ null,
+ txMetaToSet,
+ commandIndex,
+ commandTerm
+ );
+
+ if (!txStateChangeRes) {
+ onTxStateStorageCasFail(txId, txMetaBeforeCas, txMetaToSet);
+ }
+ }
+
+ private static void onTxStateStorageCasFail(UUID txId, TxMeta txMetaBeforeCas, TxMeta txMetaToSet) {
+ String errorMsg = format("Failed to update tx state in the storage, transaction txId = {} because of inconsistent state,"
+ + " expected state = {}, state to set = {}",
+ txId,
+ txMetaBeforeCas,
+ txMetaToSet
+ );
+
+ IgniteInternalException stateChangeException = new IgniteInternalException(TX_UNEXPECTED_STATE_ERR, errorMsg);
+
+ // Exception is explicitly logged because otherwise it can be lost if it did not occur on the leader.
+ LOG.error(errorMsg);
+
+ throw stateChangeException;
+ }
+
private static <T extends Comparable<T>> void updateTrackerIgnoringTrackerClosedException(
PendingComparableValuesTracker<T, Void> tracker,
T newValue
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/LeaderOrTxState.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/LeaderOrTxState.java
deleted file mode 100644
index b838156230..0000000000
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/LeaderOrTxState.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.table.distributed.replicator;
-
-import java.io.Serializable;
-import org.apache.ignite.internal.tx.TransactionMeta;
-import org.apache.ignite.internal.tx.message.TxStateCommitPartitionRequest;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * Response for the {@link TxStateCommitPartitionRequest}. Can contain either the consistent ID of the Partition Group leader, which should
- * be queried for the TX Meta, or the TX Meta itself.
- */
-public class LeaderOrTxState implements Serializable {
- private static final long serialVersionUID = -3555591755828355117L;
-
- @Nullable
- private final String leaderName;
-
- @Nullable
- private final TransactionMeta txMeta;
-
- /**
- * Creates a response.
- *
- * @param leaderName Leader consistent ID.
- * @param txMeta TX meta.
- */
- public LeaderOrTxState(@Nullable String leaderName, @Nullable TransactionMeta txMeta) {
- this.leaderName = leaderName;
- this.txMeta = txMeta;
- }
-
- public @Nullable String leaderName() {
- return leaderName;
- }
-
- public @Nullable TransactionMeta txMeta() {
- return txMeta;
- }
-}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
index fa28a201f6..99e806dc6d 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
@@ -28,6 +28,7 @@ import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
import static org.apache.ignite.internal.tx.TxState.ABANDONED;
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.FINISHING;
import static org.apache.ignite.internal.tx.TxState.PENDING;
import static org.apache.ignite.internal.tx.TxState.isFinalState;
import static org.apache.ignite.internal.util.CollectionUtils.nullOrEmpty;
@@ -117,6 +118,7 @@ import org.apache.ignite.internal.table.distributed.TableMessagesFactory;
import org.apache.ignite.internal.table.distributed.TableSchemaAwareIndexStorage;
import org.apache.ignite.internal.table.distributed.command.BuildIndexCommand;
import org.apache.ignite.internal.table.distributed.command.FinishTxCommandBuilder;
+import org.apache.ignite.internal.table.distributed.command.MarkLocksReleasedCommand;
import org.apache.ignite.internal.table.distributed.command.TablePartitionIdMessage;
import org.apache.ignite.internal.table.distributed.command.TimedBinaryRowMessage;
import org.apache.ignite.internal.table.distributed.command.TimedBinaryRowMessageBuilder;
@@ -149,13 +151,13 @@ import org.apache.ignite.internal.tx.Lock;
import org.apache.ignite.internal.tx.LockKey;
import org.apache.ignite.internal.tx.LockManager;
import org.apache.ignite.internal.tx.LockMode;
-import org.apache.ignite.internal.tx.TransactionAbandonedException;
import org.apache.ignite.internal.tx.TransactionIds;
import org.apache.ignite.internal.tx.TransactionMeta;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.internal.tx.TxMeta;
import org.apache.ignite.internal.tx.TxState;
import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.TxStateMetaFinishing;
import org.apache.ignite.internal.tx.message.TxFinishReplicaRequest;
import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
import org.apache.ignite.internal.tx.message.TxStateCommitPartitionRequest;
@@ -172,6 +174,7 @@ import org.apache.ignite.internal.util.PendingComparableValuesTracker;
import org.apache.ignite.internal.util.TrackerClosedException;
import org.apache.ignite.lang.ErrorGroups.Replicator;
import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterNodeResolver;
import org.apache.ignite.tx.TransactionException;
import org.jetbrains.annotations.Nullable;
@@ -266,6 +269,8 @@ public class PartitionReplicaListener implements ReplicaListener {
*/
private final Object commandProcessingLinearizationMutex = new Object();
+ private final ClusterNodeResolver clusterNodeResolver;
+
/**
* The constructor.
*
@@ -307,7 +312,8 @@ public class PartitionReplicaListener implements ReplicaListener {
ClusterNode localNode,
SchemaSyncService schemaSyncService,
CatalogService catalogService,
- PlacementDriver placementDriver
+ PlacementDriver placementDriver,
+ ClusterNodeResolver clusterNodeResolver
) {
this.mvDataStorage = mvDataStorage;
this.raftClient = raftClient;
@@ -326,6 +332,7 @@ public class PartitionReplicaListener implements ReplicaListener {
this.schemaSyncService = schemaSyncService;
this.catalogService = catalogService;
this.placementDriver = placementDriver;
+ this.clusterNodeResolver = clusterNodeResolver;
this.replicationGroupId = new TablePartitionId(tableId, partId);
@@ -389,12 +396,7 @@ public class PartitionReplicaListener implements ReplicaListener {
return txManager.cleanup(enlistedPartitions, commit, commitTimestamp, txId)
.handle((v, e) -> {
if (e == null) {
- return txManager.executeCleanupAsync(() -> markLocksReleased(
- txId,
- txMeta.enlistedPartitions(),
- txMeta.txState(),
- txMeta.commitTimestamp())
- );
+ return txManager.executeCleanupAsync(() -> markLocksReleased(txId));
} else {
LOG.warn("Failed to execute cleanup on commit partition primary replica switch [txId={}, commitPartition={}]",
e, txId, replicationGroupId);
@@ -409,15 +411,16 @@ public class PartitionReplicaListener implements ReplicaListener {
.thenCompose(f -> f);
}
- private void markLocksReleased(
- UUID txId,
- Collection<TablePartitionId> enlistedPartitions,
- TxState txState,
- @Nullable HybridTimestamp commitTimestamp
- ) {
- TxMeta newTxMeta = new TxMeta(txState, enlistedPartitions, commitTimestamp, true);
+ private void markLocksReleased(UUID txId) {
+ reliableCatalogVersionFor(hybridClock.now()).thenAccept(catalogVersion -> {
+ MarkLocksReleasedCommand cmd = MSG_FACTORY.markLocksReleasedCommand()
+ .txId(txId)
+ .safeTimeLong(hybridClock.nowLong())
+ .requiredCatalogVersion(catalogVersion)
+ .build();
- txStateStorage.put(txId, newTxMeta);
+ raftClient.run(cmd);
+ });
}
private CompletableFuture<Boolean> onPrimaryExpired(PrimaryReplicaEventParameters evt, @Nullable Throwable exception) {
@@ -454,11 +457,6 @@ public class PartitionReplicaListener implements ReplicaListener {
@Override
public CompletableFuture<ReplicaResult> invoke(ReplicaRequest request, String senderId) {
- if (request instanceof TxStateCommitPartitionRequest) {
- return processTxStateCommitPartitionRequest((TxStateCommitPartitionRequest) request)
- .thenApply(res -> new ReplicaResult(res, null));
- }
-
return ensureReplicaIsPrimary(request)
.thenCompose(isPrimary -> processRequest(request, isPrimary, senderId))
.thenApply(res -> {
@@ -490,7 +488,7 @@ public class PartitionReplicaListener implements ReplicaListener {
}
if (request instanceof TxRecoveryMessage) {
- return processTxRecoveryAction((TxRecoveryMessage) request);
+ return processTxRecoveryMessage((TxRecoveryMessage) request);
}
HybridTimestamp opTsIfDirectRo = (request instanceof ReadOnlyDirectReplicaRequest) ? hybridClock.now() : null;
@@ -507,7 +505,7 @@ public class PartitionReplicaListener implements ReplicaListener {
* @param request Tx recovery request.
* @return The future is complete when the transaction state is finalized.
*/
- private CompletableFuture<Void> processTxRecoveryAction(TxRecoveryMessage request) {
+ private CompletableFuture<Void> processTxRecoveryMessage(TxRecoveryMessage request) {
UUID txId = request.txId();
TxMeta txMeta = txStateStorage.get(txId);
@@ -521,8 +519,35 @@ public class PartitionReplicaListener implements ReplicaListener {
LOG.info("Orphan transaction has to be aborted [tx={}].", txId);
- // TODO: IGNITE-20735 Implement initiate recovery handling logic.
- return nullCompletedFuture();
+ return triggerTxRecovery(txId)
+ .thenApply(v -> null);
+ }
+
+ /**
+ * Starts tx recovery process. Returns the future containing transaction meta with its final state that completes
+ * when the recovery is completed.
+ *
+ * @param txId Transaction id.
+ * @return Tx recovery future, or failed future if the tx recovery is not possible.
+ */
+ private CompletableFuture<TransactionMeta> triggerTxRecovery(UUID txId) {
+ // TODO: IGNITE-20735 Implement initiate recovery handling logic. This part has to be fully rewritten.
+ TxStateMeta txStateMeta = txManager.stateMeta(txId);
+
+ if (txStateMeta == null || txStateMeta.txState() == ABANDONED) {
+ txStateStorage.put(txId, new TxMeta(ABORTED, List.of(), null));
+ return completedFuture(
+ txManager.updateTxMeta(txId, old -> {
+ if (old == null) {
+ return new TxStateMeta(ABORTED, null, replicationGroupId, null);
+ } else {
+ return new TxStateMeta(ABORTED, old.txCoordinatorId(), replicationGroupId, null);
+ }
+ })
+ );
+ } else {
+ return completedFuture(txStateMeta);
+ }
}
/**
@@ -720,6 +745,8 @@ public class PartitionReplicaListener implements ReplicaListener {
return processReadOnlyDirectSingleEntryAction((ReadOnlyDirectSingleRowReplicaRequest) request, opStartTsIfDirectRo);
} else if (request instanceof ReadOnlyDirectMultiRowReplicaRequest) {
return processReadOnlyDirectMultiEntryAction((ReadOnlyDirectMultiRowReplicaRequest) request, opStartTsIfDirectRo);
+ } else if (request instanceof TxStateCommitPartitionRequest) {
+ return processTxStateCommitPartitionRequest((TxStateCommitPartitionRequest) request);
} else {
throw new UnsupportedReplicaRequestException(request.getClass());
}
@@ -731,27 +758,69 @@ public class PartitionReplicaListener implements ReplicaListener {
* @param request Transaction state request.
* @return Result future.
*/
- private CompletableFuture<LeaderOrTxState> processTxStateCommitPartitionRequest(TxStateCommitPartitionRequest request) {
+ private CompletableFuture<TransactionMeta> processTxStateCommitPartitionRequest(TxStateCommitPartitionRequest request) {
return placementDriver.getPrimaryReplica(replicationGroupId, hybridClock.now())
- .thenCompose(primaryReplicaMeta -> {
- if (primaryReplicaMeta == null) {
+ .thenCompose(replicaMeta -> {
+ if (replicaMeta == null || replicaMeta.getLeaseholder() == null) {
return failedFuture(new PrimaryReplicaMissException(localNode.name(), null, null, null, null));
}
- if (!isLocalPeer(primaryReplicaMeta.getLeaseholder())) {
- return completedFuture(new LeaderOrTxState(primaryReplicaMeta.getLeaseholder(), null));
+ if (!isLocalPeer(replicaMeta.getLeaseholder())) {
+ return failedFuture(
+ new PrimaryReplicaMissException(localNode.name(), replicaMeta.getLeaseholder(), null, null, null)
+ );
}
- TransactionMeta txMeta = txManager.stateMeta(request.txId());
+ UUID txId = request.txId();
- if (txMeta == null) {
- txMeta = txStateStorage.get(request.txId());
- }
+ TxStateMeta txMeta = txManager.stateMeta(txId);
- return completedFuture(new LeaderOrTxState(null, txMeta));
+ if (txMeta != null && txMeta.txState() == FINISHING) {
+ assert txMeta instanceof TxStateMetaFinishing : txMeta;
+ return ((TxStateMetaFinishing) txMeta).txFinishFuture();
+ } else if (txMeta == null || !isFinalState(txMeta.txState())) {
+ // Try to trigger recovery, if needed. If the transaction will be aborted, the proper ABORTED state will be sent
+ // in response.
+ return triggerTxRecoveryOnTxStateResolutionIfNeeded(txId, txMeta);
+ } else {
+ return completedFuture(txMeta);
+ }
});
}
+ /**
+ * Checks whether tx recovery is needed with given tx meta and triggers it of needed.
+ *
+ * @param txId Transaction id.
+ * @param txStateMeta Transaction meta.
+ * @return Tx recovery future, or completed future if the recovery isn't needed, or failed future if the recovery is not possible.
+ */
+ private CompletableFuture<TransactionMeta> triggerTxRecoveryOnTxStateResolutionIfNeeded(UUID txId, @Nullable TxStateMeta txStateMeta) {
+ assert txStateMeta == null || !isFinalState(txStateMeta.txState()) : "Unexpected transaction state: " + txStateMeta;
+
+ TxMeta txMeta = txStateStorage.get(txId);
+
+ if (txMeta == null) {
+ // This means the transaction is pending and we should trigger the recovery if there is no tx coordinator in topology.
+ if (txStateMeta == null
+ || txStateMeta.txState() == ABANDONED
+ || clusterNodeResolver.getById(txStateMeta.txCoordinatorId()) == null) {
+ // This means that primary replica for commit partition has changed, since the local node doesn't have the volatile tx
+ // state; and there is no final tx state in txStateStorage, or the tx coordinator left the cluster. But we can assume
+ // that as the coordinator (or information about it) is missing, there is no need to wait a finish request from
+ // tx coordinator, the transaction can't be committed at all.
+ return triggerTxRecovery(txId);
+ } else {
+ assert txStateMeta != null && txStateMeta.txState() == PENDING : "Unexpected transaction state: " + txStateMeta;
+ return completedFuture(txStateMeta);
+ }
+ } else {
+ // Recovery is not needed.
+ assert isFinalState(txMeta.txState()) : "Unexpected transaction state: " + txMeta;
+ return completedFuture(txMeta);
+ }
+ }
+
/**
* Processes retrieve batch for read only transaction.
*
@@ -1534,13 +1603,7 @@ public class PartitionReplicaListener implements ReplicaListener {
return finishTransaction(enlistedPartitions, txId, commit, commitTimestamp, txCoordinatorId)
.thenCompose(v -> txManager.cleanup(enlistedPartitions, commit, commitTimestamp, txId))
- .thenRun(() ->
- markLocksReleased(
- txId,
- enlistedPartitions,
- commit ? COMMITTED : ABORTED,
- commitTimestamp)
- );
+ .thenRun(() -> markLocksReleased(txId));
}
/**
@@ -3453,23 +3516,23 @@ public class PartitionReplicaListener implements ReplicaListener {
* Check whether we can read write intents created by this transaction.
*
* @param txId Transaction id.
- * @param txMeta Transaction metainfo.
+ * @param txMeta Transaction meta info.
* @param timestamp Read timestamp.
* @return {@code true} if we can read from entries created in this transaction (when the transaction was committed and commit time <=
* read time).
*/
- private static Boolean canReadFromWriteIntent(UUID txId, @Nullable TransactionMeta txMeta, @Nullable HybridTimestamp timestamp) {
- if (txMeta == null || txMeta.txState() == ABANDONED) {
- // TODO https://issues.apache.org/jira/browse/IGNITE-20427 make the null value returned from commit partition
- // TODO more determined
- throw new TransactionAbandonedException(txId, txMeta);
- }
+ private static Boolean canReadFromWriteIntent(UUID txId, TransactionMeta txMeta, @Nullable HybridTimestamp timestamp) {
+ assert isFinalState(txMeta.txState()) || txMeta.txState() == PENDING
+ : format("Unexpected state defined by write intent resolution [txId={}, txMeta={}].", txId, txMeta);
+
if (txMeta.txState() == COMMITTED) {
boolean readLatest = timestamp == null;
return readLatest || txMeta.commitTimestamp().compareTo(timestamp) <= 0;
+ } else {
+ // Either ABORTED or PENDING.
+ return false;
}
- return false;
}
/**
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 73c92af2a5..683cf1c5e5 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
@@ -23,25 +23,26 @@ import static org.apache.ignite.internal.tx.TxState.FINISHING;
import static org.apache.ignite.internal.tx.TxState.PENDING;
import static org.apache.ignite.internal.tx.TxState.isFinalState;
-import java.util.Collection;
-import java.util.LinkedHashSet;
import java.util.Map;
-import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Supplier;
import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.hlc.HybridTimestamp;
-import org.apache.ignite.internal.lang.IgniteInternalException;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
import org.apache.ignite.internal.replicator.ReplicaService;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.replicator.exception.PrimaryReplicaMissException;
import org.apache.ignite.internal.tx.TransactionMeta;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.internal.tx.TxStateMeta;
import org.apache.ignite.internal.tx.TxStateMetaFinishing;
+import org.apache.ignite.internal.tx.impl.PlacementDriverHelper;
import org.apache.ignite.internal.tx.message.TxMessageGroup;
import org.apache.ignite.internal.tx.message.TxMessagesFactory;
import org.apache.ignite.internal.tx.message.TxStateCommitPartitionRequest;
+import org.apache.ignite.internal.tx.message.TxStateCommitPartitionRequestBuilder;
import org.apache.ignite.internal.tx.message.TxStateCoordinatorRequest;
import org.apache.ignite.internal.tx.message.TxStateResponse;
import org.apache.ignite.network.ClusterNode;
@@ -60,9 +61,6 @@ public class TransactionStateResolver {
/** Network timeout. */
private static final long RPC_TIMEOUT = 3000;
- /** Assignment node names per replication group. */
- private final Map<ReplicationGroupId, LinkedHashSet<String>> primaryReplicaMapping = new ConcurrentHashMap<>();
-
/** Replication service. */
private final ReplicaService replicaService;
@@ -71,6 +69,8 @@ public class TransactionStateResolver {
/** Function that resolves a node consistent ID to a cluster node. */
private final ClusterNodeResolver clusterNodeResolver;
+ private final PlacementDriverHelper placementDriverHelper;
+
private final Map<UUID, CompletableFuture<TransactionMeta>> txStateFutures = new ConcurrentHashMap<>();
private final TxManager txManager;
@@ -87,19 +87,22 @@ public class TransactionStateResolver {
* @param clock Node clock.
* @param clusterNodeResolver Cluster node resolver.
* @param messagingService Messaging service.
+ * @param placementDriver Placement driver.
*/
public TransactionStateResolver(
ReplicaService replicaService,
TxManager txManager,
HybridClock clock,
ClusterNodeResolver clusterNodeResolver,
- MessagingService messagingService
+ MessagingService messagingService,
+ PlacementDriver placementDriver
) {
this.replicaService = replicaService;
this.txManager = txManager;
this.clock = clock;
this.clusterNodeResolver = clusterNodeResolver;
this.messagingService = messagingService;
+ this.placementDriverHelper = new PlacementDriverHelper(placementDriver, clock);
}
/**
@@ -133,7 +136,7 @@ public class TransactionStateResolver {
*/
public CompletableFuture<TransactionMeta> resolveTxState(
UUID txId,
- ReplicationGroupId commitGrpId,
+ TablePartitionId commitGrpId,
@Nullable HybridTimestamp timestamp
) {
TxStateMeta localMeta = txManager.stateMeta(txId);
@@ -169,7 +172,7 @@ public class TransactionStateResolver {
private void resolveDistributiveTxState(
UUID txId,
@Nullable TxStateMeta localMeta,
- ReplicationGroupId commitGrpId,
+ TablePartitionId commitGrpId,
@Nullable HybridTimestamp timestamp,
CompletableFuture<TransactionMeta> txMetaFuture
) {
@@ -195,14 +198,15 @@ public class TransactionStateResolver {
} else {
assert localMeta.txState() == ABANDONED : "Unexpected transaction state [txId=" + txId + ", txStateMeta=" + localMeta + ']';
- txMetaFuture.complete(localMeta);
+ // Still try to resolve the state from commit partition.
+ resolveTxStateFromCommitPartition(txId, commitGrpId, txMetaFuture);
}
}
private void resolveTxStateFromTxCoordinator(
UUID txId,
String coordinatorId,
- ReplicationGroupId commitGrpId,
+ TablePartitionId commitGrpId,
HybridTimestamp timestamp,
CompletableFuture<TransactionMeta> txMetaFuture
) {
@@ -212,18 +216,17 @@ public class TransactionStateResolver {
if (coordinator == null) {
// This means the coordinator node have either left the cluster or restarted.
+ markAbandoned(txId);
+
resolveTxStateFromCommitPartition(txId, commitGrpId, txMetaFuture);
} else {
CompletableFuture<TransactionMeta> coordinatorTxMetaFuture = new CompletableFuture<>();
coordinatorTxMetaFuture.whenComplete((v, e) -> {
- assert v != null : "Unexpected result from transaction coordinator: unknown transaction state [txId=" + txId
- + ", transactionCoordinator=" + coordinator + ']';
-
if (e == null) {
txMetaFuture.complete(v);
} else {
- txMetaFuture.completeExceptionally(e);
+ resolveTxStateFromCommitPartition(txId, commitGrpId, txMetaFuture);
}
});
@@ -232,23 +235,31 @@ public class TransactionStateResolver {
.txId(txId)
.build();
- sendAndRetry(coordinatorTxMetaFuture, coordinator, request);
+ send(coordinatorTxMetaFuture, coordinator, request);
}
}
private void resolveTxStateFromCommitPartition(
UUID txId,
- ReplicationGroupId commitGrpId,
+ TablePartitionId commitGrpId,
CompletableFuture<TransactionMeta> txMetaFuture
) {
- TxStateCommitPartitionRequest request = FACTORY.txStateCommitPartitionRequest()
+ Supplier<TxStateCommitPartitionRequestBuilder> factory = () -> FACTORY.txStateCommitPartitionRequest()
.groupId(commitGrpId)
- .txId(txId)
- .build();
+ .txId(txId);
updateLocalTxMapAfterDistributedStateResolved(txId, txMetaFuture);
- sendAndRetry(txMetaFuture, commitGrpId, request);
+ sendAndRetry(txMetaFuture, commitGrpId, factory);
+ }
+
+ /**
+ * Marks the transaction as abandoned due to the absence of coordinator.
+ *
+ * @param txId Transaction id.
+ */
+ private void markAbandoned(UUID txId) {
+ txManager.updateTxMeta(txId, TxStateMeta::abandoned);
}
private void updateLocalTxMapAfterDistributedStateResolved(UUID txId, CompletableFuture<TransactionMeta> future) {
@@ -259,55 +270,42 @@ public class TransactionStateResolver {
});
}
- /**
- * Updates an assignment for the specific replication group.
- *
- * @param replicaGrpId Replication group id.
- * @param nodeNames Assignment node names.
- */
- public void updateAssignment(ReplicationGroupId replicaGrpId, Collection<String> nodeNames) {
- primaryReplicaMapping.put(replicaGrpId, new LinkedHashSet<>(nodeNames));
- }
-
/**
* Tries to send a request to primary replica of the replication group.
- * If the first node turns up not a primary one the logic sends the same request to a new primary node.
*
* @param resFut Response future.
* @param replicaGrp Replication group id.
- * @param request Request.
+ * @param requestBuilderFactory Request builder factory.
*/
private void sendAndRetry(
CompletableFuture<TransactionMeta> resFut,
- ReplicationGroupId replicaGrp,
- TxStateCommitPartitionRequest request
+ TablePartitionId replicaGrp,
+ Supplier<TxStateCommitPartitionRequestBuilder> requestBuilderFactory
) {
- ClusterNode nodeToSend = primaryReplicaMapping.get(replicaGrp).stream()
- .map(clusterNodeResolver::getByConsistentId)
- .filter(Objects::nonNull)
- .findFirst()
- .orElseThrow(() -> new IgniteInternalException("All replica nodes are unavailable"));
-
- replicaService.invoke(nodeToSend, request).thenAccept(resp -> {
- assert resp instanceof LeaderOrTxState : "Unsupported response type [type=" + resp.getClass().getSimpleName() + ']';
-
- LeaderOrTxState stateOrLeader = (LeaderOrTxState) resp;
-
- String nextNodeToSend = stateOrLeader.leaderName();
-
- if (nextNodeToSend == null) {
- resFut.complete(stateOrLeader.txMeta());
- } else {
- LinkedHashSet<String> newAssignment = new LinkedHashSet<>();
-
- newAssignment.add(nextNodeToSend);
- newAssignment.addAll(primaryReplicaMapping.get(replicaGrp));
-
- primaryReplicaMapping.put(replicaGrp, newAssignment);
-
- sendAndRetry(resFut, replicaGrp, request);
- }
- });
+ placementDriverHelper.awaitPrimaryReplicaWithExceptionHandling(replicaGrp)
+ .thenCompose(replicaMeta -> {
+ ClusterNode nodeToSend = clusterNodeResolver.getByConsistentId(replicaMeta.getLeaseholder());
+
+ TxStateCommitPartitionRequest request = requestBuilderFactory.get()
+ .enlistmentConsistencyToken(replicaMeta.getStartTime().longValue())
+ .build();
+
+ return replicaService.invoke(nodeToSend, request);
+ })
+ .whenComplete((resp, e) -> {
+ if (e == null) {
+ assert resp instanceof TransactionMeta : "Unsupported response type [type=" + resp.getClass().getSimpleName() + ']';
+
+ TransactionMeta txMeta = (TransactionMeta) resp;
+ resFut.complete(txMeta);
+ } else {
+ if (e instanceof PrimaryReplicaMissException) {
+ sendAndRetry(resFut, replicaGrp, requestBuilderFactory);
+ } else {
+ resFut.completeExceptionally(e);
+ }
+ }
+ });
}
/**
@@ -317,7 +315,7 @@ public class TransactionStateResolver {
* @param node Node to send to.
* @param request Request.
*/
- private void sendAndRetry(CompletableFuture<TransactionMeta> resFut, ClusterNode node, TxStateCoordinatorRequest request) {
+ private void send(CompletableFuture<TransactionMeta> resFut, ClusterNode node, TxStateCoordinatorRequest request) {
messagingService.invoke(node, request, RPC_TIMEOUT).thenAccept(resp -> {
assert resp instanceof TxStateResponse : "Unsupported response type [type=" + resp.getClass().getSimpleName() + ']';
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerDurableUnlockTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerDurableUnlockTest.java
index 1f0aff0121..d0b6afe400 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerDurableUnlockTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerDurableUnlockTest.java
@@ -29,8 +29,10 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
import java.util.Collection;
import java.util.List;
@@ -52,6 +54,7 @@ import org.apache.ignite.internal.raft.service.RaftGroupService;
import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.storage.impl.TestMvPartitionStorage;
import org.apache.ignite.internal.table.distributed.StorageUpdateHandler;
+import org.apache.ignite.internal.table.distributed.command.MarkLocksReleasedCommand;
import org.apache.ignite.internal.table.distributed.replicator.PartitionReplicaListener;
import org.apache.ignite.internal.table.distributed.replicator.TransactionStateResolver;
import org.apache.ignite.internal.table.distributed.schema.SchemaSyncService;
@@ -67,6 +70,7 @@ import org.apache.ignite.internal.util.Lazy;
import org.apache.ignite.internal.util.PendingComparableValuesTracker;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.ClusterNodeImpl;
+import org.apache.ignite.network.ClusterNodeResolver;
import org.apache.ignite.network.NetworkAddress;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -96,6 +100,15 @@ public class PartitionReplicaListenerDurableUnlockTest extends IgniteAbstractTes
@Mock
private TxManager txManager;
+ @Mock
+ private SchemaSyncService schemaSyncService;
+
+ @Mock
+ private CatalogService catalogService;
+
+ @Mock
+ private RaftGroupService raftClient;
+
/** Partition replication listener to test. */
private PartitionReplicaListener partitionReplicaListener;
@@ -122,9 +135,26 @@ public class PartitionReplicaListenerDurableUnlockTest extends IgniteAbstractTes
.toArray(size -> new CompletableFuture<?>[size]));
}).when(txManager).cleanup(any(), anyBoolean(), any(), any());
+ doAnswer(invocation -> nullCompletedFuture()).when(schemaSyncService).waitForMetadataCompleteness(any());
+
+ doAnswer(invocation -> 1).when(catalogService).activeCatalogVersion(anyLong());
+
+ when(raftClient.run(any())).thenAnswer(invocation -> {
+ if (invocation.getArgument(0) instanceof MarkLocksReleasedCommand) {
+ MarkLocksReleasedCommand cmd = invocation.getArgument(0);
+
+ TxMeta txMeta = txStateStorage.get(cmd.txId());
+ TxMeta txMetaToSet = new TxMeta(txMeta.txState(), txMeta.enlistedPartitions(), txMeta.commitTimestamp(), true);
+
+ txStateStorage.compareAndSet(cmd.txId(), txMeta.txState(), txMetaToSet, 0, 0);
+ }
+
+ return null;
+ });
+
partitionReplicaListener = new PartitionReplicaListener(
new TestMvPartitionStorage(PART_ID),
- mock(RaftGroupService.class),
+ raftClient,
txManager,
new HeapLockManager(),
Runnable::run,
@@ -140,9 +170,10 @@ public class PartitionReplicaListenerDurableUnlockTest extends IgniteAbstractTes
mock(StorageUpdateHandler.class),
mock(ValidationSchemasSource.class),
LOCAL_NODE,
- mock(SchemaSyncService.class),
- mock(CatalogService.class),
- placementDriver
+ schemaSyncService,
+ catalogService,
+ placementDriver,
+ mock(ClusterNodeResolver.class)
);
}
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
index b42394fb28..6de0400b36 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
@@ -103,6 +103,7 @@ import org.apache.ignite.internal.util.Lazy;
import org.apache.ignite.internal.util.Pair;
import org.apache.ignite.internal.util.PendingComparableValuesTracker;
import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterNodeResolver;
import org.hamcrest.CustomMatcher;
import org.hamcrest.Matcher;
import org.junit.jupiter.api.BeforeAll;
@@ -237,7 +238,8 @@ public class PartitionReplicaListenerIndexLockingTest extends IgniteAbstractTest
localNode,
new AlwaysSyncedSchemaSyncService(),
catalogService,
- new TestPlacementDriver(localNode)
+ new TestPlacementDriver(localNode),
+ mock(ClusterNodeResolver.class)
);
kvMarshaller = new ReflectionMarshallerFactory().create(schemaDescriptor, Integer.class, Integer.class);
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
index c86295cca2..8dc8c9ee28 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
@@ -41,6 +41,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.AdditionalMatchers.gt;
@@ -88,6 +89,7 @@ import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.marshaller.MarshallerException;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
import org.apache.ignite.internal.placementdriver.TestPlacementDriver;
import org.apache.ignite.internal.raft.Command;
import org.apache.ignite.internal.raft.Peer;
@@ -97,6 +99,7 @@ import org.apache.ignite.internal.replicator.ReplicaResult;
import org.apache.ignite.internal.replicator.ReplicaService;
import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.replicator.exception.PrimaryReplicaMissException;
import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.schema.BinaryRowConverter;
import org.apache.ignite.internal.schema.BinaryTuple;
@@ -142,7 +145,6 @@ import org.apache.ignite.internal.table.distributed.replication.request.ReadWrit
import org.apache.ignite.internal.table.distributed.replicator.IncompatibleSchemaAbortException;
import org.apache.ignite.internal.table.distributed.replicator.IncompatibleSchemaException;
import org.apache.ignite.internal.table.distributed.replicator.InternalSchemaVersionMismatchException;
-import org.apache.ignite.internal.table.distributed.replicator.LeaderOrTxState;
import org.apache.ignite.internal.table.distributed.replicator.PartitionReplicaListener;
import org.apache.ignite.internal.table.distributed.replicator.TransactionStateResolver;
import org.apache.ignite.internal.table.distributed.replicator.action.RequestType;
@@ -436,8 +438,6 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
configureTxManager(txManager);
- doAnswer(invocation -> nullCompletedFuture()).when(txManager).executeCleanupAsync(any(Runnable.class));
-
doAnswer(invocation -> {
Object argument = invocation.getArgument(1);
@@ -469,7 +469,8 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
txManager,
clock,
clusterNodeResolver,
- messagingService
+ messagingService,
+ mock(PlacementDriver.class)
);
transactionStateResolver.start();
@@ -498,7 +499,8 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
localNode,
schemaSyncService,
catalogService,
- new TestPlacementDriver(localNode)
+ new TestPlacementDriver(localNode),
+ mock(ClusterNodeResolver.class)
);
kvMarshaller = marshallerFor(schemaDescriptor);
@@ -542,12 +544,12 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
CompletableFuture<ReplicaResult> fut = partitionReplicaListener.invoke(TX_MESSAGES_FACTORY.txStateCommitPartitionRequest()
.groupId(grpId)
.txId(newTxId())
+ .enlistmentConsistencyToken(1L)
.build(), "senderId");
- LeaderOrTxState tuple = (LeaderOrTxState) fut.get(1, TimeUnit.SECONDS).result();
+ TransactionMeta txMeta = (TransactionMeta) fut.get(1, TimeUnit.SECONDS).result();
- assertNull(tuple.leaderName());
- assertNull(tuple.txMeta());
+ assertNull(txMeta);
}
@Test
@@ -561,16 +563,15 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
CompletableFuture<ReplicaResult> fut = partitionReplicaListener.invoke(TX_MESSAGES_FACTORY.txStateCommitPartitionRequest()
.groupId(grpId)
.txId(txId)
+ .enlistmentConsistencyToken(1L)
.build(), localNode.id());
- LeaderOrTxState tuple = (LeaderOrTxState) fut.get(1, TimeUnit.SECONDS).result();
+ TransactionMeta txMeta = (TransactionMeta) fut.get(1, TimeUnit.SECONDS).result();
- TransactionMeta txMeta = tuple.txMeta();
assertNotNull(txMeta);
assertEquals(TxState.COMMITTED, txMeta.txState());
assertNotNull(txMeta.commitTimestamp());
assertTrue(readTimestamp.compareTo(txMeta.commitTimestamp()) > 0);
- assertNull(tuple.leaderName());
}
@Test
@@ -583,10 +584,9 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
.txId(newTxId())
.build(), localNode.id());
- LeaderOrTxState tuple = (LeaderOrTxState) fut.get(1, TimeUnit.SECONDS).result();
-
- assertNull(tuple.txMeta());
- assertNotNull(tuple.leaderName());
+ assertThrows(PrimaryReplicaMissException.class, () -> {
+ fut.get(1, TimeUnit.SECONDS).result();
+ });
}
@Test
@@ -1870,6 +1870,8 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
});
return null;
}).when(txManager).updateTxMeta(any(), any());
+
+ doAnswer(invocation -> nullCompletedFuture()).when(txManager).executeCleanupAsync(any(Runnable.class));
}
private void testWritesAreSuppliedWithRequiredCatalogVersion(RequestType requestType, RwListenerInvocation listenerInvocation) {
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 bd3ea99b3c..b6b433bd91 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
@@ -469,14 +469,11 @@ public class ItTxTestCluster {
txManagers.get(assignment),
clocks.get(assignment),
nodeResolver,
- clusterServices.get(assignment).messagingService()
+ clusterServices.get(assignment).messagingService(),
+ placementDriver
);
transactionStateResolver.start();
- for (int part = 0; part < assignments.size(); part++) {
- transactionStateResolver.updateAssignment(grpIds.get(part), assignments.get(part));
- }
-
int indexId = globalIndexId++;
ColumnsExtractor row2Tuple = BinaryRowConverter.keyExtractor(schemaDescriptor);
@@ -560,7 +557,8 @@ public class ItTxTestCluster {
nodeResolver.getByConsistentId(assignment),
new AlwaysSyncedSchemaSyncService(),
catalogService,
- placementDriver
+ placementDriver,
+ clusterServices.get(assignment).topologyService()
);
replicaManagers.get(assignment).startReplica(
@@ -656,7 +654,8 @@ public class ItTxTestCluster {
ClusterNode localNode,
SchemaSyncService schemaSyncService,
CatalogService catalogService,
- PlacementDriver placementDriver
+ PlacementDriver placementDriver,
+ ClusterNodeResolver clusterNodeResolver
) {
return new PartitionReplicaListener(
mvDataStorage,
@@ -678,7 +677,8 @@ public class ItTxTestCluster {
localNode,
schemaSyncService,
catalogService,
- placementDriver
+ placementDriver,
+ clusterNodeResolver
);
}
@@ -845,7 +845,8 @@ public class ItTxTestCluster {
clientTxManager,
clientClock,
nodeResolver,
- client.messagingService()
+ client.messagingService(),
+ placementDriver
);
clientTxStateResolver.start();
diff --git a/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/TxAbstractTest.java b/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/TxAbstractTest.java
index 4e2040f7b2..3a281c330c 100644
--- a/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/TxAbstractTest.java
+++ b/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/TxAbstractTest.java
@@ -1999,7 +1999,7 @@ public abstract class TxAbstractTest extends IgniteAbstractTest {
Transaction tx = igniteTransactions.begin();
accounts.recordView().delete(tx, makeKey(1));
- // Imitate the restart of the client node, which is a tx coordinator, in order to make its volatile state of unavailable.
+ // Imitate the restart of the client node, which is a tx coordinator, in order to make its volatile state unavailable.
// Now coordinator path of the write intent resolution has no effect, and we should fallback to commit partition path.
UUID txId = ((ReadWriteTransactionImpl) tx).id();
@@ -2015,9 +2015,6 @@ public abstract class TxAbstractTest extends IgniteAbstractTest {
// Read-only.
Transaction readOnlyTx = igniteTransactions.begin(new TransactionOptions().readOnly(true));
assertEquals(100., accounts.recordView().get(readOnlyTx, makeKey(1)).doubleValue("balance"));
-
- // Commit pending tx.
- tx.commit();
}
@Test
diff --git a/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java b/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
index fe6fa1ceae..a078d13eb8 100644
--- a/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
+++ b/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
@@ -99,6 +99,7 @@ import org.apache.ignite.network.AbstractMessagingService;
import org.apache.ignite.network.ChannelType;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.ClusterNodeImpl;
+import org.apache.ignite.network.ClusterNodeResolver;
import org.apache.ignite.network.ClusterService;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.network.NetworkMessage;
@@ -378,7 +379,8 @@ public class DummyInternalTableImpl extends InternalTableImpl {
LOCAL_NODE,
new AlwaysSyncedSchemaSyncService(),
catalogService,
- TEST_PLACEMENT_DRIVER
+ TEST_PLACEMENT_DRIVER,
+ mock(ClusterNodeResolver.class)
);
partitionListener = new PartitionListener(
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 9a6e287120..b1f3e2a246 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
@@ -73,7 +73,8 @@ public interface TxManager extends IgniteComponent {
* @param updater Transaction meta updater.
* @return Updated transaction state.
*/
- TxStateMeta updateTxMeta(UUID txId, Function<TxStateMeta, TxStateMeta> updater);
+ @Nullable
+ <T extends TxStateMeta> T updateTxMeta(UUID txId, Function<TxStateMeta, TxStateMeta> updater);
/**
* Returns lock manager.
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxState.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxState.java
index 4176047e2d..5026af6c34 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxState.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxState.java
@@ -25,10 +25,34 @@ import org.jetbrains.annotations.Nullable;
* Transaction state.
*/
public enum TxState {
+ /**
+ * Active transaction that is in progress.
+ */
PENDING,
+
+ /**
+ * Transaction can be put in this state on a transaction coordinator or a commit partition on a start of finalization process
+ * (for commit partition this is true only in case of recovery, when the commit partition initiates this finalization process)
+ * and the transaction ends up with some final state ({@link #COMMITTED} or {@link #ABORTED}) when receives a tx finish response from
+ * commit partition on a coordinator, or finishes the transaction recovery on a commit partition. This state can be also seen locally
+ * on data nodes if they are colocated with the coordinator or the commit partition.
+ */
FINISHING,
+
+ /**
+ * Aborted (rolled back) transaction.
+ */
ABORTED,
+
+ /**
+ * Committed transaction.
+ */
COMMITTED,
+
+ /**
+ * State that is assigned to a transaction due to absence of coordinator. It is temporary and can be changed to
+ * {@link TxState#COMMITTED} or {@link TxState#ABORTED} after recovery or successful write intent resolution.
+ */
ABANDONED;
private static final boolean[][] TRANSITION_MATRIX = {
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 ebf740228c..664934ea2a 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
@@ -145,7 +145,7 @@ public class OrphanDetector {
private CompletableFuture<Boolean> lockConflictListener(LockEventParameters params, Throwable e) {
try {
- handleLockHolder(params.lockHolderTx());
+ checkTxOrphaned(params.lockHolderTx());
} catch (NodeStoppingException ex) {
return failedFuture(ex);
}
@@ -158,10 +158,10 @@ public class OrphanDetector {
*
* @param txId Transaction id that holds a lock.
*/
- private Void handleLockHolder(UUID txId) throws NodeStoppingException {
+ private void checkTxOrphaned(UUID txId) throws NodeStoppingException {
if (busyLock.enterBusy()) {
try {
- handleLockHolderInternal(txId);
+ checkTxOrphanedInternal(txId);
} finally {
busyLock.leaveBusy();
}
@@ -176,7 +176,7 @@ public class OrphanDetector {
* @param txId Transaction id that holds a lock.
* @return Future to complete.
*/
- private void handleLockHolderInternal(UUID txId) {
+ private void checkTxOrphanedInternal(UUID txId) {
TxStateMeta txState = txLocalStateStorage.state(txId);
// Transaction state for full transactions is not stored in the local map, so it can be null.
@@ -238,7 +238,7 @@ public class OrphanDetector {
}
/**
- * Does a life check for the transaction coordinator.
+ * Performs a life check for the transaction coordinator.
*
* @param txState Transaction state meta.
* @return True when the transaction coordinator is alive, false otherwise.
@@ -257,7 +257,7 @@ public class OrphanDetector {
private boolean isRecoveryNeeded(UUID txId, TxStateMeta txState) {
if (txState == null
|| isFinalState(txState.txState())
- || isTxAbandonedNotLong(txState)) {
+ || isTxAbandonedRecently(txState)) {
return false;
}
@@ -266,7 +266,7 @@ public class OrphanDetector {
TxStateMeta updatedTxState = txLocalStateStorage.updateMeta(txId, txStateMeta -> {
if (txStateMeta != null
&& !isFinalState(txStateMeta.txState())
- && (txStateMeta.txState() != ABANDONED || isTxAbandonedNotLong(txStateMeta))) {
+ && (txStateMeta.txState() != ABANDONED || isTxAbandonedRecently(txStateMeta))) {
return txAbandonedState;
}
@@ -277,12 +277,12 @@ public class OrphanDetector {
}
/**
- * Checks whether the transaction state is recently marked as abandoned or not.
+ * Checks whether the transaction state is marked as abandoned recently (less than {@link #checkTxStateInterval} millis ago).
*
* @param txState Transaction state metadata.
* @return True if the state recently updated to {@link org.apache.ignite.internal.tx.TxState#ABANDONED}.
*/
- private boolean isTxAbandonedNotLong(TxStateMeta txState) {
+ private boolean isTxAbandonedRecently(TxStateMeta txState) {
if (txState.txState() != ABANDONED) {
return false;
}
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 2d64411a78..3b021fd192 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
@@ -133,6 +133,9 @@ public class ReadWriteTransactionImpl extends IgniteAbstractTransactionImpl {
@Override
protected CompletableFuture<Void> finish(boolean commit) {
if (hasTxFinalizationBegun()) {
+ assert finishFuture != null : "Transaction is in final state but there is no finish future [id="
+ + id() + ", state=" + state() + "].";
+
return finishFuture;
}
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 34146bee95..4add674fef 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
@@ -293,8 +293,8 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler {
}
@Override
- public @Nullable TxStateMeta updateTxMeta(UUID txId, Function<TxStateMeta, TxStateMeta> updater) {
- return inBusyLock(busyLock, () -> txStateVolatileStorage.updateMeta(txId, oldMeta -> {
+ public @Nullable <T extends TxStateMeta> T updateTxMeta(UUID txId, Function<TxStateMeta, TxStateMeta> updater) {
+ return txStateVolatileStorage.updateMeta(txId, oldMeta -> {
TxStateMeta newMeta = updater.apply(oldMeta);
if (newMeta == null) {
@@ -304,7 +304,7 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler {
TxState oldState = oldMeta == null ? null : oldMeta.txState();
return checkTransitionCorrectness(oldState, newMeta.txState()) ? newMeta : oldMeta;
- }));
+ });
}
@Override
@@ -349,7 +349,7 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler {
// than all the read timestamps processed before.
// Every concurrent operation will now use a finish future from the finishing state meta and get only final transaction
// state after the transaction is finished.
- TxStateMetaFinishing finishingStateMeta = (TxStateMetaFinishing) updateTxMeta(txId, TxStateMeta::finishing);
+ TxStateMetaFinishing finishingStateMeta = updateTxMeta(txId, TxStateMeta::finishing);
TxContext tuple = txCtxMap.compute(txId, (uuid, tuple0) -> {
if (tuple0 == null) {
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 e5c95221b7..0b65ba0f48 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
@@ -62,7 +62,7 @@ public class VolatileTxStateMetaStorage {
* @param updater Transaction meta updater.
* @return Updated transaction state.
*/
- public TxStateMeta updateMeta(UUID txId, Function<TxStateMeta, TxStateMeta> updater) {
+ public <T extends TxStateMeta> T updateMeta(UUID txId, Function<TxStateMeta, TxStateMeta> updater) {
return inBusyLock(busyLock, () -> updateMetaInternal(txId, updater));
}
@@ -73,8 +73,8 @@ public class VolatileTxStateMetaStorage {
* @param updater Transaction meta updater.
* @return Updated transaction state.
*/
- private @Nullable TxStateMeta updateMetaInternal(UUID txId, Function<TxStateMeta, TxStateMeta> updater) {
- return txStateMap.compute(txId, (k, oldMeta) -> {
+ private @Nullable <T extends TxStateMeta> T updateMetaInternal(UUID txId, Function<TxStateMeta, TxStateMeta> updater) {
+ return (T) txStateMap.compute(txId, (k, oldMeta) -> {
TxStateMeta newMeta = updater.apply(oldMeta);
if (newMeta == null) {
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxStateCommitPartitionRequest.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxStateCommitPartitionRequest.java
index 76ed5803ad..fc628f3195 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxStateCommitPartitionRequest.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxStateCommitPartitionRequest.java
@@ -18,13 +18,13 @@
package org.apache.ignite.internal.tx.message;
import java.util.UUID;
-import org.apache.ignite.internal.replicator.message.ReplicaRequest;
+import org.apache.ignite.internal.replicator.message.PrimaryReplicaRequest;
import org.apache.ignite.network.annotations.Transferable;
/**
* Transaction state request.
*/
@Transferable(TxMessageGroup.TX_STATE_COMMIT_PARTITION_REQUEST)
-public interface TxStateCommitPartitionRequest extends ReplicaRequest {
+public interface TxStateCommitPartitionRequest extends PrimaryReplicaRequest {
UUID txId();
}
diff --git a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/TxStateTest.java b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/TxStateTest.java
index a642f408be..5f5af26836 100644
--- a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/TxStateTest.java
+++ b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/TxStateTest.java
@@ -93,10 +93,10 @@ public class TxStateTest {
assertFalse(TxState.checkTransitionCorrectness(TxState.ABORTED, TxState.PENDING));
assertFalse(TxState.checkTransitionCorrectness(TxState.ABORTED, TxState.FINISHING));
assertFalse(TxState.checkTransitionCorrectness(TxState.ABORTED, TxState.COMMITTED));
+ assertFalse(TxState.checkTransitionCorrectness(TxState.ABORTED, TxState.ABANDONED));
// Allowed.
assertTrue(TxState.checkTransitionCorrectness(TxState.ABORTED, TxState.ABORTED));
- assertFalse(TxState.checkTransitionCorrectness(TxState.ABORTED, TxState.ABANDONED));
}
@Test
@@ -105,10 +105,10 @@ public class TxStateTest {
assertFalse(TxState.checkTransitionCorrectness(TxState.COMMITTED, TxState.PENDING));
assertFalse(TxState.checkTransitionCorrectness(TxState.COMMITTED, TxState.FINISHING));
assertFalse(TxState.checkTransitionCorrectness(TxState.COMMITTED, TxState.ABORTED));
+ assertFalse(TxState.checkTransitionCorrectness(TxState.COMMITTED, TxState.ABANDONED));
// Allowed.
assertTrue(TxState.checkTransitionCorrectness(TxState.COMMITTED, TxState.COMMITTED));
- assertFalse(TxState.checkTransitionCorrectness(TxState.COMMITTED, TxState.ABANDONED));
}
/**
@@ -116,8 +116,10 @@ public class TxStateTest {
*/
@Test
void testTransitionsFromAbandoned() {
- // Allowed.
+ // Not allowed.
assertFalse(TxState.checkTransitionCorrectness(TxState.ABANDONED, TxState.PENDING));
+
+ // Allowed.
assertTrue(TxState.checkTransitionCorrectness(TxState.ABANDONED, TxState.FINISHING));
assertTrue(TxState.checkTransitionCorrectness(TxState.ABANDONED, TxState.ABORTED));
assertTrue(TxState.checkTransitionCorrectness(TxState.ABANDONED, TxState.COMMITTED));