IGNITE-24244: Support timeout in RW transactions

modules/client/src/main/java/org/apache/ignite/internal/client/tx/ClientTransactions.java

@@ -62,18 +62,14 @@ static CompletableFuture<ClientTransaction> beginAsync(
             @Nullable String preferredNodeName,
             @Nullable TransactionOptions options,
             long observableTimestamp) {
-        if (options != null && options.timeoutMillis() != 0 && !options.readOnly()) {
-            // TODO: IGNITE-16193
-            throw new UnsupportedOperationException("Timeouts are not supported yet for RW transactions");
-        }
-
         boolean readOnly = options != null && options.readOnly();
+        long timeout = options == null ? 0 : options.timeoutMillis();
 
         return ch.serviceAsync(
                 ClientOp.TX_BEGIN,
                 w -> {
                     w.out().packBoolean(readOnly);
-                    w.out().packLong(options == null ? 0 : options.timeoutMillis());
+                    w.out().packLong(timeout);
                     w.out().packLong(observableTimestamp);
                 },
                 r -> readTx(r, readOnly),

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

@@ -159,7 +159,9 @@ public boolean implicit() {
             }
 
             @Override
-            public CompletableFuture<Void> finish(boolean commit, HybridTimestamp executionTimestamp, boolean full) {
+            public CompletableFuture<Void> finish(
+                    boolean commit, HybridTimestamp executionTimestamp, boolean full, boolean timeoutExceeded
+            ) {
                 return nullCompletedFuture();
             }
 
@@ -205,6 +207,7 @@ public CompletableFuture<Void> finish(
             HybridTimestampTracker timestampTracker,
             TablePartitionId commitPartition,
             boolean commit,
+            boolean timeoutExceeded,
             Map<ReplicationGroupId, IgniteBiTuple<ClusterNode, Long>> enlistedGroups,
             Set<Integer> enlistedTableIds,
             UUID txId
@@ -260,7 +263,9 @@ public int pending() {
     }
 
     @Override
-    public void finishFull(HybridTimestampTracker timestampTracker, UUID txId, HybridTimestamp ts, boolean commit) {
+    public void finishFull(
+            HybridTimestampTracker timestampTracker, UUID txId, HybridTimestamp ts, boolean commit, boolean timeoutExceeded
+    ) {
         // No-op.
     }
 }

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

@@ -55,7 +55,8 @@ public void markFinished(UUID txId, TxState txState, @Nullable HybridTimestamp c
                 txState == COMMITTED ? commitTimestamp : null,
                 old == null ? null : old.tx(),
                 old == null ? null : old.initialVacuumObservationTimestamp(),
-                old == null ? null : old.cleanupCompletionTimestamp()
+                old == null ? null : old.cleanupCompletionTimestamp(),
+                old == null ? null : old.isFinishedDueToTimeout()
         ));
     }
 }

modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/RaftTxFinishMarker.java

@@ -58,7 +58,8 @@ public void markFinished(
                 commit ? commitTimestamp : null,
                 old == null ? null : old.tx(),
                 old == null ? null : old.initialVacuumObservationTimestamp(),
-                old == null ? null : old.cleanupCompletionTimestamp()
+                old == null ? null : old.cleanupCompletionTimestamp(),
+                old == null ? null : old.isFinishedDueToTimeout()
         ));
     }
 }

modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItDmlTest.java

@@ -40,6 +40,7 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Phaser;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.apache.calcite.sql.type.SqlTypeName;
@@ -147,7 +148,6 @@ public void batchWithConflictShouldBeRejectedEntirely() {
                 () -> sql("INSERT INTO test VALUES (0, 0), (1, 1), (2, 2)")
         );
 
-
         assertQuery("SELECT count(*) FROM test")
                 .returns(1L)
                 .check();
@@ -214,7 +214,7 @@ public void testNullDefault() {
         assertQuery("SELECT col FROM test_null_def WHERE id = 2").returns(null).check();
     }
 
