IGNITE-24242 Add RO/RW Transaction Timeout configuration schema

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

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

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

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

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

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

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

@@ -198,8 +198,9 @@ 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;
+    private final long roTransactionTimeout;
+
+    private final long rwTransactionTimeout;
 
     /** Attempts to take lock. */
     private final int attemptsObtainLock;
@@ -218,7 +219,8 @@ 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 roTransactionTimeout Read-only transaction timeout.
+     * @param rwTransactionTimeout RW transaction timeout.
      * @param attemptsObtainLock Attempts to take lock.
      */
     public InternalTableImpl(
@@ -234,7 +236,8 @@ public InternalTableImpl(
             HybridTimestampTracker observableTimestampTracker,
             PlacementDriver placementDriver,
             TransactionInflights transactionInflights,
-            long implicitTransactionTimeout,
+            long roTransactionTimeout,
+            long rwTransactionTimeout,
             int attemptsObtainLock,
             Supplier<ScheduledExecutorService> streamerFlushExecutor,
             StreamerReceiverRunner streamerReceiverRunner
@@ -251,7 +254,8 @@ public InternalTableImpl(
         this.observableTimestampTracker = observableTimestampTracker;
         this.placementDriver = placementDriver;
         this.transactionInflights = transactionInflights;
-        this.implicitTransactionTimeout = implicitTransactionTimeout;
+        this.roTransactionTimeout = roTransactionTimeout;
+        this.rwTransactionTimeout = rwTransactionTimeout;
         this.attemptsObtainLock = attemptsObtainLock;
         this.streamerFlushExecutor = streamerFlushExecutor;
         this.streamerReceiverRunner = streamerReceiverRunner;
@@ -368,9 +372,11 @@ private <R> CompletableFuture<R> enlistInTx(
         return postEnlist(fut, false, actualTx, actualTx.implicit()).handle((r, e) -> {
             if (e != null) {
                 if (actualTx.implicit()) {
+                    long timeout = actualTx.isReadOnly() ? roTransactionTimeout : rwTransactionTimeout;
+
                     long ts = (txStartTs == null) ? actualTx.startTimestamp().getPhysical() : txStartTs;
 
-                    if (exceptionAllowsImplicitTxRetry(e) && coarseCurrentTimeMillis() - ts < implicitTransactionTimeout) {
+                    if (exceptionAllowsImplicitTxRetry(e) && coarseCurrentTimeMillis() - ts < timeout) {
                         return enlistInTx(row, null, fac, noWriteChecker, ts);
                     }
                 }
@@ -486,9 +492,11 @@ private <T> CompletableFuture<T> enlistInTx(
         return postEnlist(fut, actualTx.implicit() && !singlePart, actualTx, full).handle((r, e) -> {
             if (e != null) {
                 if (actualTx.implicit()) {
+                    long timeout = actualTx.isReadOnly() ? roTransactionTimeout : rwTransactionTimeout;
+
                     long ts = (txStartTs == null) ? actualTx.startTimestamp().getPhysical() : txStartTs;
 
-                    if (exceptionAllowsImplicitTxRetry(e) && coarseCurrentTimeMillis() - ts < implicitTransactionTimeout) {
+                    if (exceptionAllowsImplicitTxRetry(e) && coarseCurrentTimeMillis() - ts < timeout) {
                         return enlistInTx(keyRows, null, fac, reducer, noOpChecker, ts);
                     }
                 }
@@ -1185,9 +1193,11 @@ private CompletableFuture<Void> updateAllWithRetry(
         // Will be finished in one RTT.
         return postEnlist(fut, false, tx, true).handle((r, e) -> {
             if (e != null) {
+                long timeout = tx.isReadOnly() ? roTransactionTimeout : rwTransactionTimeout;
+
                 long ts = (txStartTs == null) ? tx.startTimestamp().getPhysical() : txStartTs;
 
-                if (exceptionAllowsImplicitTxRetry(e) && coarseCurrentTimeMillis() - ts < implicitTransactionTimeout) {
+                if (exceptionAllowsImplicitTxRetry(e) && coarseCurrentTimeMillis() - ts < timeout) {
                     return updateAllWithRetry(rows, deleted, partition, ts);
                 }
 

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

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

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

@@ -73,6 +73,7 @@ void testUpdatePartitionTrackers() {
                 mock(PlacementDriver.class),
                 mock(TransactionInflights.class),
                 3_000,
+                3_000,
                 0,
                 null,
                 mock(StreamerReceiverRunner.class)
@@ -123,6 +124,7 @@ void testRowBatchByPartitionId() {
                 mock(PlacementDriver.class),
                 mock(TransactionInflights.class),
                 3_000,
+                3_000,
                 0,
                 null,
                 null

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

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

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

@@ -269,6 +269,7 @@ public DummyInternalTableImpl(
                 placementDriver,
                 transactionInflights,
                 3_000,
+                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,"
+                + "      roTimeout: 30000,"
+                + "      rwTimeout: 30000,"
                 + "      txnResourceTtl: 2"
                 + "  },"
                 + "  replication: {"

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

@@ -41,10 +41,29 @@ public class TransactionConfigurationSchema {
     @Value(hasDefault = true)
     public final long timeout = 10_000;
 
-    /** Timeout for implicit transactions (milliseconds). */
     @Range(min = 1)
     @Value(hasDefault = true)
-    public final long implicitTransactionTimeout = 3_000;
+    public final long minRoTimeout = 1;
+
+    @Range(min = 1)
+    @Value(hasDefault = true)
+    public final long maxRoTimeout = Long.MAX_VALUE;
+
+    @Range(min = 1)
+    @Value(hasDefault = true)
+    public final long minRwTimeout = 1;
+
+    @Range(min = 1)
+    @Value(hasDefault = true)
+    public final long maxRwTimeout = Long.MAX_VALUE;
+
+    @Range(min = 1)
+    @Value(hasDefault = true)
+    public final long roTimeout = 10_000;
+
+    @Range(min = 1)
+    @Value(hasDefault = true)
+    public final long rwTimeout = 3_000;
 
     /** A transaction tries to take lock several times until it throws an exception {@lonk org.apache.ignite.tx.TransactionException}. */
     @Range(min = 0)