IGNITE-24963 Final cleanup + WW

ascherbakoff · ascherbakoff · commit c9b36523cf2a · 2026-04-06T11:02:14.000+03:00
diff --git a/modules/api/src/main/java/org/apache/ignite/tx/RunInTransactionInternalImpl.java b/modules/api/src/main/java/org/apache/ignite/tx/RunInTransactionInternalImpl.java
@@ -30,7 +30,6 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 import java.util.function.Function;
 import org.jetbrains.annotations.Nullable;
 
diff --git a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/exception/ReplicationException.java b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/exception/ReplicationException.java
@@ -23,7 +23,6 @@
 import org.apache.ignite.internal.lang.IgniteInternalException;
 import org.apache.ignite.internal.replicator.ReplicationGroupId;
 import org.apache.ignite.tx.RetriableReplicaRequestException;
-import org.apache.ignite.tx.RetriableTransactionException;
 
 /**
  * The exception is thrown when some issue happened during a replication.
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItDataConsistencyTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItDataConsistencyTest.java
@@ -156,7 +156,7 @@ public void testDataConsistency() throws InterruptedException {
             readThread.join(3_000);
         }
 
-        // TODO unregisted from timeout tracker killed transactions!!!!
+        // TODO IGNITE-28464 unregister from expiration tracker.
         validate();
     }
 
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionInflights.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionInflights.java
@@ -17,12 +17,13 @@
 
 package org.apache.ignite.internal.table.distributed.replicator;
 
+import static java.util.concurrent.atomic.AtomicLongFieldUpdater.newUpdater;
 import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
 
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 import java.util.function.Predicate;
 import org.apache.ignite.internal.partition.replicator.network.replication.RequestType;
 import org.jetbrains.annotations.Nullable;
@@ -35,6 +36,9 @@ public class PartitionInflights {
     /** Hint for maximum concurrent txns. */
     private static final int MAX_CONCURRENT_TXNS_HINT = 1024;
 
+    /** Field updater for inflights. */
+    private static final AtomicLongFieldUpdater<CleanupContext> UPDATER = newUpdater(CleanupContext.class, "inflights");
+
     /** Txn contexts. */
     private final ConcurrentHashMap<UUID, CleanupContext> txCtxMap = new ConcurrentHashMap<>(MAX_CONCURRENT_TXNS_HINT);
 