-    /**Test full MERGE command. */
+    /** Test full MERGE command. */
     @Test
     public void testMerge() {
         clearAndPopulateMergeTable1();
@@ -373,9 +373,12 @@ public void testMergeBatch() {
 
         assertQuery("SELECT count(*) FROM test2 WHERE b = 0").returns(10_000L).check();
 
-        sql("MERGE INTO test2 dst USING test1 src ON dst.a = src.a"
+        var longerTimeoutOptions = new TransactionOptions().readOnly(false).timeoutMillis(TimeUnit.MINUTES.toMillis(2));
+        var tx = igniteTx().begin(longerTimeoutOptions);
+        sql(tx, "MERGE INTO test2 dst USING test1 src ON dst.a = src.a"
                 + " WHEN MATCHED THEN UPDATE SET b = 1 "
                 + " WHEN NOT MATCHED THEN INSERT (key, a, b) VALUES (src.key, src.a, 2)");
+        tx.commit();
 
         assertQuery("SELECT count(*) FROM test2 WHERE b = 0").returns(5_000L).check();
         assertQuery("SELECT count(*) FROM test2 WHERE b = 1").returns(5_000L).check();

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

@@ -102,7 +102,7 @@ public void commit() throws TransactionException {
 
     @Override
     public CompletableFuture<Void> commitAsync() {
-        return finish(true, nullableHybridTimestamp(NULL_HYBRID_TIMESTAMP), false);
+        return finish(true, nullableHybridTimestamp(NULL_HYBRID_TIMESTAMP), false, false);
     }
 
     @Override
@@ -112,7 +112,7 @@ public void rollback() throws TransactionException {
 
     @Override
     public CompletableFuture<Void> rollbackAsync() {
-        return finish(false, nullableHybridTimestamp(NULL_HYBRID_TIMESTAMP), false);
+        return finish(false, nullableHybridTimestamp(NULL_HYBRID_TIMESTAMP), false, false);
     }
 
     @Override
@@ -169,7 +169,7 @@ public boolean implicit() {
     }
 
     @Override
-    public CompletableFuture<Void> finish(boolean commit, HybridTimestamp executionTimestamp, boolean full) {
+    public CompletableFuture<Void> finish(boolean commit, HybridTimestamp executionTimestamp, boolean full, boolean timeoutExceeded) {
         CompletableFuture<Void> fut = commit ? commitFut : rollbackFut;
 
         fut.complete(null);

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

@@ -145,13 +145,13 @@ private void testTransaction(Consumer<Transaction> touchOp, boolean checkAfterTo
 
         ReadWriteTransactionImpl tx = (ReadWriteTransactionImpl) testCluster.igniteTransactions().begin();
 
-        checkLocalTxStateOnNodes(tx.id(), new TxStateMeta(PENDING, coordinatorId, tx.commitPartition(), null, null), List.of(0));
+        checkLocalTxStateOnNodes(tx.id(), new TxStateMeta(PENDING, coordinatorId, tx.commitPartition(), null, null, null), List.of(0));
         checkLocalTxStateOnNodes(tx.id(), null, IntStream.range(1, NODES).boxed().collect(toList()));
 
         touchOp.accept(tx);
 
         if (checkAfterTouch) {
-            checkLocalTxStateOnNodes(tx.id(), new TxStateMeta(PENDING, coordinatorId, tx.commitPartition(), null, null));
+            checkLocalTxStateOnNodes(tx.id(), new TxStateMeta(PENDING, coordinatorId, tx.commitPartition(), null, null, null));
         }
 
         if (commit) {
@@ -167,6 +167,7 @@ private void testTransaction(Consumer<Transaction> touchOp, boolean checkAfterTo
                         coordinatorId,
                         tx.commitPartition(),
                         commit ? testCluster.clockServices.get(coord.name()).now() : null,
+                        null,
                         null
                 )
         );

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

@@ -220,6 +220,7 @@ public CompletableFuture<Void> finish(
                     HybridTimestampTracker observableTimestampTracker,
                     TablePartitionId commitPartition,
                     boolean commitIntent,
+                    boolean timeoutExceeded,
                     Map<ReplicationGroupId, IgniteBiTuple<ClusterNode, Long>> enlistedGroups,
                     Set<Integer> enlistedTableIds,
                     UUID txId

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

@@ -687,7 +687,8 @@ private void replicaTouch(UUID txId, UUID txCoordinatorId, HybridTimestamp commi
                 txCoordinatorId,
                 old == null ? null : old.commitPartitionId(),
                 full ? commitTimestamp : null,
-                old == null ? null : old.tx()
+                old == null ? null : old.tx(),
+                old == null ? null : old.isFinishedDueToTimeout()
         ));
     }
 

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

@@ -545,7 +545,8 @@ private CompletableFuture<?> processRequest(ReplicaRequest request, @Nullable Bo
                         req.coordinatorId(),
                         req.commitPartitionId().asTablePartitionId(),
                         null,
-                        old == null ? null : old.tx()
+                        old == null ? null : old.tx(),
+                        old == null ? null : old.isFinishedDueToTimeout()
                 ));
             }
         }
@@ -725,6 +726,7 @@ private CompletableFuture<Void> triggerTxRecovery(UUID txId, UUID senderId) {
                         // Tx recovery is executed on the commit partition.
                         replicationGroupId,
                         false,
+                        false,
                         // Enlistment consistency token is not required for the rollback, so it is 0L.
                         Map.of(replicationGroupId, new IgniteBiTuple<>(clusterNodeResolver.getById(senderId), 0L)),
                         Set.of(replicationGroupId.tableId()),
@@ -842,7 +844,8 @@ private CompletableFuture<?> processOperationRequest(
                     req.coordinatorId(),
                     req.commitPartitionId().asTablePartitionId(),
                     null,
-                    old == null ? null : old.tx()
+                    old == null ? null : old.tx(),
+                    old == null ? null : old.isFinishedDueToTimeout()
             ));
 
             var opId = new OperationId(senderId, req.timestamp().longValue());

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

@@ -666,7 +666,7 @@ private <R> CompletableFuture<R> trackingInvoke(
                 assert hasError || r instanceof TimestampAware;
 
                 // Timestamp is set to commit timestamp for full transactions.
-                tx.finish(!hasError, hasError ? null : ((TimestampAware) r).timestamp(), true);
+                tx.finish(!hasError, hasError ? null : ((TimestampAware) r).timestamp(), true, false);
 
                 if (e != null) {
                     sneakyThrow(e);
@@ -679,10 +679,11 @@ private <R> CompletableFuture<R> trackingInvoke(
                 if (!transactionInflights.addInflight(tx.id(), false)) {
                     return failedFuture(
                             new TransactionException(TX_ALREADY_FINISHED_ERR, format(
-                                    "Transaction is already finished [tableName={}, partId={}, txState={}].",
+                                    "Transaction is already finished [tableName={}, partId={}, txState={}, timeoutExceeded={}].",
                                     tableName,
                                     partId,
-                                    tx.state()
+                                    tx.state(),
+                                    tx.isTimeoutExceeded()
                             )));
                 }
 
@@ -880,7 +881,7 @@ private <R> CompletableFuture<R> sendReadOnlyToPrimaryReplica(
     private <R> CompletableFuture<R> postEvaluate(CompletableFuture<R> fut, InternalTransaction tx) {
         return fut.handle((BiFunction<R, Throwable, CompletableFuture<R>>) (r, e) -> {
             if (e != null) {
-                return tx.finish(false, clockService.current(), false)
+                return tx.finish(false, clockService.current(), false, false)
                         .handle((ignored, err) -> {
                             if (err != null) {
                                 e.addSuppressed(err);
@@ -891,7 +892,7 @@ private <R> CompletableFuture<R> postEvaluate(CompletableFuture<R> fut, Internal
                         }); // Preserve failed state.
             }
 
-            return tx.finish(true, clockService.current(), false).thenApply(ignored -> r);
+            return tx.finish(true, clockService.current(), false, false).thenApply(ignored -> r);
         }).thenCompose(identity());
     }
 
@@ -2342,9 +2343,10 @@ private static long enlistmentConsistencyToken(ReplicaMeta replicaMeta) {
     private void checkTransactionFinishStarted(@Nullable InternalTransaction transaction) {
         if (transaction != null && transaction.isFinishingOrFinished()) {
             throw new TransactionException(TX_ALREADY_FINISHED_ERR, format(
-                    "Transaction is already finished () [txId={}, readOnly={}].",
+                    "Transaction is already finished () [txId={}, readOnly={}, timeoutExceeded={}].",
                     transaction.id(),
-                    transaction.isReadOnly()
+                    transaction.isReadOnly(),
+                    transaction.isTimeoutExceeded()
             ));
         }
     }

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

@@ -749,11 +749,12 @@ public void testTxStateReplicaRequestEmptyState() throws Exception {
                     localNode.id(),
                     commitPartitionId,
                     null,
+                    null,
                     null
             ));
 
             return nullCompletedFuture();
-        }).when(txManager).finish(any(), any(), anyBoolean(), any(), any(), any());
+        }).when(txManager).finish(any(), any(), anyBoolean(), any(), any(), any(), any());
 
         CompletableFuture<ReplicaResult> fut = partitionReplicaListener.invoke(TX_MESSAGES_FACTORY.txStateCommitPartitionRequest()
                 .groupId(tablePartitionIdMessage(grpId))
@@ -796,7 +797,7 @@ void testExecuteRequestOnFinishedTx(TxState txState, RequestType requestType) {
         UUID txId = newTxId();
 
         txStateStorage.putForRebalance(txId, new TxMeta(txState, singletonList(grpId), null));
-        txManager.updateTxMeta(txId, old -> new TxStateMeta(txState, null, null, null, null));
+        txManager.updateTxMeta(txId, old -> new TxStateMeta(txState, null, null, null, null, null));
 
         BinaryRow testRow = binaryRow(0);
 
@@ -931,7 +932,7 @@ public void testReadOnlySingleRowReplicaRequestResolveWriteIntentCommitted() thr
 
         pkStorage().put(testBinaryRow, rowId);
         testMvPartitionStorage.addWrite(rowId, testBinaryRow, txId, TABLE_ID, PART_ID);
-        txManager.updateTxMeta(txId, old -> new TxStateMeta(COMMITTED, localNode.id(), commitPartitionId, clock.now(), null));
+        txManager.updateTxMeta(txId, old -> new TxStateMeta(COMMITTED, localNode.id(), commitPartitionId, clock.now(), null, null));
 
         CompletableFuture<ReplicaResult> fut = doReadOnlySingleGet(testBinaryKey);
 
@@ -949,7 +950,7 @@ public void testReadOnlySingleRowReplicaRequestResolveWriteIntentPending() throw
 
         pkStorage().put(testBinaryRow, rowId);
         testMvPartitionStorage.addWrite(rowId, testBinaryRow, txId, TABLE_ID, PART_ID);
-        txManager.updateTxMeta(txId, old -> new TxStateMeta(TxState.PENDING, localNode.id(), commitPartitionId, null, null));
+        txManager.updateTxMeta(txId, old -> new TxStateMeta(TxState.PENDING, localNode.id(), commitPartitionId, null, null, null));
 
         CompletableFuture<ReplicaResult> fut = doReadOnlySingleGet(testBinaryKey);
 
@@ -968,7 +969,7 @@ public void testReadOnlySingleRowReplicaRequestResolveWriteIntentAborted() throw
 
         pkStorage().put(testBinaryRow, rowId);
         testMvPartitionStorage.addWrite(rowId, testBinaryRow, txId, TABLE_ID, PART_ID);
-        txManager.updateTxMeta(txId, old -> new TxStateMeta(ABORTED, localNode.id(), commitPartitionId, null, null));
+        txManager.updateTxMeta(txId, old -> new TxStateMeta(ABORTED, localNode.id(), commitPartitionId, null, null, null));
 
         CompletableFuture<ReplicaResult> fut = doReadOnlySingleGet(testBinaryKey);
 
@@ -1661,7 +1662,9 @@ private void testWriteIntentOnPrimaryReplica(
 
             // Imitation of tx commit.
             txStateStorage.putForRebalance(txId, new TxMeta(COMMITTED, new ArrayList<>(), now));
-            txManager.updateTxMeta(txId, old -> new TxStateMeta(COMMITTED, UUID.randomUUID(), commitPartitionId, now, null));
+            txManager.updateTxMeta(txId, old -> new TxStateMeta(
+                    COMMITTED, UUID.randomUUID(), commitPartitionId, now, null, null)
+            );
 
             CompletableFuture<?> replicaCleanupFut = partitionReplicaListener.invoke(
                     TX_MESSAGES_FACTORY.writeIntentSwitchReplicaRequest()
@@ -2250,8 +2253,10 @@ private static void configureTxManager(TxManager txManager) {
 
         doAnswer(invocation -> nullCompletedFuture()).when(txManager).executeWriteIntentSwitchAsync(any(Runnable.class));
 
-        doAnswer(invocation -> nullCompletedFuture()).when(txManager).finish(any(), any(), anyBoolean(), any(), any(), any());
-        doAnswer(invocation -> nullCompletedFuture()).when(txManager).cleanup(any(), anyString(), any());
+        doAnswer(invocation -> nullCompletedFuture())
+                .when(txManager).finish(any(), any(), anyBoolean(), anyBoolean(), any(), any(), any());
+        doAnswer(invocation -> nullCompletedFuture())
+                .when(txManager).cleanup(any(), anyString(), any());
     }
 
     private void testWritesAreSuppliedWithRequiredCatalogVersion(RequestType requestType, RwListenerInvocation listenerInvocation) {
@@ -2931,7 +2936,7 @@ private void cleanup(UUID txId) {
     private void cleanup(UUID txId, boolean commit) {
         HybridTimestamp commitTs = clock.now();
 
-        txManager.updateTxMeta(txId, old -> new TxStateMeta(COMMITTED, UUID.randomUUID(), commitPartitionId, commitTs, null));
+        txManager.updateTxMeta(txId, old -> new TxStateMeta(COMMITTED, UUID.randomUUID(), commitPartitionId, commitTs, null, null));
 
         WriteIntentSwitchReplicaRequest message = TX_MESSAGES_FACTORY.writeIntentSwitchReplicaRequest()
                 .groupId(tablePartitionIdMessage(grpId))

modules/table/src/testFixtures/java/org/apache/ignite/internal/table/TxAbstractTest.java

@@ -2172,7 +2172,8 @@ public void testWriteIntentResolutionFallbackToCommitPartitionPath() {
                     new UUID(1, 2),
                     old.commitPartitionId(),
                     old.commitTimestamp(),
-                    old == null ? null : old.tx()
+                    old == null ? null : old.tx(),
+                    old == null ? null : old.isFinishedDueToTimeout()
             ));
         }
 

modules/transactions/src/integrationTest/java/org/apache/ignite/internal/tx/readonly/ItClientReadOnlyTxTimeoutOneNodeTest.java → modules/transactions/src/integrationTest/java/org/apache/ignite/internal/tx/ItClientTxTimeoutOneNodeTest.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.ignite.internal.tx.readonly;
+package org.apache.ignite.internal.tx;
 
 import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
 
@@ -24,12 +24,11 @@
 import org.apache.ignite.client.handler.ClientResourceRegistry;
 import org.apache.ignite.internal.client.tx.ClientLazyTransaction;
 import org.apache.ignite.internal.lang.IgniteInternalCheckedException;
-import org.apache.ignite.internal.tx.impl.ReadOnlyTransactionImpl;
 import org.apache.ignite.tx.Transaction;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 
-class ItClientReadOnlyTxTimeoutOneNodeTest extends ItReadOnlyTxTimeoutOneNodeTest {
+class ItClientTxTimeoutOneNodeTest extends ItTxTimeoutOneNodeTest {
     private IgniteClient client;
 
     @BeforeEach
@@ -52,12 +51,12 @@ Ignite ignite() {
     }
 
     @Override
-    ReadOnlyTransactionImpl transactionImpl(Transaction tx) {
+    InternalTransaction toInternalTransaction(Transaction tx) {
         long txId = ClientLazyTransaction.get(tx).startedTx().id();
 
         ClientResourceRegistry resources = unwrapIgniteImpl(cluster.aliveNode()).clientInboundMessageHandler().resources();
         try {
-            return resources.get(txId).get(ReadOnlyTransactionImpl.class);
+            return resources.get(txId).get(InternalTransaction.class);
         } catch (IgniteInternalCheckedException e) {
             throw new RuntimeException(e);
         }