IGNITE-24242 Add RO/RW Transaction Timeout configuration schema

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

@@ -164,6 +164,11 @@ public CompletableFuture<Void> finish(boolean commit, HybridTimestamp executionT
             public boolean isFinishingOrFinished() {
                 return false;
             }
+
+            @Override
+            public long timeout() {
+                return 10_000;
+            }
         };
     }
 

modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/TableScanNodeExecutionTest.java

@@ -272,7 +272,6 @@ private static class TestInternalTableImpl extends InternalTableImpl {
                     timestampTracker,
                     mock(PlacementDriver.class),
                     mock(TransactionInflights.class),
-                    3_000,
                     0,
                     null,
                     mock(StreamerReceiverRunner.class)

modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/NoOpTransaction.java

@@ -181,6 +181,11 @@ public boolean isFinishingOrFinished() {
         return commitFut.isDone() || rollbackFut.isDone();
     }
 
+    @Override
+    public long timeout() {
+        return 10_000;
+    }
+
     @Override
     public IgniteBiTuple<ClusterNode, Long> enlist(TablePartitionId tablePartitionId,
             IgniteBiTuple<ClusterNode, Long> nodeAndConsistencyToken) {

modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxAbstractDistributedTestSingleNode.java

@@ -34,6 +34,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
 import org.apache.ignite.internal.table.TxAbstractTest;
 import org.apache.ignite.internal.tx.impl.ReadWriteTransactionImpl;
 import org.apache.ignite.internal.util.CollectionUtils;
@@ -205,7 +206,7 @@ public void testImplicitTransactionTimeout() {
 
         assertFalse(implicitOpFut.isDone());
 
-        assertThat(implicitOpFut, willThrow(TransactionException.class));
+        assertThat(implicitOpFut, willThrow(TransactionException.class, 40_000, TimeUnit.SECONDS));
 
         assertNull(rv.get(null, makeKey(1)));
     }

modules/table/src/integrationTest/java/org/apache/ignite/internal/table/ItColocationTest.java

@@ -346,7 +346,6 @@ public NetworkMessage clone() {
                 observableTimestampTracker,
                 new TestPlacementDriver(clusterNode),
                 transactionInflights,
-                3_000,
                 0,
                 null,
                 mock(StreamerReceiverRunner.class)

modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java

@@ -418,8 +418,6 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
 
     private final PartitionReplicaLifecycleManager partitionReplicaLifecycleManager;
 
-    private long implicitTransactionTimeout;
-
     private int attemptsObtainLock;
 
     @Nullable
@@ -679,7 +677,6 @@ public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
 
             partitionReplicatorNodeRecovery.start();
 
-            implicitTransactionTimeout = txCfg.implicitTransactionTimeout().value();
             attemptsObtainLock = txCfg.attemptsObtainLock().value();
 
             executorInclinedPlacementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_EXPIRED, this::onPrimaryReplicaExpired);
@@ -1628,7 +1625,6 @@ private TableImpl createTableImpl(
                 observableTimestampTracker,
                 executorInclinedPlacementDriver,
                 transactionInflights,
-                implicitTransactionTimeout,
                 attemptsObtainLock,
                 this::streamerFlushExecutor,
                 Objects.requireNonNull(streamerReceiverRunner)

modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java

@@ -150,6 +150,8 @@ public class InternalTableImpl implements InternalTable {
     /** Replica messages factory. */
     private static final ReplicaMessagesFactory REPLICA_MESSAGES_FACTORY = new ReplicaMessagesFactory();
 
+    public static final int DEFAULT_RW_TIMEOUT = 10_000;
+
     /** Partitions. */
     private final int partitions;
 
@@ -198,9 +200,6 @@ public class InternalTableImpl implements InternalTable {
     /** Map update guarded by {@link #updatePartitionMapsMux}. */
     private volatile Int2ObjectMap<PendingComparableValuesTracker<Long, Void>> storageIndexTrackerByPartitionId = emptyMap();
 
-    /** Implicit transaction timeout. */
-    private final long implicitTransactionTimeout;
-
     /** Attempts to take lock. */
     private final int attemptsObtainLock;
 
@@ -218,7 +217,6 @@ public class InternalTableImpl implements InternalTable {
      * @param clockService A hybrid logical clock service.
      * @param placementDriver Placement driver.
      * @param transactionInflights Transaction inflights.
-     * @param implicitTransactionTimeout Implicit transaction timeout.
      * @param attemptsObtainLock Attempts to take lock.
      */
     public InternalTableImpl(
@@ -234,7 +232,6 @@ public InternalTableImpl(
             HybridTimestampTracker observableTimestampTracker,
             PlacementDriver placementDriver,
             TransactionInflights transactionInflights,
-            long implicitTransactionTimeout,
             int attemptsObtainLock,
             Supplier<ScheduledExecutorService> streamerFlushExecutor,
             StreamerReceiverRunner streamerReceiverRunner
@@ -251,7 +248,6 @@ public InternalTableImpl(
         this.observableTimestampTracker = observableTimestampTracker;
         this.placementDriver = placementDriver;
         this.transactionInflights = transactionInflights;
-        this.implicitTransactionTimeout = implicitTransactionTimeout;
         this.attemptsObtainLock = attemptsObtainLock;
         this.streamerFlushExecutor = streamerFlushExecutor;
         this.streamerReceiverRunner = streamerReceiverRunner;
@@ -368,9 +364,12 @@ private <R> CompletableFuture<R> enlistInTx(
         return postEnlist(fut, false, actualTx, actualTx.implicit()).handle((r, e) -> {
             if (e != null) {
                 if (actualTx.implicit()) {
+                    // TODO: IGNITE-24244
+                    long timeout = actualTx.isReadOnly() ? actualTx.timeout() : DEFAULT_RW_TIMEOUT;
+
                     long ts = (txStartTs == null) ? actualTx.startTimestamp().getPhysical() : txStartTs;
 
-                    if (exceptionAllowsImplicitTxRetry(e) && coarseCurrentTimeMillis() - ts < implicitTransactionTimeout) {
+                    if (canRetry(e, ts, timeout)) {
                         return enlistInTx(row, null, fac, noWriteChecker, ts);
                     }
                 }
@@ -382,6 +381,10 @@ private <R> CompletableFuture<R> enlistInTx(
         }).thenCompose(identity());
     }
 
+    private static boolean canRetry(Throwable e, long ts, long timeout) {
+        return exceptionAllowsImplicitTxRetry(e) && coarseCurrentTimeMillis() - ts < timeout;
+    }
+
     /**
      * Enlists a single row into a transaction.
      *
@@ -486,9 +489,12 @@ private <T> CompletableFuture<T> enlistInTx(
         return postEnlist(fut, actualTx.implicit() && !singlePart, actualTx, full).handle((r, e) -> {
             if (e != null) {
                 if (actualTx.implicit()) {
+                    // TODO: IGNITE-24244
+                    long timeout = actualTx.isReadOnly() ? actualTx.timeout() : 3_000;
+
                     long ts = (txStartTs == null) ? actualTx.startTimestamp().getPhysical() : txStartTs;
 
-                    if (exceptionAllowsImplicitTxRetry(e) && coarseCurrentTimeMillis() - ts < implicitTransactionTimeout) {
+                    if (canRetry(e, ts, timeout)) {
                         return enlistInTx(keyRows, null, fac, reducer, noOpChecker, ts);
                     }
                 }
@@ -1185,9 +1191,12 @@ private CompletableFuture<Void> updateAllWithRetry(
         // Will be finished in one RTT.
         return postEnlist(fut, false, tx, true).handle((r, e) -> {
             if (e != null) {
+                // TODO: IGNITE-24244
+                long timeout = tx.isReadOnly() ? tx.timeout() : 3_000;
+
                 long ts = (txStartTs == null) ? tx.startTimestamp().getPhysical() : txStartTs;
 
-                if (exceptionAllowsImplicitTxRetry(e) && coarseCurrentTimeMillis() - ts < implicitTransactionTimeout) {
+                if (canRetry(e, ts, timeout)) {
                     return updateAllWithRetry(rows, deleted, partition, ts);
                 }
 

modules/table/src/test/java/org/apache/ignite/internal/table/distributed/storage/InternalTableEstimatedSizeTest.java

@@ -210,7 +210,6 @@ void setUp(
                 placementDriver,
                 new TransactionInflights(placementDriver, clockService),
                 0,
-                0,
                 () -> null,
                 mock(StreamerReceiverRunner.class)
         );

modules/table/src/test/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImplTest.java

@@ -72,7 +72,6 @@ void testUpdatePartitionTrackers() {
                 HybridTimestampTracker.atomicTracker(null),
                 mock(PlacementDriver.class),
                 mock(TransactionInflights.class),
-                3_000,
                 0,
                 null,
                 mock(StreamerReceiverRunner.class)
@@ -122,7 +121,6 @@ void testRowBatchByPartitionId() {
                 HybridTimestampTracker.atomicTracker(null),
                 mock(PlacementDriver.class),
                 mock(TransactionInflights.class),
-                3_000,
                 0,
                 null,
                 null

modules/table/src/testFixtures/java/org/apache/ignite/distributed/ItTxTestCluster.java

@@ -647,7 +647,6 @@ public TableViewInternal startTable(String tableName, SchemaDescriptor schemaDes
                 timestampTracker,
                 placementDriver,
                 clientTransactionInflights,
-                500,
                 0,
                 null,
                 mock(StreamerReceiverRunner.class)

modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java

@@ -268,7 +268,6 @@ public DummyInternalTableImpl(
                 tracker,
                 placementDriver,
                 transactionInflights,
-                3_000,
                 0,
                 null,
                 mock(StreamerReceiverRunner.class)

modules/transactions/src/integrationTest/java/org/apache/ignite/tx/distributed/ItTransactionPrimaryChangeTest.java

@@ -85,7 +85,8 @@ protected void customizeInitParameters(InitParametersBuilder builder) {
 
         builder.clusterConfiguration("ignite {"
                 + "  transaction: {"
-                + "      implicitTransactionTimeout: 30000,"
+                + "      readOnlyTimeout: 30000,"
+                + "      readWriteTimeout: 30000,"
                 + "      txnResourceTtl: 2"
                 + "  },"
                 + "  replication: {"

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

@@ -123,4 +123,11 @@ public interface InternalTransaction extends Transaction {
      * @return Whether the transaction is finishing or finished
      */
     boolean isFinishingOrFinished();
+
+    /**
+     * Returns the transaction timeout in millis.
+     *
+     * @return The transaction timeout.
+     */
+    long timeout();
 }

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

@@ -28,23 +28,20 @@
  */
 @Config
 public class TransactionConfigurationSchema {
-    /** Default checking transaction interval. */
-    public static final long DEFAULT_ABANDONED_CHECK_TS = 5_000;
-
     /** How often abandoned transactions are searched for (milliseconds). */
     @Range(min = 0)
     @Value(hasDefault = true)
-    public final long abandonedCheckTs = DEFAULT_ABANDONED_CHECK_TS;
+    public final long abandonedCheckTs = 5_000;
 
-    /** Default transaction timeout (milliseconds). */
+    /** Default timeout for read-only transactions. */
     @Range(min = 1)
     @Value(hasDefault = true)
-    public final long timeout = 10_000;
+    public final long readOnlyTimeout = TimeUnit.MINUTES.toMillis(10);
 
-    /** Timeout for implicit transactions (milliseconds). */
+    /** Default timeout for read-write transactions. */
     @Range(min = 1)
     @Value(hasDefault = true)
-    public final long implicitTransactionTimeout = 3_000;
+    public final long readWriteTimeout = TimeUnit.SECONDS.toMillis(30);
 
     /** A transaction tries to take lock several times until it throws an exception {@lonk org.apache.ignite.tx.TransactionException}. */
     @Range(min = 0)

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

@@ -55,6 +55,9 @@ public abstract class IgniteAbstractTransactionImpl implements InternalTransacti
     /** Implicit transaction flag. */
     private final boolean implicit;
 
+    /** Transaction timeout. */
+    protected final long timeout;
+
     /**
      * The constructor.
      *
@@ -63,19 +66,22 @@ public abstract class IgniteAbstractTransactionImpl implements InternalTransacti
      * @param observableTsTracker Observation timestamp tracker.
      * @param coordinatorId Transaction coordinator inconsistent ID.
      * @param implicit True for an implicit transaction, false for an ordinary one.
+     * @param timeout Transaction timeout in milliseconds.
      */
     public IgniteAbstractTransactionImpl(
             TxManager txManager,
             HybridTimestampTracker observableTsTracker,
             UUID id,
             UUID coordinatorId,
-            boolean implicit
+            boolean implicit,
+            long timeout
     ) {
         this.txManager = txManager;
         this.observableTsTracker = observableTsTracker;
         this.id = id;
         this.coordinatorId = coordinatorId;
         this.implicit = implicit;
+        this.timeout = timeout;
     }
 
     /** {@inheritDoc} */
@@ -145,4 +151,10 @@ private static Throwable tryToCopyExceptionWithCause(ExecutionException exceptio
 
         return copy;
     }
+
+    /** {@inheritDoc} */
+    @Override
+    public long timeout() {
+        return timeout;
+    }
 }

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

@@ -141,6 +141,11 @@ public boolean isFinishingOrFinished() {
         return transaction.isFinishingOrFinished();
     }
 
+    @Override
+    public long timeout() {
+        return transaction.timeout();
+    }
+
     @Override
     public <T> T unwrap(Class<T> classToUnwrap) {
         return classToUnwrap.cast(transaction);

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

@@ -59,10 +59,11 @@ public class ReadOnlyTransactionImpl extends IgniteAbstractTransactionImpl {
             UUID id,
             UUID txCoordinatorId,
             boolean implicit,
+            long timeout,
             HybridTimestamp readTimestamp,
             CompletableFuture<Void> txFuture
     ) {
-        super(txManager, observableTsTracker, id, txCoordinatorId, implicit);
+        super(txManager, observableTsTracker, id, txCoordinatorId, implicit, timeout);
 
         this.readTimestamp = readTimestamp;
         this.txFuture = txFuture;

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

@@ -73,9 +73,10 @@ public ReadWriteTransactionImpl(
             HybridTimestampTracker observableTsTracker,
             UUID id,
             UUID txCoordinatorId,
-            boolean implicit
+            boolean implicit,
+            long timeout
     ) {
-        super(txManager, observableTsTracker, id, txCoordinatorId, implicit);
+        super(txManager, observableTsTracker, id, txCoordinatorId, implicit, timeout);
     }
 
     /** {@inheritDoc} */

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

@@ -416,7 +416,11 @@ private InternalTransaction beginBusy(
         if (!readOnly) {
             txStateVolatileStorage.initialize(txId, localNodeId);
 
-            return new ReadWriteTransactionImpl(this, timestampTracker, txId, localNodeId, implicit);
+            // TODO: RW timeouts will be supported in https://issues.apache.org/jira/browse/IGNITE-24244
+            //  long timeout = options.timeoutMillis() == 0 ? txConfig.readWriteTimeout().value() : options.timeoutMillis();
+            long timeout = 3_000;
+
+            return new ReadWriteTransactionImpl(this, timestampTracker, txId, localNodeId, implicit, timeout);
         } else {
             return beginReadOnlyTransaction(timestampTracker, beginTimestamp, txId, implicit, options);
         }
@@ -447,7 +451,11 @@ private ReadOnlyTransactionImpl beginReadOnlyTransaction(
         try {
             CompletableFuture<Void> txFuture = new CompletableFuture<>();
 
-            var transaction = new ReadOnlyTransactionImpl(this, timestampTracker, txId, localNodeId, implicit, readTimestamp, txFuture);
+            long timeout = options.timeoutMillis() == 0 ? defaultReadOnlyTransactionTimeoutMillis() : options.timeoutMillis();
+
+            var transaction = new ReadOnlyTransactionImpl(
+                    this, timestampTracker, txId, localNodeId, implicit, timeout, readTimestamp, txFuture
+            );
 
             // Implicit transactions are finished as soon as their operation/query is finished, they cannot be abandoned, so there is
             // no need to register them.
@@ -475,7 +483,7 @@ private ReadOnlyTransactionImpl beginReadOnlyTransaction(
     }
 
     private long roExpirationPhysicalTimeFor(HybridTimestamp beginTimestamp, InternalTxOptions options) {
-        long effectiveTimeoutMillis = options.timeoutMillis() == 0 ? defaultTransactionTimeoutMillis() : options.timeoutMillis();
+        long effectiveTimeoutMillis = options.timeoutMillis() == 0 ? defaultReadOnlyTransactionTimeoutMillis() : options.timeoutMillis();
         return sumWithSaturation(beginTimestamp.getPhysical(), effectiveTimeoutMillis);
     }
 
@@ -493,8 +501,8 @@ private static long sumWithSaturation(long a, long b) {
         }
     }
 
-    private long defaultTransactionTimeoutMillis() {
-        return txConfig.timeout().value();
+    private long defaultReadOnlyTransactionTimeoutMillis() {
+        return txConfig.readOnlyTimeout().value();
     }
 
     /**