@@ -58,7 +62,7 @@ public class PartitionInflights {
             if (ctx.finishFut != null || testPred.test(txId)) {
                 res[0] = false;
             } else {
-                ctx.inflights.incrementAndGet();
+                UPDATER.incrementAndGet(ctx);
                 if (requestType.isWrite()) {
                     ctx.hasWrites = true;
                 }
@@ -76,7 +80,7 @@ public class PartitionInflights {
      * @param ctx Cleanup context.
      */
     static void removeInflight(CleanupContext ctx) {
-        long val = ctx.inflights.decrementAndGet();
+        long val = UPDATER.decrementAndGet(ctx);
 
         if (ctx.finishFut != null && val == 0) {
             ctx.finishFut.complete(null);
@@ -96,11 +100,11 @@ static void removeInflight(CleanupContext ctx) {
             }
 
             if (ctx.finishFut == null) {
-                ctx.finishFut = ctx.inflights.get() == 0 ? nullCompletedFuture() : new CompletableFuture<>();
+                ctx.finishFut = UPDATER.get(ctx) == 0 ? nullCompletedFuture() : new CompletableFuture<>();
             }
 
             // Avoiding a data race with a concurrent decrementing thread, which might not see finishFut publication.
-            if (ctx.inflights.get() == 0 && !ctx.finishFut.isDone()) {
+            if (UPDATER.get(ctx) == 0 && !ctx.finishFut.isDone()) {
                 ctx.finishFut = nullCompletedFuture();
             }
 
@@ -132,18 +136,8 @@ public boolean contains(UUID txId) {
      */
     public static class CleanupContext {
         volatile CompletableFuture<Void> finishFut;
-        AtomicLong inflights = new AtomicLong(0); // TODO atomic updater
+        volatile long inflights = 0;
         volatile boolean hasWrites = false;
-
-//        void addInflight() {
-//            inflights.incrementAndGet();
-//        }
-//
-//        void removeInflight(UUID txId) {
-//            //assert inflights > 0 : format("No inflights, cannot remove any [txId={}, ctx={}]", txId, this);
-//
-//            inflights.decrementAndGet();
-//        }
     }
 
     @TestOnly
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
@@ -263,8 +263,6 @@ public class PartitionReplicaListener implements ReplicaTableProcessor {
     /** Factory for creating replica command messages. */
     private static final ReplicaMessagesFactory REPLICA_MESSAGES_FACTORY = new ReplicaMessagesFactory();
 
-    private static final int CONCURRENCY = Runtime.getRuntime().availableProcessors();
-
     private final ZonePartitionId replicationGroupId;
 
     private final int tableId;
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/TraceableFuture.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/TraceableFuture.java
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
@@ -649,7 +649,7 @@ private <R> CompletableFuture<R> trackingInvoke(
             if (req.isWrite()) {
                 // Track only write requests from explicit transactions.
                 if (!tx.remote() && !transactionInflights.addInflight(tx.id())) {
-                    // TODO can add inflight even if the error
+                    // TODO IGNITE-28461 fail fast if TxContext.err != null.
                     return failedFuture(tx.enlistFailedException());
                 }
 
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
@@ -408,6 +408,10 @@ public void testLockOrdering() throws InterruptedException {
         InternalTransaction tx3 = (InternalTransaction) igniteTransactions.begin();
         InternalTransaction tx4 = (InternalTransaction) igniteTransactions.begin();
 
+        assertTrue(tx3.id().compareTo(tx4.id()) < 0);
+        assertTrue(tx2.id().compareTo(tx3.id()) < 0);
+        assertTrue(tx1.id().compareTo(tx2.id()) < 0);
+
         boolean reversed = txManager(accounts).lockManager().policy().reverse();
         if (reversed) {
             InternalTransaction tmp = tx1;
@@ -419,10 +423,6 @@ public void testLockOrdering() throws InterruptedException {
             tx3 = tmp;
         }
 
-        assertTrue(tx3.id().compareTo(tx4.id()) < 0);
-        assertTrue(tx2.id().compareTo(tx3.id()) < 0);
-        assertTrue(tx1.id().compareTo(tx2.id()) < 0);
-
         RecordView<Tuple> acc0 = accounts.recordView();
         RecordView<Tuple> acc2 = accounts.recordView();
         RecordView<Tuple> acc3 = accounts.recordView();
@@ -978,34 +978,25 @@ public void testGetAllAbort() throws TransactionException {
     public void testGetAllConflict() throws Exception {
         accounts.recordView().upsertAll(null, List.of(makeValue(1, 100.), makeValue(2, 200.)));
 
-        InternalTransaction tx1 = (InternalTransaction) igniteTransactions.begin();
-        InternalTransaction tx2 = (InternalTransaction) igniteTransactions.begin();
+        InternalTransaction older = (InternalTransaction) igniteTransactions.begin();
+        InternalTransaction younger = (InternalTransaction) igniteTransactions.begin();
 
         boolean reversed = txManager(accounts).lockManager().policy().reverse();
 
-        Transaction owner = reversed ? tx2 : tx1;
-        Transaction waiter = reversed ? tx1 : tx2;
-
         RecordView<Tuple> txAcc = accounts.recordView();
         RecordView<Tuple> txAcc2 = accounts.recordView();
 
-        txAcc2.upsert(owner, makeValue(1, 300.));
-        txAcc.upsert(waiter, makeValue(2, 400.));
-
-        CompletableFuture<List<Tuple>> fut = txAcc.getAllAsync(waiter, List.of(makeKey(2), makeKey(1)));
-        ensureFutureNotCompleted(fut, 100);
+        txAcc2.upsert(older, makeValue(1, 300.));
+        txAcc.upsert(younger, makeValue(2, 400.));
 
-        validateBalance(txAcc2.getAll(owner, List.of(makeKey(2), makeKey(1))), 200., 300.);
-        validateBalance(txAcc2.getAll(owner, List.of(makeKey(1), makeKey(2))), 300., 200.);
+        // Triggers a conflict, which invalidates younger transaction.
+        txAcc.getAllAsync(reversed ? younger : older, List.of(makeKey(2), makeKey(1)));
+        assertTrue(waitForCondition(() -> TxState.ABORTED == younger.state(), 5_000), younger.state().toString());
 
-        assertTrue(waitForCondition(() -> TxState.ABORTED == tx2.state(), 5_000), tx2.state().toString());
+        validateBalance(txAcc2.getAll(older, List.of(makeKey(2), makeKey(1))), 200., 300.);
+        validateBalance(txAcc2.getAll(older, List.of(makeKey(1), makeKey(2))), 300., 200.);
 
-        owner.commit();
-        try {
-            waiter.rollback();
-        } catch (TransactionException e) {
-            // Expected.
-        }
+        older.commit();
 
         validateBalance(accounts.recordView().getAll(null, List.of(makeKey(2), makeKey(1))), 200., 300.);
     }
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TransactionKilledException.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TransactionKilledException.java
@@ -45,14 +45,6 @@ public TransactionKilledException(UUID txId, TxManager txManager) {
         this.txId = txId;
     }
 
-    public TransactionKilledException(UUID txId) {
-        super(
-                TX_KILLED_ERR,
-                "Transaction is killed " + txId
-        );
-        this.txId = txId;
-    }
-
     /**
      * Returns a transaction id.
      *
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
@@ -33,8 +33,6 @@
 import org.apache.ignite.internal.tx.impl.RemotelyTriggeredResourceRegistry;
 import org.apache.ignite.internal.tx.metrics.ResourceVacuumMetrics;
 import org.apache.ignite.internal.tx.metrics.TransactionMetricsSource;
-import org.apache.ignite.tx.Transaction;
-import org.apache.ignite.tx.TransactionOptions;
 import org.jetbrains.annotations.Nullable;
 import org.jetbrains.annotations.TestOnly;
 
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/HeapLockManager.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/HeapLockManager.java
@@ -77,12 +77,12 @@
  *
  * <p>Lock waiters are placed in the queue, ordered according to transaction priority: older transactions are first.
  * When a new waiter is placed in the queue, it's validated against current lock owners: if a waiter is not allowed to wait,
- * according to the {@link HeapLockManager#deadlockPreventionPolicy}, lock request is denied.
+ * according to the {@link HeapLockManager#deadlockPreventionPolicy}, lock request is denied or current owner is invalidated.
  *
- * When an owner is removed from the queue (on lock release), first we try to lock anything possible in the first pass, in the second pass
- * fail conflicting waiters.
+ * <p>When an owner is removed from the queue (on lock release), first try locking anything possible.
+ * In the second conflicts, which can appear on first path, are resolved.
  *
- * <p>Additionally limits the lock map size.
+ * <p>Lock table size is limited and implicitly defines the maximum size of a transaction.
  */
 public class HeapLockManager extends AbstractEventProducer<LockEvent, LockEventParameters> implements LockManager {
     /** Table size. */
@@ -923,7 +923,7 @@ IgniteBiTuple<CompletableFuture<Void>, LockMode> tryAcquire(UUID txId, LockMode
 
             synchronized (waiters) {
                 if (!isUsed()) {
-                    return new IgniteBiTuple<>(null, lockMode); // TODO ugly
+                    return new IgniteBiTuple<>(null, lockMode);
                 }
 
                 // We always replace the previous waiter with the new one. If the previous waiter has lock intention then incomplete
@@ -989,7 +989,7 @@ private List<Runnable> tryAcquireInternal(WaiterImpl waiter, boolean track, bool
                     // Notify once on first found conflict.
                     notified[0] = true;
                     if (notifyListeners(waiter.txId())) {
-                        // If there is an abandoned owner, fail waiter. TODO ticket
+                        // If there is an abandoned owner, fail waiter. TODO IGNITE-28458 wait instead of failing.
                         failWaiter(waiter, notifications, createLockException(waiter, owner, true));
 
                         return true;
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
@@ -1036,35 +1036,27 @@ public void processDelayedAck(Object ignored, @Nullable Throwable err) {
 
     @Override
     public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
-//        var deadlockPreventionPolicy = new WoundWaitDeadlockPreventionPolicy() {
-//            @Override
-//            public long waitTimeout() {
-//                return DEFAULT_LOCK_TIMEOUT;
-//            }
-//
-//            @Override
-//            public void failAction(UUID owner) {
-//                // TODO IGNITE-28447 sendTxRecoveryMessage and delete locks.
-//                TxStateMeta state = txStateVolatileStorage.state(owner);
-//                if (state == null || state.txCoordinatorId() == null) {
-//                    return; // Tx state is invalid. Locks will be cleaned up by tx recovery process.
-//                }
-//
-//                InternalClusterNode coordinator = topologyService.getById(state.txCoordinatorId());
-//                if (coordinator == null) {
-//                    return; // Tx is abandoned. Locks will be cleaned up by tx recovery process.
-//                }
-//
-//                txMessageSender.kill(coordinator, owner);
-//            }
-//        };
-
-        // This commented section is left intentionally.
-        var deadlockPreventionPolicy = new WaitDieDeadlockPreventionPolicy() {
+        var deadlockPreventionPolicy = new WoundWaitDeadlockPreventionPolicy() {
             @Override
             public long waitTimeout() {
                 return DEFAULT_LOCK_TIMEOUT;
             }
+
+            @Override
+            public void failAction(UUID owner) {
+                // TODO IGNITE-28447 sendTxRecoveryMessage and delete locks.
+                TxStateMeta state = txStateVolatileStorage.state(owner);
+                if (state == null || state.txCoordinatorId() == null) {
+                    return; // Tx state is invalid. Locks will be cleaned up by tx recovery process.
+                }
+
+                InternalClusterNode coordinator = topologyService.getById(state.txCoordinatorId());
+                if (coordinator == null) {
+                    return; // Tx is abandoned. Locks will be cleaned up by tx recovery process.
+                }
+
+                txMessageSender.kill(coordinator, owner);
+            }
         };
 
         txStateVolatileStorage.start();
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/WoundWaitDeadlockPreventionPolicy.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/WoundWaitDeadlockPreventionPolicy.java
@@ -23,7 +23,15 @@
 import org.apache.ignite.internal.tx.Waiter;
 
 /**
- * Wound-wait prevention policy. TODO desc.
+ * Implements a deadlock prevention policy that resolves conflicts between two transactions (tx1 and tx2) contending for the same key. When
+ * tx1 holds a lock and tx2 attempts to acquire it, the policy allows tx2 to wait for the lock if any of the following conditions are
+ * met:
+ * <ul>
+ *     <li>tx2 is younger than tx1.</li>
+ *     <li>tx2 is older than tx1 but has a lower {@link org.apache.ignite.internal.tx.TxPriority}.</li>
+ *     <li>The wait timeout is greater than 0.</li>
+ * </ul>
+ * If none of these conditions are met, tx1 is killed to prevent deadlock.
  */
 public class WoundWaitDeadlockPreventionPolicy implements DeadlockPreventionPolicy {
     private static final TxIdPriorityComparator TX_ID_PRIORITY_COMPARATOR = new TxIdPriorityComparator();
diff --git a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/AbstractDeadlockPreventionTest.java b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/AbstractDeadlockPreventionTest.java
@@ -36,7 +36,6 @@
 
 /**
  * Abstract class containing some tests for deadlock prevention that check common scenarios for different policies.
- * TODO move all single keys tests to heap lm test as they cant produce deadlock.
  */
 public abstract class AbstractDeadlockPreventionTest extends AbstractLockingTest {
     protected abstract Matcher<CompletableFuture<Lock>> conflictMatcher(UUID txId);
diff --git a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/AbstractLockManagerTest.java b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/AbstractLockManagerTest.java
diff --git a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/HeapLockManagerTest.java b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/HeapLockManagerTest.java
@@ -63,7 +63,6 @@
 import org.apache.ignite.internal.tx.test.TestTransactionIds;
 import org.apache.ignite.lang.IgniteException;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.params.provider.MethodSource;
 
 /**
  * Test class for {@link HeapLockManager}.
diff --git a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/WoundWaitDeadlockPreventionRollbackFailActionTest.java b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/WoundWaitDeadlockPreventionRollbackFailActionTest.java
@@ -19,16 +19,9 @@
 
 import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willSucceedFast;
 import static org.hamcrest.MatcherAssert.assertThat;
-import static org.junit.jupiter.api.Assertions.assertFalse;
 
 import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
 import org.apache.ignite.internal.tx.impl.WoundWaitDeadlockPreventionPolicy;
-import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 
 /**
diff --git a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/OrphanDetectorTest.java b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/OrphanDetectorTest.java
@@ -33,7 +33,6 @@
 import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.when;
 
-import java.lang.reflect.InvocationTargetException;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicInteger;
diff --git a/modules/transactions/src/testFixtures/java/org/apache/ignite/internal/tx/test/LockConflictMatcher.java b/modules/transactions/src/testFixtures/java/org/apache/ignite/internal/tx/test/LockConflictMatcher.java
@@ -29,9 +29,11 @@
 import org.hamcrest.Description;
 import org.hamcrest.TypeSafeMatcher;
 
+/**
+ * Validates if a lock operations ends with a conflict with expected transaction.
+ */
 public class LockConflictMatcher extends TypeSafeMatcher<CompletableFuture<Lock>> {
     private final UUID conflictId;
-    private CompletableFuture<Lock> item;
 
     private LockConflictMatcher(UUID txId) {
         this.conflictId = txId;
@@ -40,7 +42,6 @@ private LockConflictMatcher(UUID txId) {
     @Override
     protected boolean matchesSafely(CompletableFuture<Lock> item) {
         try {
-            this.item = item;
             item.get(100, TimeUnit.MILLISECONDS);
             return false; // Exception is expected.
         } catch (InterruptedException | TimeoutException e) {
diff --git a/modules/transactions/src/testFixtures/java/org/apache/ignite/internal/tx/test/LockFutureMatcher.java b/modules/transactions/src/testFixtures/java/org/apache/ignite/internal/tx/test/LockFutureMatcher.java
diff --git a/modules/transactions/src/testFixtures/java/org/apache/ignite/internal/tx/test/LockWaiterMatcher.java b/modules/transactions/src/testFixtures/java/org/apache/ignite/internal/tx/test/LockWaiterMatcher.java

Original file line number	Diff line number	Diff line change
`@@ -156,7 +156,7 @@ public void testDataConsistency() throws InterruptedException {`
`156`	`156`	`readThread.join(3_000);`
`157`	`157`	`}`
`158`	`158`
`159`		`- // TODO unregisted from timeout tracker killed transactions!!!!`
	`159`	`+ // TODO IGNITE-28464 unregister from expiration tracker.`
`160`	`160`	`validate();`
`161`	`161`	`}`
`162`	`162`
Original file line number	Diff line number	Diff line change
`@@ -649,7 +649,7 @@ private <R> CompletableFuture<R> trackingInvoke(`
`649`	`649`	`if (req.isWrite()) {`
`650`	`650`	`// Track only write requests from explicit transactions.`
`651`	`651`	`if (!tx.remote() && !transactionInflights.addInflight(tx.id())) {`
`652`		`- // TODO can add inflight even if the error`
	`652`	`+ // TODO IGNITE-28461 fail fast if TxContext.err != null.`
`653`	`653`	`return failedFuture(tx.enlistFailedException());`
`654`	`654`	`}`
`655`	`655`
Original file line number	Diff line number	Diff line change
`@@ -45,14 +45,6 @@ public TransactionKilledException(UUID txId, TxManager txManager) {`
`45`	`45`	`this.txId = txId;`
`46`	`46`	`}`
`47`	`47`
`48`		`- public TransactionKilledException(UUID txId) {`
`49`		`- super(`
`50`		`- TX_KILLED_ERR,`
`51`		`- "Transaction is killed " + txId`
`52`		`- );`
`53`		`- this.txId = txId;`
`54`		`- }`
`55`		`-`
`56`	`48`	`/**`
`57`	`49`	`* Returns a transaction id.`
`58`	`50`	`*`