vldpyatkov commented on code in PR #2751:
URL: https://github.com/apache/ignite-3/pull/2751#discussion_r1378515481
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -1437,6 +1450,47 @@ private CompletableFuture<Void> finishAndCleanup(
UUID txId,
String txCoordinatorId
) {
+ TxMeta txMeta = txStateStorage.get(txId);
+
+ // Check that a transaction has already been finished.
+ boolean transactionAlreadyFinished = txMeta != null &&
isFinalState(txMeta.txState());
+
+ // Check locksReleased flag. If it is already set, do nothing and
return a successful result.
+ // Even if the outcome is different (the transaction was aborted, but
we want to commit it),
+ // we return 'success' to be in alignment with common transaction
handling.
+ if (transactionAlreadyFinished) {
+ if (txMeta.locksReleased()) {
+ return completedFuture(null);
+ }
+ // If the locks were not released, we are likely to be in a
recovery mode and retrying the finish request.
+ // In this case we want to check the expected outcome and the
actual one.
+ if (commit && txMeta.txState() == ABORTED) {
Review Comment:
And what happens if `!commit && txMeta.txState() == COMMITED`?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -1475,24 +1529,13 @@ private CompletableFuture<Void> cleanupWithRetry(
UUID txId,
TablePartitionId partitionId,
int attempts) {
- HybridTimestamp now = hybridClock.now();
-
- return findPrimaryReplica(partitionId, now)
+ return findPrimaryReplica(partitionId, hybridClock.now())
.thenCompose(leaseHolder ->
- cleanupWithRetryOnReplica(commit, commitTimestamp,
txId, partitionId, leaseHolder, attempts));
- }
-
- private CompletableFuture<Void> cleanupWithRetryOnReplica(
- boolean commit,
- @Nullable HybridTimestamp commitTimestamp,
- UUID txId,
- TablePartitionId partitionId,
- String primaryConsistentId,
- int attempts) {
- return txManager.cleanup(primaryConsistentId, partitionId, txId,
commit, commitTimestamp)
+ txManager.cleanup(leaseHolder, partitionId, txId,
commit, commitTimestamp))
.handle((res, ex) -> {
if (ex != null) {
- LOG.warn("Failed to perform cleanup on Tx {}." +
(attempts > 0 ? " The operation will be retried." : ""), txId, ex);
+ LOG.warn("Failed to perform cleanup on Tx {}." +
(attempts > 0 ? " The operation will be retried." : ""),
Review Comment:
Let's correct the message according to our rules.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -1437,6 +1450,47 @@ private CompletableFuture<Void> finishAndCleanup(
UUID txId,
String txCoordinatorId
) {
+ TxMeta txMeta = txStateStorage.get(txId);
+
+ // Check that a transaction has already been finished.
+ boolean transactionAlreadyFinished = txMeta != null &&
isFinalState(txMeta.txState());
+
+ // Check locksReleased flag. If it is already set, do nothing and
return a successful result.
+ // Even if the outcome is different (the transaction was aborted, but
we want to commit it),
+ // we return 'success' to be in alignment with common transaction
handling.
+ if (transactionAlreadyFinished) {
+ if (txMeta.locksReleased()) {
+ return completedFuture(null);
+ }
+ // If the locks were not released, we are likely to be in a
recovery mode and retrying the finish request.
+ // In this case we want to check the expected outcome and the
actual one.
+ if (commit && txMeta.txState() == ABORTED) {
+ LOG.error("Failed to commit a transaction {} that is
aborted.", txId);
Review Comment:
Let's stick to the format where we have a message and parameters at the tail
of the square brackets.
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java:
##########
@@ -295,139 +312,244 @@ public void finishFull(HybridTimestampTracker
timestampTracker, UUID txId, boole
updateTxMeta(txId, old -> new TxStateMeta(finalState,
old.txCoordinatorId(), old.commitTimestamp()));
}
+ private @Nullable HybridTimestamp getCommitTimestamp(boolean commit) {
+ return commit ? clock.now() : null;
+ }
+
+ private String coordinatorId() {
+ return localNodeId.get();
+ }
+
+ @Override
+ public CompletableFuture<Void> finish(boolean commit, UUID txId) {
+ // If there are no enlisted groups, just update local state - we
already marked the tx as finished.
+ updateTxMeta(txId, old -> coordinatorFinalTxStateMeta(commit,
getCommitTimestamp(commit)));
+
+ return completedFuture(null);
+ }
+
@Override
public CompletableFuture<Void> finish(
HybridTimestampTracker observableTimestampTracker,
TablePartitionId commitPartition,
- ClusterNode recipientNode,
- Long term,
boolean commit,
Map<TablePartitionId, Long> enlistedGroups,
UUID txId
) {
assert enlistedGroups != null;
+ assert !enlistedGroups.isEmpty() : "No enlisted partitions found";
// Here we put finishing state meta into the local map, so that all
concurrent operations trying to read tx state
// with using read timestamp could see that this transaction is
finishing, see #transactionMetaReadTimestampAware(txId, timestamp).
// None of them now are able to update node's clock with read
timestamp and we can create the commit timestamp that is greater
// 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 = new
TxStateMetaFinishing(localNodeId.get());
- updateTxMeta(txId, old -> finishingStateMeta);
- HybridTimestamp commitTimestamp = commit ? clock.now() : null;
-
- // If there are no enlisted groups, just return - we already marked
the tx as finished.
- boolean finishRequestNeeded = !enlistedGroups.isEmpty();
-
- if (!finishRequestNeeded) {
- updateTxMeta(txId, old -> {
- TxStateMeta finalStateMeta =
coordinatorFinalTxStateMeta(commit, commitTimestamp);
-
- finishingStateMeta.txFinishFuture().complete(finalStateMeta);
-
- return finalStateMeta;
- });
+ TxStateMetaFinishing finishingStateMeta = new
TxStateMetaFinishing(coordinatorId());
- return completedFuture(null);
- }
+ updateTxMeta(txId, old -> finishingStateMeta);
- Function<Void, CompletableFuture<Void>> clo = ignored -> {
- // In case of commit it's required to check whether current
primaries are still the same that were enlisted and whether
- // given primaries are not expired or, in other words, whether
commitTimestamp is less or equal to the enlisted primaries
- // expiration timestamps.
- CompletableFuture<Void> verificationFuture =
- commit ? verifyCommitTimestamp(enlistedGroups,
commitTimestamp) : completedFuture(null);
-
- return verificationFuture.handle(
- (unused, throwable) -> {
- Collection<ReplicationGroupId> replicationGroupIds =
new HashSet<>(enlistedGroups.keySet());
-
- boolean verifiedCommit = throwable == null && commit;
-
- TxFinishReplicaRequest req =
FACTORY.txFinishReplicaRequest()
- .txId(txId)
- .timestampLong(clock.nowLong())
- .groupId(commitPartition)
- .groups(replicationGroupIds)
- // In case of verification future failure
transaction will be rolled back.
- .commit(verifiedCommit)
-
.commitTimestampLong(hybridTimestampToLong(commitTimestamp))
- .term(term)
- .build();
-
- return replicaService.invoke(recipientNode,
req).thenRun(
- () -> {
- updateTxMeta(txId, old -> {
- if (isFinalState(old.txState())) {
-
finishingStateMeta.txFinishFuture().complete(old);
-
- return old;
- }
-
- assert old instanceof
TxStateMetaFinishing;
-
- TxStateMeta finalTxStateMeta =
coordinatorFinalTxStateMeta(verifiedCommit, commitTimestamp);
-
-
finishingStateMeta.txFinishFuture().complete(finalTxStateMeta);
-
- return finalTxStateMeta;
- });
-
- if (verifiedCommit) {
-
observableTimestampTracker.update(commitTimestamp);
- }
- });
- })
- .thenCompose(Function.identity())
- // verification future is added in order to share proper
exception with the client
- .thenCompose(r -> verificationFuture);
- };
-
- AtomicReference<CompletableFuture<Void>> ref = new AtomicReference<>();
+ AtomicBoolean performingFinish = new AtomicBoolean();
TxContext tuple = txCtxMap.compute(txId, (uuid, tuple0) -> {
if (tuple0 == null) {
tuple0 = new TxContext(); // No writes enlisted.
}
- if (tuple0.finishFut == null) {
- tuple0.finishFut = new CompletableFuture<>();
- ref.set(tuple0.finishFut);
+ if (!tuple0.isTxFinishing()) {
+ tuple0.finishTx();
+
+ performingFinish.set(true);
}
return tuple0;
});
- if (ref.get() != null) { // This is a finishing thread.
- if (!commit) {
- clo.apply(null).handle((ignored, err) -> {
+ // This is a finishing thread.
+ if (performingFinish.get()) {
+ Function<Void, CompletableFuture<Void>> finishAction = ignored ->
+ makeFinishRequest(
+ observableTimestampTracker,
+ commitPartition,
+ commit,
+ enlistedGroups,
+ txId,
+ finishingStateMeta.txFinishFuture()
+ );
+
+ runFinish(commit, tuple, finishAction);
+ }
+
+ // The method `runFinish` has a side effect on
`finishInProgressFuture` future, it kicks off another future that will complete
it.
+ return tuple.finishInProgressFuture;
+ }
+
+ private static void runFinish(boolean commit, TxContext tuple,
Function<Void, CompletableFuture<Void>> finishAction) {
+ // Wait for commit acks first, then proceed with the finish request.
+ CompletableFuture<Void> finisher = commit ? tuple.waitNoInflights() :
completedFuture(null);
+
+ finisher
+ .thenCompose(finishAction)
+ .handle((ignored, err) -> {
if (err == null) {
- tuple.finishFut.complete(null);
+ tuple.finishInProgressFuture.complete(null);
} else {
- tuple.finishFut.completeExceptionally(err);
+
tuple.finishInProgressFuture.completeExceptionally(err);
}
return null;
});
- } else {
+ }
- // All inflights have been completed before the finish.
- if (tuple.inflights == 0) {
- tuple.waitRepFut.complete(null);
- }
+ private CompletableFuture<Void> makeFinishRequest(
+ HybridTimestampTracker observableTimestampTracker,
+ TablePartitionId commitPartition,
+ boolean commit,
+ Map<TablePartitionId, Long> enlistedGroups,
+ UUID txId,
+ CompletableFuture<TransactionMeta> txFinishFuture
+ ) {
+ HybridTimestamp commitTimestamp = getCommitTimestamp(commit);
+ // In case of commit it's required to check whether current primaries
are still the same that were enlisted and whether
+ // given primaries are not expired or, in other words, whether
commitTimestamp is less or equal to the enlisted primaries
+ // expiration timestamps.
+ CompletableFuture<Void> verificationFuture =
+ commit ? verifyCommitTimestamp(enlistedGroups,
commitTimestamp) : completedFuture(null);
+
+ return verificationFuture.handle(
+ (unused, throwable) -> {
+ boolean verifiedCommit = throwable == null &&
commit;
+
+ Collection<ReplicationGroupId> replicationGroupIds
= new HashSet<>(enlistedGroups.keySet());
+
+ return makeDurableFinishRequest(
+ observableTimestampTracker,
+ commitPartition,
+ verifiedCommit,
+ replicationGroupIds,
+ txId,
+ commitTimestamp,
+ txFinishFuture);
+ })
+ .thenCompose(Function.identity())
+ // verification future is added in order to share proper
exception with the client
+ .thenCompose(r -> verificationFuture);
+ }
- // Wait for commit acks first, then proceed with the finish
request.
- tuple.waitRepFut.thenCompose(clo).handle((ignored, err) -> {
- if (err == null) {
- tuple.finishFut.complete(null);
- } else {
- tuple.finishFut.completeExceptionally(err);
+ /**
+ * Durable finish request.
+ */
+ private CompletableFuture<Void> makeDurableFinishRequest(
+ HybridTimestampTracker observableTimestampTracker,
+ TablePartitionId commitPartition,
+ boolean commit,
+ Collection<ReplicationGroupId> replicationGroupIds,
+ UUID txId,
+ HybridTimestamp commitTimestamp,
+ CompletableFuture<TransactionMeta> txFinishFuture
+ ) {
+ return inBusyLockAsync(busyLock, () ->
findPrimaryReplica(commitPartition, clock.now())
+ .thenCompose(meta ->
+ finishOnPrimary(
+ observableTimestampTracker,
+ commitPartition,
+ meta.getLeaseholder(),
+ meta.getStartTime().longValue(),
+ commit,
+ replicationGroupIds,
+ txId,
+ commitTimestamp,
+ txFinishFuture
+ ))
+ .handle((res, ex) -> {
+ if (ex != null) {
+ if (ex.getCause() != null && ex.getCause() instanceof
TransactionException) {
+ TransactionException cause =
(TransactionException) ex.getCause();
+
+ if (cause.code() == TX_WAS_ABORTED_ERR) {
+ updateTxMeta(txId, old -> new
TxStateMeta(ABORTED, old.txCoordinatorId(), null));
+
+ return
CompletableFuture.<Void>failedFuture(cause);
+ }
+ }
+ LOG.warn("Failed to finish Tx {}. The operation will
be retried.", ex, txId);
Review Comment:
Do not forget about the log message rules.
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxManager.java:
##########
@@ -116,22 +115,27 @@ public interface TxManager extends IgniteComponent {
* transaction execution. The tracker is also used to determine
the read timestamp for read-only transactions. Each client
* should pass its own tracker to provide linearizability between
read-write and read-only transactions started by this client.
* @param commitPartition Partition to store a transaction state.
- * @param recipientNode Recipient node.
- * @param term Raft term.
- * @param commit {@code True} if a commit requested.
+ * @param commit {@code true} if a commit requested.
* @param enlistedGroups Enlisted partition groups with consistency token.
* @param txId Transaction id.
*/
CompletableFuture<Void> finish(
HybridTimestampTracker timestampTracker,
TablePartitionId commitPartition,
- ClusterNode recipientNode,
- Long term,
boolean commit,
Map<TablePartitionId, Long> enlistedGroups,
UUID txId
);
+
+ /**
+ * Make sure the state of the provided transaction is updated correctly.
+ *
+ * @param commit {@code true} if a commit requested.
+ * @param txId Transaction id.
+ */
+ CompletableFuture<Void> finish(boolean commit, UUID txId);
Review Comment:
It is not an overloded finish method. It would be better to rename it
(_finishEmpty_, for example).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
