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/19 12:01:37 UTC
(ignite-3) branch main updated: IGNITE-20735 Implement initiate recovery handling logic (#2856)
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 443dfaea1e IGNITE-20735 Implement initiate recovery handling logic (#2856)
443dfaea1e is described below
commit 443dfaea1e51fc5b1eb795a32df315f0c149096f
Author: Cyrill <cy...@gmail.com>
AuthorDate: Tue Dec 19 15:01:31 2023 +0300
IGNITE-20735 Implement initiate recovery handling logic (#2856)
---
.../java/org/apache/ignite/lang/ErrorGroups.java | 8 +-
.../apache/ignite/client/fakes/FakeTxManager.java | 5 +
.../ignite/distributed/ItTxStateLocalMapTest.java | 6 +-
.../table/distributed/raft/PartitionListener.java | 25 ++-
.../raft/UnexpectedTransactionStateException.java | 42 +++++
.../replicator/PartitionReplicaListener.java | 170 ++++++++++++++-------
.../replicator/TransactionStateResolver.java | 4 +-
.../replication/PartitionReplicaListenerTest.java | 26 +++-
.../apache/ignite/distributed/ItTxTestCluster.java | 2 +-
modules/table/tech-notes/finish.puml | 26 ++++
modules/table/tech-notes/recovery-finish.puml | 104 +++++++++++++
.../internal/tx/TransactionAbandonedException.java | 54 -------
.../tx/TransactionAlreadyFinishedException.java | 42 +++++
.../ignite/internal/tx/TransactionResult.java | 51 +++++++
.../org/apache/ignite/internal/tx/TxManager.java | 9 ++
.../org/apache/ignite/internal/tx/TxState.java | 2 +-
.../org/apache/ignite/internal/tx/TxStateMeta.java | 15 +-
.../ignite/internal/tx/TxStateMetaFinishing.java | 2 +-
.../ignite/internal/tx/impl/OrphanDetector.java | 2 +-
.../internal/tx/impl/PlacementDriverHelper.java | 9 +-
.../internal/tx/impl/TxCleanupRequestSender.java | 19 ++-
.../ignite/internal/tx/impl/TxManagerImpl.java | 67 ++++++--
.../ignite/internal/tx/impl/TxMessageSender.java | 2 +-
.../org/apache/ignite/internal/tx/TxStateTest.java | 4 +-
24 files changed, 532 insertions(+), 164 deletions(-)
diff --git a/modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java b/modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java
index 4f8195e6dd..5d2d396d07 100755
--- a/modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java
+++ b/modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java
@@ -346,14 +346,8 @@ public class ErrorGroups {
/** Failure due to an incompatible schema change. */
public static final int TX_INCOMPATIBLE_SCHEMA_ERR = TX_ERR_GROUP.registerErrorCode((short) 12);
- /** Failure due to an abandoned transaction. */
- public static final int TX_ABANDONED_ERR = TX_ERR_GROUP.registerErrorCode((short) 13);
-
/** Failure due to primary replica expiration. */
- public static final int TX_PRIMARY_REPLICA_EXPIRED_ERR = TX_ERR_GROUP.registerErrorCode((short) 14);
-
- /** Coordinator tries to commit a transaction that has already been aborted. */
- public static final int TX_WAS_ABORTED_ERR = TX_ERR_GROUP.registerErrorCode((short) 15);
+ public static final int TX_PRIMARY_REPLICA_EXPIRED_ERR = TX_ERR_GROUP.registerErrorCode((short) 13);
}
/** Replicator error group. */
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 cdb5501380..91d425830b 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
@@ -190,6 +190,11 @@ public class FakeTxManager implements TxManager {
return nullCompletedFuture();
}
+ @Override
+ public CompletableFuture<Void> cleanup(String node, UUID txId) {
+ return nullCompletedFuture();
+ }
+
@Override
public int finished() {
return 0;
diff --git a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxStateLocalMapTest.java b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxStateLocalMapTest.java
index 6c7f69ddc7..93a558f913 100644
--- a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxStateLocalMapTest.java
+++ b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxStateLocalMapTest.java
@@ -25,6 +25,7 @@ import static org.apache.ignite.internal.tx.TxState.PENDING;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.util.List;
+import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
@@ -45,6 +46,7 @@ import org.apache.ignite.internal.type.NativeTypes;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.table.Tuple;
import org.apache.ignite.tx.Transaction;
+import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.TestInfo;
@@ -181,7 +183,7 @@ public class ItTxStateLocalMapTest extends IgniteAbstractTest {
}
return (expected.txState() == meta.get().txState()
- && expected.txCoordinatorId().equals(meta.get().txCoordinatorId())
+ && Objects.equals(expected.txCoordinatorId(), meta.get().txCoordinatorId())
&& checkTimestamps(expected.commitTimestamp(), meta.get().commitTimestamp()));
}, 5_000));
} catch (InterruptedException e) {
@@ -192,7 +194,7 @@ public class ItTxStateLocalMapTest extends IgniteAbstractTest {
}
}
- private boolean checkTimestamps(HybridTimestamp expected, HybridTimestamp actual) {
+ private boolean checkTimestamps(@Nullable HybridTimestamp expected, @Nullable HybridTimestamp actual) {
if (expected == null) {
return actual == null;
} else {
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 54aaff0021..749caf5f19 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
@@ -26,8 +26,8 @@ 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;
+import java.io.Serializable;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collection;
@@ -67,6 +67,7 @@ import org.apache.ignite.internal.table.distributed.command.TablePartitionIdMess
import org.apache.ignite.internal.table.distributed.command.UpdateAllCommand;
import org.apache.ignite.internal.table.distributed.command.UpdateCommand;
import org.apache.ignite.internal.table.distributed.command.WriteIntentSwitchCommand;
+import org.apache.ignite.internal.tx.TransactionResult;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.internal.tx.TxMeta;
import org.apache.ignite.internal.tx.TxState;
@@ -194,13 +195,15 @@ public class PartitionListener implements RaftGroupListener, BeforeApplyHandler
storage.acquirePartitionSnapshotsReadLock();
+ Serializable result = null;
+
try {
if (command instanceof UpdateCommand) {
handleUpdateCommand((UpdateCommand) command, commandIndex, commandTerm);
} else if (command instanceof UpdateAllCommand) {
handleUpdateAllCommand((UpdateAllCommand) command, commandIndex, commandTerm);
} else if (command instanceof FinishTxCommand) {
- handleFinishTxCommand((FinishTxCommand) command, commandIndex, commandTerm);
+ result = handleFinishTxCommand((FinishTxCommand) command, commandIndex, commandTerm);
} else if (command instanceof WriteIntentSwitchCommand) {
handleWriteIntentSwitchCommand((WriteIntentSwitchCommand) command, commandIndex, commandTerm);
} else if (command instanceof SafeTimeSyncCommand) {
@@ -213,7 +216,7 @@ public class PartitionListener implements RaftGroupListener, BeforeApplyHandler
assert false : "Command was not found [cmd=" + command + ']';
}
- clo.result(null);
+ clo.result(result);
} catch (IgniteInternalException e) {
clo.result(e);
} catch (CompletionException e) {
@@ -316,12 +319,14 @@ public class PartitionListener implements RaftGroupListener, BeforeApplyHandler
* @param cmd Command.
* @param commandIndex Index of the RAFT command.
* @param commandTerm Term of the RAFT command.
+ * @return The actually stored transaction state {@link TransactionResult}.
* @throws IgniteInternalException if an exception occurred during a transaction state change.
*/
- private void handleFinishTxCommand(FinishTxCommand cmd, long commandIndex, long commandTerm) throws IgniteInternalException {
+ private TransactionResult handleFinishTxCommand(FinishTxCommand cmd, long commandIndex, long commandTerm)
+ throws IgniteInternalException {
// Skips the write command because the storage has already executed it.
if (commandIndex <= txStateStorage.lastAppliedIndex()) {
- return;
+ return null;
}
UUID txId = cmd.txId();
@@ -354,6 +359,8 @@ public class PartitionListener implements RaftGroupListener, BeforeApplyHandler
if (!txStateChangeRes) {
onTxStateStorageCasFail(txId, txMetaBeforeCas, txMetaToSet);
}
+
+ return new TransactionResult(stateToSet, cmd.commitTimestamp());
}
@@ -557,7 +564,7 @@ public class PartitionListener implements RaftGroupListener, BeforeApplyHandler
boolean txStateChangeRes = txStateStorage.compareAndSet(
txId,
- null,
+ txMetaBeforeCas.txState(),
txMetaToSet,
commandIndex,
commandTerm
@@ -576,7 +583,11 @@ public class PartitionListener implements RaftGroupListener, BeforeApplyHandler
txMetaToSet
);
- IgniteInternalException stateChangeException = new IgniteInternalException(TX_UNEXPECTED_STATE_ERR, errorMsg);
+ IgniteInternalException stateChangeException =
+ new UnexpectedTransactionStateException(
+ errorMsg,
+ new TransactionResult(txMetaBeforeCas.txState(), txMetaBeforeCas.commitTimestamp())
+ );
// Exception is explicitly logged because otherwise it can be lost if it did not occur on the leader.
LOG.error(errorMsg);
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/UnexpectedTransactionStateException.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/UnexpectedTransactionStateException.java
new file mode 100644
index 0000000000..6523bfa852
--- /dev/null
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/UnexpectedTransactionStateException.java
@@ -0,0 +1,42 @@
+/*
+ * 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.raft;
+
+import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_UNEXPECTED_STATE_ERR;
+
+import org.apache.ignite.internal.lang.IgniteInternalException;
+import org.apache.ignite.internal.tx.TransactionResult;
+
+/**
+ * The exception is thrown when a transaction state has already been set.
+ */
+public class UnexpectedTransactionStateException extends IgniteInternalException {
+ private static final long serialVersionUID = 3791709101302280451L;
+
+ /** Stored transaction result. */
+ private final TransactionResult transactionResult;
+
+ public UnexpectedTransactionStateException(String errorMessage, TransactionResult transactionResult) {
+ super(TX_UNEXPECTED_STATE_ERR, errorMessage);
+ this.transactionResult = transactionResult;
+ }
+
+ public TransactionResult transactionResult() {
+ return transactionResult;
+ }
+}
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 a7b06067ec..43123622c3 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
@@ -40,8 +40,9 @@ import static org.apache.ignite.internal.util.IgniteUtils.findAny;
import static org.apache.ignite.internal.util.IgniteUtils.findFirst;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
+import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_COMMIT_ERR;
import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_FAILED_READ_WRITE_OPERATION_ERR;
-import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_WAS_ABORTED_ERR;
+import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_ROLLBACK_ERR;
import static org.apache.ignite.raft.jraft.util.internal.ThrowUtil.hasCause;
import java.nio.ByteBuffer;
@@ -126,6 +127,7 @@ import org.apache.ignite.internal.table.distributed.command.UpdateAllCommand;
import org.apache.ignite.internal.table.distributed.command.UpdateCommand;
import org.apache.ignite.internal.table.distributed.command.UpdateCommandBuilder;
import org.apache.ignite.internal.table.distributed.command.WriteIntentSwitchCommand;
+import org.apache.ignite.internal.table.distributed.raft.UnexpectedTransactionStateException;
import org.apache.ignite.internal.table.distributed.replication.request.BinaryRowMessage;
import org.apache.ignite.internal.table.distributed.replication.request.BinaryTupleMessage;
import org.apache.ignite.internal.table.distributed.replication.request.BuildIndexReplicaRequest;
@@ -147,12 +149,15 @@ import org.apache.ignite.internal.table.distributed.replication.request.ReadWrit
import org.apache.ignite.internal.table.distributed.replicator.action.RequestType;
import org.apache.ignite.internal.table.distributed.schema.SchemaSyncService;
import org.apache.ignite.internal.table.distributed.schema.ValidationSchemasSource;
+import org.apache.ignite.internal.tx.HybridTimestampTracker;
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.TransactionAlreadyFinishedException;
import org.apache.ignite.internal.tx.TransactionIds;
import org.apache.ignite.internal.tx.TransactionMeta;
+import org.apache.ignite.internal.tx.TransactionResult;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.internal.tx.TxMeta;
import org.apache.ignite.internal.tx.TxState;
@@ -488,7 +493,7 @@ public class PartitionReplicaListener implements ReplicaListener {
}
if (request instanceof TxRecoveryMessage) {
- return processTxRecoveryMessage((TxRecoveryMessage) request);
+ return processTxRecoveryMessage((TxRecoveryMessage) request, senderId);
}
HybridTimestamp opTsIfDirectRo = (request instanceof ReadOnlyDirectReplicaRequest) ? hybridClock.now() : null;
@@ -505,49 +510,65 @@ public class PartitionReplicaListener implements ReplicaListener {
* @param request Tx recovery request.
* @return The future is complete when the transaction state is finalized.
*/
- private CompletableFuture<Void> processTxRecoveryMessage(TxRecoveryMessage request) {
+ private CompletableFuture<Void> processTxRecoveryMessage(TxRecoveryMessage request, String senderId) {
UUID txId = request.txId();
TxMeta txMeta = txStateStorage.get(txId);
// Check whether a transaction has already been finished.
- boolean transactionAlreadyFinished = txMeta != null && isFinalState(txMeta.txState());
+ if (txMeta != null && isFinalState(txMeta.txState())) {
+ return recoverFinishedTx(txId, txMeta)
+ // If the sender has sent a recovery message, it failed to handle it on its own,
+ // so sending cleanup to the sender for the transaction we know is finished.
+ .whenComplete((v, ex) -> runCleanupOnNode(txId, senderId));
+ }
- if (transactionAlreadyFinished) {
+ LOG.info("Orphan transaction has to be aborted [tx={}].", txId);
+
+ return triggerTxRecovery(txId, senderId);
+ }
+
+ private CompletableFuture<Void> recoverFinishedTx(UUID txId, TxMeta txMeta) {
+ if (txMeta.locksReleased() || txMeta.enlistedPartitions().isEmpty()) {
+ // Nothing to do if the locks have been released already or there are no enlistedPartitions available.
return nullCompletedFuture();
}
- LOG.info("Orphan transaction has to be aborted [tx={}].", txId);
+ // Otherwise run a cleanup on the known set of partitions.
+ return (CompletableFuture<Void>) durableCleanup(txId, txMeta);
+ }
+
+ /**
+ * Run cleanup on a node.
+ *
+ * @param txId Transaction id.
+ * @param nodeId Node id (inconsistent).
+ */
+ private CompletableFuture<Void> runCleanupOnNode(UUID txId, String nodeId) {
+ // Get node id of the sender to send back cleanup requests.
+ String nodeConsistentId = clusterNodeResolver.getConsistentIdById(nodeId);
- return triggerTxRecovery(txId)
- .thenApply(v -> null);
+ return nodeConsistentId == null ? nullCompletedFuture() : txManager.cleanup(nodeConsistentId, txId);
}
/**
- * Starts tx recovery process. Returns the future containing transaction meta with its final state that completes
- * when the recovery is completed.
+ * Abort the abandoned transaction.
*
* @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);
- }
+ * @param senderId Sender inconsistent id.
+ */
+ private CompletableFuture<Void> triggerTxRecovery(UUID txId, String senderId) {
+ // If the transaction state is pending, then the transaction should be rolled back,
+ // meaning that the state is changed to aborted and a corresponding cleanup request
+ // is sent in a common durable manner to a partition that have initiated recovery.
+ return txManager.finish(
+ new HybridTimestampTracker(),
+ replicationGroupId,
+ false,
+ Map.of(replicationGroupId, 0L), // term is not required for the rollback.
+ txId
+ )
+ .whenComplete((v, ex) -> runCleanupOnNode(txId, senderId));
}
/**
@@ -777,6 +798,7 @@ public class PartitionReplicaListener implements ReplicaListener {
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
@@ -795,8 +817,13 @@ public class PartitionReplicaListener implements ReplicaListener {
* @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;
+ private CompletableFuture<TransactionMeta> triggerTxRecoveryOnTxStateResolutionIfNeeded(
+ UUID txId,
+ @Nullable TxStateMeta txStateMeta
+ ) {
+ // The state is either null or PENDING or ABANDONED, other states have been filtered out previously.
+ assert txStateMeta == null || txStateMeta.txState() == PENDING || txStateMeta.txState() == ABANDONED
+ : "Unexpected transaction state: " + txStateMeta;
TxMeta txMeta = txStateStorage.get(txId);
@@ -804,19 +831,25 @@ public class PartitionReplicaListener implements ReplicaListener {
// 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
+ || txStateMeta.txCoordinatorId() == null
|| 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);
+ return triggerTxRecovery(txId, localNode.id())
+ .handle((v, ex) ->
+ CompletableFuture.<TransactionMeta>completedFuture(txManager.stateMeta(txId)))
+ .thenCompose(v -> v);
} 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);
}
}
@@ -1575,33 +1608,38 @@ public class PartitionReplicaListener implements ReplicaListener {
// The transaction is finished, but the locks are not released.
// If we got here, it means we are retrying the finish request.
// Let's make sure the desired state is valid.
- // Tx logic does not allow to send a rollback over a finished transaction:
// - The Coordinator calls use same tx state over retries, both abort and commit are possible.
- // - Server side recovery (which is not implemented yet) may only change tx state to aborted.
+ // - Server side recovery may only change tx state to aborted.
// - The Coordinator itself should prevent user calls with different proposed state to the one,
- // that was already triggered (e.g. the client side -> txCoordinator.commitAsync(); txCoordinator.rollbackAsync())
- //
- // To sum it up, the possible states that a 'commit' is allowed to see:
+ // that was already triggered (e.g. the client side -> txCoordinator.commitAsync(); txCoordinator.rollbackAsync()).
+ // - A coordinator might send a commit, then die, but the commit message might still arrive at the commit partition primary.
+ // If it arrived with a delay, another node might come across a write intent/lock from that tx
+ // and realize that the coordinator is no longer available and start tx recovery.
+ // The original commit message might arrive later than the recovery one,
+ // hence a 'commit over rollback' case.
+ // The possible states that a 'commit' is allowed to see:
// - null (if it's the first change state attempt)
// - committed (if it was already updated in the previous attempt)
// - aborted (if it was aborted by the initiate recovery logic,
// though this is a very unlikely case because initiate recovery will only roll back the tx if coordinator is dead).
//
// Within 'roll back' it's allowed to see:
- // - null
- // - aborted
- // Other combinations of states are not possible.
-
- // First, throw an exception if we are trying to abort an already committed tx.
- assert !(txMeta.txState() == COMMITTED && !commit) : "Not allowed to abort an already committed transaction.";
+ // - null (if it's the first change state attempt)
+ // - aborted (if it was already updated in the previous attempt or the result of a concurrent recovery)
+ // - commit (if initiate recovery has started, but a delayed message from the coordinator finally arrived and executed earlier).
- // If a 'commit' sees a tx in the ABORTED state (valid as per the explanation above), let the client know with an exception.
- if (commit && txMeta.txState() == ABORTED) {
- LOG.error("Failed to commit a transaction that is already aborted [txId={}].", txId);
+ // Let the client know a transaction has finished with a different outcome.
+ if (commit != (txMeta.txState() == COMMITTED)) {
+ LOG.error("Failed to finish a transaction that is already finished [txId={}, expectedState={}, actualState={}].",
+ txId,
+ commit ? COMMITTED : ABORTED,
+ txMeta.txState()
+ );
- throw new TransactionException(TX_WAS_ABORTED_ERR,
- "Failed to change the outcome of a finished transaction"
- + " [txId=" + txId + ", txState=" + txMeta.txState() + "].");
+ throw new TransactionAlreadyFinishedException(
+ "Failed to change the outcome of a finished transaction [txId=" + txId + ", txState=" + txMeta.txState() + "].",
+ new TransactionResult(txMeta.txState(), txMeta.commitTimestamp())
+ );
}
}
@@ -1643,10 +1681,27 @@ public class PartitionReplicaListener implements ReplicaListener {
.collect(toList())
)
)
- .whenComplete((o, throwable) -> {
- TxState txState = commit ? COMMITTED : ABORTED;
+ .handle((txOutcome, ex) -> {
+ if (ex != null) {
+ // RAFT 'finish' command failed because the state has already been written by someone else.
+ // In that case we throw a corresponding exception.
+ if (ex instanceof UnexpectedTransactionStateException) {
+ UnexpectedTransactionStateException utse = (UnexpectedTransactionStateException) ex;
+ TransactionResult result = utse.transactionResult();
+
+ markFinished(txId, result.transactionState(), result.commitTimestamp());
+
+ throw new TransactionAlreadyFinishedException(utse.getMessage(), utse.transactionResult());
+ }
+ // Otherwise we convert from the internal exception to the client one.
+ throw new TransactionException(commit ? TX_COMMIT_ERR : TX_ROLLBACK_ERR, ex);
+ }
+
+ TransactionResult result = (TransactionResult) txOutcome;
+
+ markFinished(txId, result.transactionState(), result.commitTimestamp());
- markFinished(txId, txState, commitTimestamp);
+ return null;
});
}
@@ -1678,7 +1733,6 @@ public class PartitionReplicaListener implements ReplicaListener {
}
}
-
/**
* Processes transaction cleanup request:
* <ol>
@@ -1786,7 +1840,7 @@ public class PartitionReplicaListener implements ReplicaListener {
.thenApply(res -> null);
}
- private String getTxCoordinatorId(UUID txId) {
+ private @Nullable String getTxCoordinatorId(UUID txId) {
TxStateMeta meta = txManager.stateMeta(txId);
assert meta != null : "Trying to cleanup a transaction that was not enlisted, txId=" + txId;
@@ -2456,11 +2510,11 @@ public class PartitionReplicaListener implements ReplicaListener {
});
}
- private void applyCmdWithRetryOnSafeTimeReorderException(Command cmd, CompletableFuture<Object> resultFuture) {
+ private <T> void applyCmdWithRetryOnSafeTimeReorderException(Command cmd, CompletableFuture<T> resultFuture) {
applyCmdWithRetryOnSafeTimeReorderException(cmd, resultFuture, 0);
}
- private void applyCmdWithRetryOnSafeTimeReorderException(Command cmd, CompletableFuture<Object> resultFuture, int attemptsCounter) {
+ private <T> void applyCmdWithRetryOnSafeTimeReorderException(Command cmd, CompletableFuture<T> resultFuture, int attemptsCounter) {
attemptsCounter++;
if (attemptsCounter >= MAX_RETIES_ON_SAFE_TIME_REORDERING) {
resultFuture.completeExceptionally(
@@ -2501,7 +2555,7 @@ public class PartitionReplicaListener implements ReplicaListener {
resultFuture.completeExceptionally(ex);
}
} else {
- resultFuture.complete(res);
+ resultFuture.complete((T) res);
}
});
}
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 683cf1c5e5..5841724de3 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
@@ -205,14 +205,14 @@ public class TransactionStateResolver {
private void resolveTxStateFromTxCoordinator(
UUID txId,
- String coordinatorId,
+ @Nullable String coordinatorId,
TablePartitionId commitGrpId,
HybridTimestamp timestamp,
CompletableFuture<TransactionMeta> txMetaFuture
) {
updateLocalTxMapAfterDistributedStateResolved(txId, txMetaFuture);
- ClusterNode coordinator = clusterNodeResolver.getById(coordinatorId);
+ ClusterNode coordinator = coordinatorId == null ? null : clusterNodeResolver.getById(coordinatorId);
if (coordinator == null) {
// This means the coordinator node have either left the cluster or restarted.
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 8f4051a4ff..5b104e26c5 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
@@ -28,6 +28,8 @@ import static org.apache.ignite.internal.testframework.matchers.CompletableFutur
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureExceptionMatcher.willThrowFast;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willSucceedFast;
+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.checkTransitionCorrectness;
import static org.apache.ignite.internal.util.ArrayUtils.asList;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
@@ -49,6 +51,7 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.doAnswer;
@@ -159,6 +162,7 @@ import org.apache.ignite.internal.tostring.S;
import org.apache.ignite.internal.tx.LockManager;
import org.apache.ignite.internal.tx.TransactionIds;
import org.apache.ignite.internal.tx.TransactionMeta;
+import org.apache.ignite.internal.tx.TransactionResult;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.internal.tx.TxMeta;
import org.apache.ignite.internal.tx.TxState;
@@ -179,6 +183,7 @@ import org.apache.ignite.network.ClusterNodeImpl;
import org.apache.ignite.network.ClusterNodeResolver;
import org.apache.ignite.network.MessagingService;
import org.apache.ignite.network.NetworkAddress;
+import org.apache.ignite.network.SingleClusterNodeResolver;
import org.apache.ignite.network.TopologyService;
import org.apache.ignite.sql.ColumnType;
import org.apache.ignite.tx.TransactionException;
@@ -256,6 +261,10 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
return v;
});
+ } else if (cmd instanceof FinishTxCommand) {
+ FinishTxCommand command = (FinishTxCommand) cmd;
+
+ return completedFuture(new TransactionResult(command.commit() ? COMMITTED : ABORTED, command.commitTimestamp()));
}
return nullCompletedFuture();
@@ -501,7 +510,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
schemaSyncService,
catalogService,
new TestPlacementDriver(localNode),
- mock(ClusterNodeResolver.class)
+ new SingleClusterNodeResolver(localNode)
);
kvMarshaller = marshallerFor(schemaDescriptor);
@@ -547,6 +556,14 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
@Test
public void testTxStateReplicaRequestEmptyState() throws Exception {
+ doAnswer(invocation -> {
+ UUID txId = invocation.getArgument(4);
+
+ txManager.updateTxMeta(txId, old -> new TxStateMeta(ABORTED, localNode.id(), commitPartitionId, null));
+
+ return nullCompletedFuture();
+ }).when(txManager).finish(any(), any(), anyBoolean(), any(), any());
+
CompletableFuture<ReplicaResult> fut = partitionReplicaListener.invoke(TX_MESSAGES_FACTORY.txStateCommitPartitionRequest()
.groupId(grpId)
.txId(newTxId())
@@ -555,7 +572,9 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
TransactionMeta txMeta = (TransactionMeta) fut.get(1, TimeUnit.SECONDS).result();
- assertNull(txMeta);
+ assertNotNull(txMeta);
+
+ assertEquals(ABORTED, txMeta.txState());
}
@Test
@@ -1878,6 +1897,9 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
}).when(txManager).updateTxMeta(any(), any());
doAnswer(invocation -> nullCompletedFuture()).when(txManager).executeCleanupAsync(any(Runnable.class));
+
+ doAnswer(invocation -> nullCompletedFuture()).when(txManager).finish(any(), any(), anyBoolean(), any(), any());
+ doAnswer(invocation -> nullCompletedFuture()).when(txManager).cleanup(anyString(), any());
}
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 b6b433bd91..88a44ecc3e 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
@@ -558,7 +558,7 @@ public class ItTxTestCluster {
new AlwaysSyncedSchemaSyncService(),
catalogService,
placementDriver,
- clusterServices.get(assignment).topologyService()
+ nodeResolver
);
replicaManagers.get(assignment).startReplica(
diff --git a/modules/table/tech-notes/finish.puml b/modules/table/tech-notes/finish.puml
new file mode 100644
index 0000000000..c61d57f68d
--- /dev/null
+++ b/modules/table/tech-notes/finish.puml
@@ -0,0 +1,26 @@
+@startuml
+
+!pragma teoz true
+group Successful commit
+coordinator -> commitPartitionPrimary: COMMIT
+commitPartitionPrimary -> RAFT : Finish Command
+RAFT -> RAFT : CAS tx state null to COMMIT.
+RAFT --> commitPartitionPrimary
+commitPartitionPrimary -> commitPartitionPrimary: store COMMIT \nin a volatile map.\nCOMMIT Stored
+commitPartitionPrimary --> coordinator
+end
+
+group Response timeout
+coordinator -> commitPartitionPrimary: COMMIT
+commitPartitionPrimary -> RAFT : Finish Command
+RAFT -> RAFT : CAS tx state null to COMMIT.
+RAFT --> commitPartitionPrimary
+commitPartitionPrimary -> commitPartitionPrimary: store COMMIT \nin a volatile map.\nCOMMIT Stored
+commitPartitionPrimary --> coordinator !! : response timeout
+
+coordinator -> commitPartitionPrimary: COMMIT - durable finish
+commitPartitionPrimary -> commitPartitionPrimary: TX already finished.
+commitPartitionPrimary --> coordinator
+end
+
+@enduml
diff --git a/modules/table/tech-notes/recovery-finish.puml b/modules/table/tech-notes/recovery-finish.puml
new file mode 100644
index 0000000000..90a387e920
--- /dev/null
+++ b/modules/table/tech-notes/recovery-finish.puml
@@ -0,0 +1,104 @@
+@startuml
+
+!pragma teoz true
+participant coordinator
+participant messaging
+participant commitPartitionPrimary
+participant tx2Coordinator
+participant RAFT
+participant commitPartitionPrimary2
+
+group Coordinator commits and dies. Recovery.
+coordinator -[#green]> messaging: COMMIT TX0
+note over coordinator: Coordinator dies
+note over messaging: commit message is delayed in messaging
+tx2Coordinator -[#red]> tx2Coordinator: Start TX0 recovery
+messaging -[#green]> commitPartitionPrimary: commit TX0
+tx2Coordinator -[#red]> commitPartitionPrimary: recover TX0
+commitPartitionPrimary -[#green]> RAFT : Finish Command with COMMIT
+RAFT -[#green]> RAFT : CAS tx state\nnull to COMMIT.
+RAFT --[#green]> commitPartitionPrimary
+commitPartitionPrimary -[#green]> commitPartitionPrimary: store COMMIT \nin a volatile map.\nCOMMIT Stored
+commitPartitionPrimary --[#green]> coordinator !! : No recipient
+commitPartitionPrimary -[#red]> RAFT : Finish Command with ABORT
+RAFT -[#red]> RAFT : FAILED CAS tx state\nCOMMIT to ABORT.
+RAFT --[#red]> commitPartitionPrimary
+commitPartitionPrimary -[#red]> commitPartitionPrimary: store ABORT \nin a volatile map.\nCOMMIT not changed
+commitPartitionPrimary --[#red]> tx2Coordinator
+end
+
+
+group Coordinator commits and dies. Recovery. Reorder
+coordinator -[#green]> messaging: COMMIT TX0
+note over coordinator: Coordinator dies
+note over messaging: commit message is delayed in messaging
+tx2Coordinator -[#red]> tx2Coordinator: Start TX0 recovery
+messaging -[#green]> commitPartitionPrimary: commit TX0
+tx2Coordinator -[#red]> commitPartitionPrimary: recover TX0
+commitPartitionPrimary -[#green]> RAFT : Finish Command with COMMIT
+commitPartitionPrimary -[#red]> RAFT : Finish Command with ABORT
+RAFT -[#green]> RAFT : CAS tx state\nnull to COMMIT.
+RAFT -[#red]> RAFT : FAILED CAS tx state\nCOMMIT to ABORT.
+RAFT --[#red]> commitPartitionPrimary
+commitPartitionPrimary -[#red]> commitPartitionPrimary: store COMMIT \nin a volatile map.\nCOMMIT stored
+commitPartitionPrimary --[#red]> tx2Coordinator
+RAFT --[#green]> commitPartitionPrimary
+commitPartitionPrimary -[#green]> commitPartitionPrimary: store COMMIT \nin a volatile map.\nCOMMIT is not changed
+
+commitPartitionPrimary --[#green]> coordinator !! : No recipient
+end
+
+group Coordinator sends commit and dies. Recovery wins.
+coordinator -[#green]> messaging: COMMIT TX0
+note over coordinator: Coordinator dies
+note over messaging: commit message is delayed in messaging
+tx2Coordinator -[#red]> tx2Coordinator: Start TX0 recovery
+tx2Coordinator -[#red]> commitPartitionPrimary: recover TX0
+messaging -[#green]> commitPartitionPrimary: commit TX0
+commitPartitionPrimary -[#red]> RAFT : Finish Command with ABORT
+RAFT -[#red]> RAFT : CAS tx state\nnull to ABORT.
+RAFT --[#red]> commitPartitionPrimary
+commitPartitionPrimary -[#red]> commitPartitionPrimary: store ABORT \nin a volatile map.\nABORT stored
+commitPartitionPrimary --[#red]> tx2Coordinator
+commitPartitionPrimary -[#green]> RAFT : Finish Command with COMMIT
+RAFT -[#green]> RAFT : FAILED CAS tx state\nABORT to COMMIT.
+RAFT --[#green]> commitPartitionPrimary
+commitPartitionPrimary -[#green]> commitPartitionPrimary: store COMMIT \nin a volatile map.\nABORT is not changed
+commitPartitionPrimary --[#green]> coordinator !! : No recipient
+end
+
+group Coordinator sends commit and dies. Recovery finishes before commit execution.
+coordinator -[#green]> messaging: COMMIT TX0
+note over coordinator: Coordinator dies
+note over messaging: commit message is delayed in messaging
+tx2Coordinator -[#red]> tx2Coordinator: Start TX0 recovery
+tx2Coordinator -[#red]> commitPartitionPrimary: recover TX0
+
+commitPartitionPrimary -[#red]> RAFT : Finish Command with ABORT
+RAFT -[#red]> RAFT : CAS tx state\nnull to ABORT.
+RAFT --[#red]> commitPartitionPrimary
+commitPartitionPrimary -[#red]> commitPartitionPrimary: store ABORT \nin a volatile map.\nABORT stored
+commitPartitionPrimary --[#red]> tx2Coordinator
+messaging -[#green]> commitPartitionPrimary: commit TX0
+commitPartitionPrimary -[#green]> commitPartitionPrimary : TX is already finished.\nCOMMIT over ABORT\nresutls in TX_WAS_ABORTED_ERR
+commitPartitionPrimary --[#green]> coordinator !! : No recipient
+end
+
+group Coordinator dead. Two recoveries.
+tx2Coordinator -[#red]> tx2Coordinator: Start TX0 recovery
+tx2Coordinator -[#red]> commitPartitionPrimary: recover TX0
+commitPartitionPrimary -[#red]> RAFT : Finish Command with ABORT
+note over commitPartitionPrimary: Lease expires
+tx2Coordinator -[#blue]> tx2Coordinator: Start TX0 recovery
+tx2Coordinator -[#blue]> commitPartitionPrimary2: recover TX0
+RAFT -[#red]> RAFT : CAS tx state\nnull to ABORT.
+RAFT --[#red]> commitPartitionPrimary
+commitPartitionPrimary -[#red]> commitPartitionPrimary: store ABORT \nin a volatile map.\nABORT stored
+commitPartitionPrimary2 -[#blue]> RAFT : Finish Command with ABORT
+RAFT -[#blue]> RAFT : FAILED CAS tx state\nABORT to ABORT.
+RAFT --[#blue]> commitPartitionPrimary2
+commitPartitionPrimary2 -[#blue]> commitPartitionPrimary2: store ABORT \nin a volatile map.\nABORT stored
+commitPartitionPrimary --[#red]> tx2Coordinator
+commitPartitionPrimary2 --[#blue]> tx2Coordinator
+end
+@enduml
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TransactionAbandonedException.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TransactionAbandonedException.java
deleted file mode 100644
index c35e1ada47..0000000000
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TransactionAbandonedException.java
+++ /dev/null
@@ -1,54 +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.tx;
-
-import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_ABANDONED_ERR;
-
-import java.util.UUID;
-import org.apache.ignite.tx.TransactionException;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * Exception that is thrown when write intent can't be resolved because it has been created by a transaction that is now abandoned
- * (has a state that currently can't be known).
- */
-public class TransactionAbandonedException extends TransactionException {
- private final UUID abandonedTxId;
-
- @Nullable
- private final TransactionMeta transactionMeta;
-
- /**
- * Constructor.
- *
- * @param abandonedTxId ID of the transaction that is abandoned.
- * @param transactionMeta Transaction meta that was able to be retrieved.
- */
- public TransactionAbandonedException(UUID abandonedTxId, @Nullable TransactionMeta transactionMeta) {
- super(UUID.randomUUID(), TX_ABANDONED_ERR, null);
-
- this.abandonedTxId = abandonedTxId;
- this.transactionMeta = transactionMeta;
- }
-
- @Override
- public String getMessage() {
- return "Operation failed due to abandoned transaction [abandonedTxId=" + abandonedTxId
- + ", transactionMeta=" + transactionMeta + ']';
- }
-}
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TransactionAlreadyFinishedException.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TransactionAlreadyFinishedException.java
new file mode 100644
index 0000000000..2121aafa3d
--- /dev/null
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TransactionAlreadyFinishedException.java
@@ -0,0 +1,42 @@
+/*
+ * 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.tx;
+
+import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_UNEXPECTED_STATE_ERR;
+
+import org.apache.ignite.tx.TransactionException;
+
+/**
+ * The exception is thrown when a transaction has already been finished.
+ */
+public class TransactionAlreadyFinishedException extends TransactionException {
+
+ private static final long serialVersionUID = -7953057695915339651L;
+
+ /** Stored transaction result. */
+ private final TransactionResult transactionResult;
+
+ public TransactionAlreadyFinishedException(String message, TransactionResult transactionResult) {
+ super(TX_UNEXPECTED_STATE_ERR, message);
+ this.transactionResult = transactionResult;
+ }
+
+ public TransactionResult transactionResult() {
+ return transactionResult;
+ }
+}
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TransactionResult.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TransactionResult.java
new file mode 100644
index 0000000000..a0a2fb0ea3
--- /dev/null
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TransactionResult.java
@@ -0,0 +1,51 @@
+/*
+ * 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.tx;
+
+import static org.apache.ignite.internal.hlc.HybridTimestamp.hybridTimestampToLong;
+import static org.apache.ignite.internal.hlc.HybridTimestamp.nullableHybridTimestamp;
+
+import java.io.Serializable;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Represents transaction final result - the state and the commit timestamp.
+ */
+public class TransactionResult implements Serializable {
+ private static final long serialVersionUID = -4975086804837572978L;
+
+ /** Stored transaction state. */
+ private final TxState transactionState;
+
+ /** Commit timestamp if the {@link #transactionState} is {@link TxState#COMMITTED}. */
+ private final long commitTimestamp;
+
+ public TransactionResult(TxState transactionState, @Nullable HybridTimestamp commitTimestamp) {
+ this.transactionState = transactionState;
+ this.commitTimestamp = hybridTimestampToLong(commitTimestamp);
+ }
+
+ public TxState transactionState() {
+ return transactionState;
+ }
+
+ public @Nullable HybridTimestamp commitTimestamp() {
+ return nullableHybridTimestamp(commitTimestamp);
+ }
+}
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 b1f3e2a246..428f1b63f4 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
@@ -146,6 +146,15 @@ public interface TxManager extends IgniteComponent {
UUID txId
);
+ /**
+ * Sends cleanup request to the nodes than initiated recovery.
+ *
+ * @param node Target node.
+ * @param txId Transaction id.
+ * @return Completable future of Void.
+ */
+ CompletableFuture<Void> cleanup(String node, UUID txId);
+
/**
* Returns a number of finished transactions.
*
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 5026af6c34..dad2f07648 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
@@ -56,7 +56,7 @@ public enum TxState {
ABANDONED;
private static final boolean[][] TRANSITION_MATRIX = {
- { false, true, false, true, true, true },
+ { false, true, true, true, true, true },
{ false, true, true, true, true, true },
{ false, false, false, true, true, true },
{ false, false, false, true, false, false },
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 0990bbfb59..7cb0917710 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,6 @@
package org.apache.ignite.internal.tx;
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;
import java.util.Objects;
@@ -52,7 +51,7 @@ public class TxStateMeta implements TransactionMeta {
*/
public TxStateMeta(
TxState txState,
- String txCoordinatorId,
+ @Nullable String txCoordinatorId,
@Nullable TablePartitionId commitPartitionId,
@Nullable HybridTimestamp commitTimestamp
) {
@@ -70,9 +69,9 @@ public class TxStateMeta implements TransactionMeta {
*/
private TxStateMeta(
TxState txState,
- String txCoordinatorId,
- TablePartitionId commitPartitionId,
- HybridTimestamp commitTimestamp,
+ @Nullable String txCoordinatorId,
+ @Nullable TablePartitionId commitPartitionId,
+ @Nullable HybridTimestamp commitTimestamp,
long lastAbandonedMarkerTs
) {
this.txState = txState;
@@ -98,8 +97,6 @@ public class TxStateMeta implements TransactionMeta {
* @return Transaction state meta.
*/
public TxStateMetaFinishing finishing() {
- assert checkTransitionCorrectness(txState, FINISHING) : "Transaction state is incorrect [txState=" + txState + "].";
-
return new TxStateMetaFinishing(txCoordinatorId, commitPartitionId);
}
@@ -108,11 +105,11 @@ public class TxStateMeta implements TransactionMeta {
return txState;
}
- public String txCoordinatorId() {
+ public @Nullable String txCoordinatorId() {
return txCoordinatorId;
}
- public TablePartitionId commitPartitionId() {
+ public @Nullable TablePartitionId commitPartitionId() {
return commitPartitionId;
}
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 b08b2f57e0..13659d8070 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
@@ -38,7 +38,7 @@ public class TxStateMetaFinishing extends TxStateMeta {
* @param txCoordinatorId Transaction coordinator id.
* @param commitPartitionId Commit partition id.
*/
- public TxStateMetaFinishing(String txCoordinatorId, @Nullable TablePartitionId commitPartitionId) {
+ public TxStateMetaFinishing(@Nullable String txCoordinatorId, @Nullable TablePartitionId commitPartitionId) {
super(TxState.FINISHING, txCoordinatorId, commitPartitionId, null);
}
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 664934ea2a..f8918051f4 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
@@ -244,7 +244,7 @@ public class OrphanDetector {
* @return True when the transaction coordinator is alive, false otherwise.
*/
private boolean isTxCoordinatorAlive(TxStateMeta txState) {
- return topologyService.getById(txState.txCoordinatorId()) != null;
+ return txState.txCoordinatorId() != null && topologyService.getById(txState.txCoordinatorId()) != null;
}
/**
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 0df5bc3495..af4a7ef15a 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
@@ -17,7 +17,10 @@
package org.apache.ignite.internal.tx.impl;
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.emptySet;
import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.ignite.internal.util.ExceptionUtils.withCause;
import static org.apache.ignite.lang.ErrorGroups.Replicator.REPLICA_UNAVAILABLE_ERR;
@@ -96,6 +99,10 @@ public class PlacementDriverHelper {
* failed to find the primary for.
*/
public CompletableFuture<PartitionData> findPrimaryReplicas(Collection<TablePartitionId> partitions) {
+ if (partitions == null || partitions.isEmpty()) {
+ return completedFuture(new PartitionData(emptyMap(), emptySet()));
+ }
+
HybridTimestamp timestamp = clock.now();
Map<TablePartitionId, CompletableFuture<ReplicaMeta>> primaryReplicaFutures = new HashMap<>();
@@ -121,7 +128,7 @@ public class PlacementDriverHelper {
TablePartitionId partition = entry.getKey();
- if (meta != null) {
+ if (meta != null && meta.getLeaseholder() != null) {
partitionsByNode.computeIfAbsent(meta.getLeaseholder(), s -> new HashSet<>())
.add(partition);
} else {
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 46815c062f..b8f2468cff 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
@@ -63,6 +63,17 @@ public class TxCleanupRequestSender {
this.writeIntentSwitchProcessor = writeIntentSwitchProcessor;
}
+ /**
+ * Sends unlock request to the nodes than initiated recovery.
+ *
+ * @param node Target node.
+ * @param txId Transaction id.
+ * @return Completable future of Void.
+ */
+ public CompletableFuture<Void> cleanup(String node, UUID txId) {
+ return sendCleanupMessageWithRetries(false, null, txId, node, null);
+ }
+
/**
* Sends cleanup request to the primary nodes of each one of {@code partitions}.
*
@@ -123,7 +134,7 @@ public class TxCleanupRequestSender {
@Nullable HybridTimestamp commitTimestamp,
UUID txId,
String node,
- Collection<TablePartitionId> partitions
+ @Nullable Collection<TablePartitionId> partitions
) {
Collection<ReplicationGroupId> enlistedPartitions = (Collection<ReplicationGroupId>) (Collection<?>) partitions;
@@ -138,6 +149,12 @@ public class TxCleanupRequestSender {
// either have a new mapping of primary to its partitions
// or will run `switchWriteIntentsOnPartitions` for partitions with no primary.
// At the end of the day all write intents will be properly converted.
+ if (partitions == null) {
+ // If we don't have any partition, which is the recovery case,
+ // just try again with the same node.
+ return sendCleanupMessageWithRetries(commit, commitTimestamp, txId, node, partitions);
+ }
+
return cleanup(partitions, commit, commitTimestamp, txId);
}
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 fcc14cb871..2c6d4e4243 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
@@ -22,6 +22,7 @@ import static java.util.concurrent.CompletableFuture.runAsync;
import static java.util.concurrent.CompletableFuture.supplyAsync;
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.checkTransitionCorrectness;
import static org.apache.ignite.internal.tx.TxState.isFinalState;
@@ -30,7 +31,6 @@ import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination;
import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_READ_ONLY_TOO_OLD_ERR;
-import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_WAS_ABORTED_ERR;
import java.io.IOException;
import java.util.Collection;
@@ -73,7 +73,9 @@ import org.apache.ignite.internal.thread.NamedThreadFactory;
import org.apache.ignite.internal.tx.HybridTimestampTracker;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.tx.LockManager;
+import org.apache.ignite.internal.tx.TransactionAlreadyFinishedException;
import org.apache.ignite.internal.tx.TransactionMeta;
+import org.apache.ignite.internal.tx.TransactionResult;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.internal.tx.TxState;
import org.apache.ignite.internal.tx.TxStateMeta;
@@ -85,7 +87,6 @@ import org.apache.ignite.internal.util.IgniteSpinBusyLock;
import org.apache.ignite.network.ClusterService;
import org.apache.ignite.network.NetworkMessage;
import org.apache.ignite.network.NetworkMessageHandler;
-import org.apache.ignite.tx.TransactionException;
import org.jetbrains.annotations.Nullable;
/**
@@ -349,7 +350,28 @@ 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 = updateTxMeta(txId, TxStateMeta::finishing);
+
+ // First we check the current tx state to guarantee txFinish idempotence.
+ TxStateMeta txMeta = stateMeta(txId);
+
+ TxStateMetaFinishing finishingStateMeta =
+ txMeta == null
+ ? new TxStateMetaFinishing(null, commitPartition)
+ : txMeta.finishing();
+
+ TxStateMeta stateMeta = updateTxMeta(txId, oldMeta -> finishingStateMeta);
+
+ // Means we failed to CAS the state, someone else did it.
+ if (finishingStateMeta != stateMeta) {
+ // If the state is FINISHING then someone else hase in in the middle of finishing this tx.
+ if (stateMeta.txState() == FINISHING) {
+ return ((TxStateMetaFinishing) stateMeta).txFinishFuture()
+ .thenCompose(meta -> checkTxOutcome(commit, txId, meta));
+ } else {
+ // The TX has already been finished. Check whether it finished with the same outcome.
+ return checkTxOutcome(commit, txId, stateMeta);
+ }
+ }
TxContext tuple = txCtxMap.compute(txId, (uuid, tuple0) -> {
if (tuple0 == null) {
@@ -375,6 +397,17 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler {
));
}
+ private static CompletableFuture<Void> checkTxOutcome(boolean commit, UUID txId, TransactionMeta stateMeta) {
+ if ((stateMeta.txState() == COMMITTED) == commit) {
+ return nullCompletedFuture();
+ }
+
+ return CompletableFuture.failedFuture(new TransactionAlreadyFinishedException(
+ "Failed to change the outcome of a finished transaction [txId=" + txId + ", txState=" + stateMeta.txState() + "].",
+ new TransactionResult(stateMeta.txState(), stateMeta.commitTimestamp()))
+ );
+ }
+
private CompletableFuture<Void> prepareFinish(
HybridTimestampTracker observableTimestampTracker,
TablePartitionId commitPartition,
@@ -439,20 +472,23 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler {
if (ex != null) {
Throwable cause = ExceptionUtils.unwrapCause(ex);
- if (cause instanceof TransactionException) {
- TransactionException transactionException = (TransactionException) cause;
+ if (cause instanceof TransactionAlreadyFinishedException) {
+ TransactionAlreadyFinishedException transactionException = (TransactionAlreadyFinishedException) cause;
- if (transactionException.code() == TX_WAS_ABORTED_ERR) {
- updateTxMeta(txId, old -> {
- TxStateMeta txStateMeta = new TxStateMeta(ABORTED, old.txCoordinatorId(), commitPartition, null);
+ TransactionResult result = transactionException.transactionResult();
- txFinishFuture.complete(txStateMeta);
+ TxStateMeta updatedMeta = updateTxMeta(txId, old ->
+ new TxStateMeta(
+ result.transactionState(),
+ old.txCoordinatorId(),
+ commitPartition,
+ result.commitTimestamp()
+ )
+ );
- return txStateMeta;
- });
+ txFinishFuture.complete(updatedMeta);
- return CompletableFuture.<Void>failedFuture(cause);
- }
+ return CompletableFuture.<Void>failedFuture(cause);
}
if (TransactionFailureHandler.isRecoverable(cause)) {
@@ -581,6 +617,11 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler {
return txCleanupRequestSender.cleanup(partitions, commit, commitTimestamp, txId);
}
+ @Override
+ public CompletableFuture<Void> cleanup(String node, UUID txId) {
+ return txCleanupRequestSender.cleanup(node, txId);
+ }
+
@Override
public CompletableFuture<Void> executeCleanupAsync(Runnable runnable) {
return runAsync(runnable, cleanupExecutor);
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 3016c3dd96..ccae5f127c 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
@@ -105,7 +105,7 @@ public class TxMessageSender {
*/
public CompletableFuture<NetworkMessage> cleanup(
String primaryConsistentId,
- Collection<ReplicationGroupId> replicationGroupIds,
+ @Nullable Collection<ReplicationGroupId> replicationGroupIds,
UUID txId,
boolean commit,
@Nullable HybridTimestamp commitTimestamp
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 5f5af26836..2f36308c50 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
@@ -55,14 +55,12 @@ public class TxStateTest {
@Test
void testTransitionsFromNull() {
- // Not allowed.
- assertFalse(TxState.checkTransitionCorrectness(null, TxState.FINISHING));
-
// Allowed.
assertTrue(TxState.checkTransitionCorrectness(null, TxState.PENDING));
assertTrue(TxState.checkTransitionCorrectness(null, TxState.ABORTED));
assertTrue(TxState.checkTransitionCorrectness(null, TxState.COMMITTED));
assertTrue(TxState.checkTransitionCorrectness(null, TxState.ABANDONED));
+ assertTrue(TxState.checkTransitionCorrectness(null, TxState.FINISHING));
}
@Test