IGNITE-28385 Fix missing enlistment in client commit request handler (#7899)

Co-authored-by: Kirill Sizov <sizov.kirill.y@gmail.com>

Author: Cyrill

modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionCommitRequest.java

@@ -172,8 +172,11 @@ public static boolean merge(InternalTable table, int partId, String consistentId
         if (existing == null) {
             tx.enlist(replicationGroupId, table.tableId(), consistentId, token);
         } else {
+            boolean tokenMatch = existing.consistencyToken() == token;
+            existing.addTableId(table.tableId());
+
             // Enlistment tokens should be equal on commit.
-            return !commit || existing.consistencyToken() == token;
+            return !commit || tokenMatch;
         }
 
         return true;
@@ -207,5 +210,10 @@ String consistentId() {
         long token() {
             return token;
         }
+
+        @Override
+        public String toString() {
+            return "(tableId=" + tableId + ", partId=" + partitionId + ", node=" + consistentId + ")";
+        }
     }
 }

modules/client/src/test/java/org/apache/ignite/client/fakes/FakeTxManager.java

@@ -272,7 +272,13 @@ public CompletableFuture<Void> cleanup(
     }
 
     @Override
-    public CompletableFuture<Void> cleanup(ZonePartitionId commitPartitionId, String node, UUID txId) {
+    public CompletableFuture<Void> cleanup(
+            ZonePartitionId commitPartitionId,
+            String node,
+            UUID txId,
+            boolean commit,
+            @Nullable HybridTimestamp commitTimestamp
+    ) {
         return nullCompletedFuture();
     }
 

modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/TxFinishReplicaRequestHandler.java

@@ -224,10 +224,13 @@ private CompletableFuture<TransactionResult> finishAndCleanup(
                 .map(entry -> new EnlistedPartitionGroup(entry.getKey(), entry.getValue().tableIds()))
                 .collect(toList());
         return finishTransaction(enlistedPartitionGroups, txId, commit, commitTimestamp)
-                .thenCompose(txResult ->
-                    txManager.cleanup(replicationGroupId, enlistedPartitions, commit, commitTimestamp, txId)
-                            .thenApply(v -> txResult)
-                );
+                .thenCompose(txResult -> {
+                    boolean actualCommit = txResult.transactionState() == COMMITTED;
+                    HybridTimestamp actualCommitTs = txResult.commitTimestamp();
+
+                    return txManager.cleanup(replicationGroupId, enlistedPartitions, actualCommit, actualCommitTs, txId)
+                            .thenApply(v -> txResult);
+                });
     }
 
     private static void throwIfSchemaValidationOnCommitFailed(CompatValidationResult validationResult, TransactionResult txResult) {

modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/TxRecoveryMessageHandler.java

@@ -17,6 +17,7 @@
 
 package org.apache.ignite.internal.partition.replicator.handlers;
 
+import static org.apache.ignite.internal.tx.TxState.COMMITTED;
 import static org.apache.ignite.internal.tx.TxState.isFinalState;
 
 import java.util.UUID;
@@ -73,7 +74,8 @@ public CompletableFuture<?> handle(TxRecoveryMessage request, UUID senderId) {
         // Check whether a transaction has already been finished.
         if (txMeta != null && isFinalState(txMeta.txState())) {
             // Tx recovery message is processed on the commit partition.
-            return txRecoveryEngine.runCleanupOnNode(replicationGroupId, txId, senderId);
+            boolean commit = txMeta.txState() == COMMITTED;
+            return txRecoveryEngine.runCleanupOnNode(replicationGroupId, txId, senderId, commit, txMeta.commitTimestamp());
         }
 
         LOG.info(

modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java

@@ -2050,7 +2050,7 @@ private static void configureTxManager(TxManager txManager) {
         doAnswer(invocation -> nullCompletedFuture())
                 .when(txManager).finish(any(), any(), anyBoolean(), any(), anyBoolean(), anyBoolean(), any(), any());
         doAnswer(invocation -> nullCompletedFuture())
-                .when(txManager).cleanup(any(), anyString(), any());
+                .when(txManager).cleanup(any(), anyString(), any(), anyBoolean(), any());
     }
 
     private void testWritesAreSuppliedWithRequiredCatalogVersion(RequestType requestType, RwListenerInvocation listenerInvocation) {

modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/ZonePartitionReplicaListenerTest.java

@@ -991,7 +991,7 @@ private static void configureTxManager(TxManager txManager) {
         doAnswer(invocation -> nullCompletedFuture())
                 .when(txManager).finish(any(), any(), anyBoolean(), any(), anyBoolean(), anyBoolean(), any(), any());
         doAnswer(invocation -> nullCompletedFuture())
-                .when(txManager).cleanup(any(), anyString(), any());
+                .when(txManager).cleanup(any(), anyString(), any(), anyBoolean(), any());
     }
 
     private void upsertInNewTxFor(TestKey key) {

modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxManager.java

@@ -263,9 +263,17 @@ CompletableFuture<Void> cleanup(
      * @param commitPartitionId Commit partition id.
      * @param node Target node.
      * @param txId Transaction id.
+     * @param commit Whether the transaction was committed.
+     * @param commitTimestamp Commit timestamp, if committed.
      * @return Completable future of Void.
      */
-    CompletableFuture<Void> cleanup(ZonePartitionId commitPartitionId, String node, UUID txId);
+    CompletableFuture<Void> cleanup(
+            ZonePartitionId commitPartitionId,
+            String node,
+            UUID txId,
+            boolean commit,
+            @Nullable HybridTimestamp commitTimestamp
+    );
 
     /**
      * Locally vacuums no longer needed transactional resources, like txnState both persistent and volatile.

modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxCleanupRequestSender.java

@@ -188,15 +188,23 @@ private void markTxnCleanupReplicated(UUID txId, TxState state, ZonePartitionId
     }
 
     /**
-     * Sends unlock request to the nodes than initiated recovery.
+     * Sends cleanup request to the node that initiated recovery.
      *
      * @param commitPartitionId Commit partition id.
      * @param node Target node.
      * @param txId Transaction id.
+     * @param commit Whether the transaction was committed.
+     * @param commitTimestamp Commit timestamp, if committed.
      * @return Completable future of Void.
      */
-    public CompletableFuture<Void> cleanup(ZonePartitionId commitPartitionId, String node, UUID txId) {
-        return sendCleanupMessageWithRetries(commitPartitionId, false, null, txId, node, null, RETRY_INITIAL_TIMEOUT_MS, 0);
+    public CompletableFuture<Void> cleanup(
+            ZonePartitionId commitPartitionId,
+            String node,
+            UUID txId,
+            boolean commit,
+            @Nullable HybridTimestamp commitTimestamp
+    ) {
+        return sendCleanupMessageWithRetries(commitPartitionId, commit, commitTimestamp, txId, node, null, RETRY_INITIAL_TIMEOUT_MS, 0);
     }
 
     /**

modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java

@@ -1163,8 +1163,14 @@ public CompletableFuture<Void> cleanup(
     }
 
     @Override
-    public CompletableFuture<Void> cleanup(ZonePartitionId commitPartitionId, String node, UUID txId) {
-        return txCleanupRequestSender.cleanup(commitPartitionId, node, txId);
+    public CompletableFuture<Void> cleanup(
+            ZonePartitionId commitPartitionId,
+            String node,
+            UUID txId,
+            boolean commit,
+            @Nullable HybridTimestamp commitTimestamp
+    ) {
+        return txCleanupRequestSender.cleanup(commitPartitionId, node, txId, commit, commitTimestamp);
     }
 
     @Override

modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxRecoveryEngine.java

@@ -22,6 +22,7 @@
 import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
 import static org.apache.ignite.internal.tx.TransactionLogUtils.formatTxInfo;
 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.isFinalState;
 import static org.apache.ignite.internal.tx.TxStateMetaFinishing.castToFinishing;
@@ -33,6 +34,7 @@
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Function;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
 import org.apache.ignite.internal.hlc.HybridTimestampTracker;
 import org.apache.ignite.internal.network.ClusterNodeResolver;
 import org.apache.ignite.internal.network.InternalClusterNode;
@@ -127,10 +129,20 @@ public CompletableFuture<TransactionMeta> triggerTxRecovery(
                 })
                 .thenCompose(Function.identity())
                 .whenComplete((v, ex) -> {
-                    runCleanupOnNode(commitPartitionId, txId, commitPartitionNode);
+                    // Cleanup must use the actual resolved tx state; using commit=false unconditionally
+                    // would corrupt data when the tx is actually COMMITTED (abort cleanup races with
+                    // commit cleanup, producing a mix of committed and aborted rows).
+                    boolean commit = v != null && v.txState() == COMMITTED;
+                    @Nullable HybridTimestamp commitTs = v != null ? v.commitTimestamp() : null;
+
+                    runCleanupOnNode(commitPartitionId, txId, commitPartitionNode, commit, commitTs);
 
                     if (senderGroupId != null && senderId != null) {
-                        runCleanupOnNode(senderGroupId, txId, senderId);
+                        String senderConsistentId = clusterNodeResolver.getConsistentIdById(senderId);
+
+                        if (senderConsistentId != null) {
+                            runCleanupOnNode(senderGroupId, txId, senderConsistentId, commit, commitTs);
+                        }
                     }
                 });
     }
@@ -163,13 +175,21 @@ private static PendingTxPartitionEnlistment createAbandonedTxRecoveryEnlistment(
      *
      * @param groupId Group id.
      * @param txId Transaction id.
-     * @param nodeId Node id (inconsistent).
+     * @param nodeId Node id (ephemeral).
+     * @param commit Whether the transaction was committed.
+     * @param commitTimestamp Commit timestamp, if committed.
      */
-    public CompletableFuture<Void> runCleanupOnNode(ZonePartitionId groupId, UUID txId, UUID nodeId) {
-        // Get node id of the sender to send back cleanup requests.
+    public CompletableFuture<Void> runCleanupOnNode(
+            ZonePartitionId groupId,
+            UUID txId,
+            UUID nodeId,
+            boolean commit,
+            @Nullable HybridTimestamp commitTimestamp
+    ) {
         String nodeConsistentId = clusterNodeResolver.getConsistentIdById(nodeId);
 
-        return nodeConsistentId == null ? nullCompletedFuture() : runCleanupOnNode(groupId, txId, nodeConsistentId);
+        return nodeConsistentId == null ? nullCompletedFuture()
+                : runCleanupOnNode(groupId, txId, nodeConsistentId, commit, commitTimestamp);
     }
 
     /**
@@ -178,8 +198,16 @@ public CompletableFuture<Void> runCleanupOnNode(ZonePartitionId groupId, UUID tx
      * @param commitPartitionId Commit partition id.
      * @param txId Transaction id.
      * @param nodeName Node consistent id.
+     * @param commit Whether the transaction was committed.
+     * @param commitTimestamp Commit timestamp, if committed.
      */
-    private CompletableFuture<Void> runCleanupOnNode(ZonePartitionId commitPartitionId, UUID txId, String nodeName) {
-        return txManager.cleanup(commitPartitionId, nodeName, txId);
+    private CompletableFuture<Void> runCleanupOnNode(
+            ZonePartitionId commitPartitionId,
+            UUID txId,
+            String nodeName,
+            boolean commit,
+            @Nullable HybridTimestamp commitTimestamp
+    ) {
+        return txManager.cleanup(commitPartitionId, nodeName, txId, commit, commitTimestamp);
     }
 }