IGNITE-27002 Add labels to transactions

modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTableCommon.java

@@ -601,7 +601,7 @@ private static InternalTransaction startImplicitTx(
             TxManager txManager,
             boolean readOnly
     ) {
-        return txManager.beginImplicit(tsTracker, readOnly);
+        return txManager.beginImplicit(tsTracker, readOnly, null);
     }
 
     /**

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

@@ -67,7 +67,7 @@ public CompletableFuture<Void> stopAsync(ComponentContext componentContext) {
     }
 
     @Override
-    public InternalTransaction beginImplicit(HybridTimestampTracker timestampTracker, boolean readOnly) {
+    public InternalTransaction beginImplicit(HybridTimestampTracker timestampTracker, boolean readOnly, String txLabel) {
         return begin(timestampTracker, true, readOnly, InternalTxOptions.defaults());
     }
 

modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItIdempotentCommandCacheTest.java

@@ -110,7 +110,6 @@
 import org.apache.ignite.raft.jraft.rpc.impl.RaftGroupEventsClientListener;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInfo;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -399,7 +398,6 @@ public void testIdempotentInvoke() throws InterruptedException {
     }
 
     @Test
-    @Disabled("https://issues.apache.org/jira/browse/IGNITE-26870")
     public void testIdempotentInvokeAfterLeaderChange() {
         InvokeCommand invokeCommand = (InvokeCommand) buildKeyNotExistsInvokeCommand(TEST_KEY, TEST_VALUE, ANOTHER_VALUE);
 

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

@@ -19,13 +19,13 @@
 
 import static org.apache.ignite.internal.tx.TxState.COMMITTED;
 import static org.apache.ignite.internal.tx.TxState.isFinalState;
+import static org.apache.ignite.internal.tx.TxStateMeta.builder;
 
 import java.util.UUID;
 import org.apache.ignite.internal.hlc.HybridTimestamp;
 import org.apache.ignite.internal.replicator.listener.ReplicaListener;
 import org.apache.ignite.internal.tx.TxManager;
 import org.apache.ignite.internal.tx.TxState;
-import org.apache.ignite.internal.tx.TxStateMeta;
 import org.jetbrains.annotations.Nullable;
 
 /**
@@ -48,15 +48,9 @@ public ReplicaTxFinishMarker(TxManager txManager) {
     public void markFinished(UUID txId, TxState txState, @Nullable HybridTimestamp commitTimestamp) {
         assert isFinalState(txState) : "Unexpected state [txId=" + txId + ", txState=" + txState + ']';
 
-        txManager.updateTxMeta(txId, old -> new TxStateMeta(
-                txState,
-                old == null ? null : old.txCoordinatorId(),
-                old == null ? null : old.commitPartitionId(),
-                txState == COMMITTED ? commitTimestamp : null,
-                old == null ? null : old.tx(),
-                old == null ? null : old.initialVacuumObservationTimestamp(),
-                old == null ? null : old.cleanupCompletionTimestamp(),
-                old == null ? null : old.isFinishedDueToTimeout()
-        ));
+        txManager.updateTxMeta(txId, old -> builder(old, txState)
+                .commitTimestamp(txState == COMMITTED ? commitTimestamp : null)
+                .build()
+        );
     }
 }

modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteReplicaRequest.java

@@ -67,4 +67,11 @@ default boolean isWrite() {
      * @return {@code True} to disable the delayed ack optimization.
      */
     boolean skipDelayedAck();
+
+    /**
+     * Transaction label.
+     *
+     * @return Transaction label.
+     */
+    @Nullable String txLabel();
 }

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

@@ -51,15 +51,10 @@ public void markFinished(
             @Nullable HybridTimestamp commitTimestamp,
             @Nullable ReplicationGroupId commitPartitionId
     ) {
-        txManager.updateTxMeta(txId, old -> new TxStateMeta(
-                commit ? COMMITTED : ABORTED,
-                old == null ? null : old.txCoordinatorId(),
-                old == null ? commitPartitionId : old.commitPartitionId(),
-                commit ? commitTimestamp : null,
-                old == null ? null : old.tx(),
-                old == null ? null : old.initialVacuumObservationTimestamp(),
-                old == null ? null : old.cleanupCompletionTimestamp(),
-                old == null ? null : old.isFinishedDueToTimeout()
-        ));
+        txManager.updateTxMeta(txId, old -> TxStateMeta.builder(old, commit ? COMMITTED : ABORTED)
+                .commitPartitionId(commitPartitionId)
+                .commitTimestamp(commit ? commitTimestamp : null)
+                .build()
+        );
     }
 }

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/QueryTransactionContextImpl.java

@@ -56,7 +56,7 @@ public QueryTransactionWrapper getOrStartSqlManaged(boolean readOnly, boolean im
 
         if (tx == null) {
             if (implicit) {
-                transaction = txManager.beginImplicit(observableTimeTracker, readOnly);
+                transaction = txManager.beginImplicit(observableTimeTracker, readOnly, null);
             } else {
                 transaction = txManager.beginExplicit(observableTimeTracker, readOnly, InternalTxOptions.defaults());
             }

modules/table/build.gradle

@@ -88,6 +88,7 @@ dependencies {
     testImplementation testFixtures(project(':ignite-low-watermark'))
     testImplementation testFixtures(project(':ignite-failure-handler'))
     testImplementation testFixtures(project(':ignite-metrics'))
+    testImplementation testFixtures(project(':ignite-transactions'))
     testImplementation libs.hamcrest.core
     testImplementation libs.jmh.core
     testImplementation libs.javax.annotations

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

@@ -170,14 +170,11 @@ private void testTransaction(Consumer<Transaction> touchOp, boolean checkAfterTo
         } else {
             checkLocalTxStateOnNodes(
                     tx.id(),
-                    new TxStateMeta(
-                            commit ? COMMITTED : ABORTED,
-                            coordinatorId,
-                            tx.commitPartition(),
-                            commit ? testCluster.clockServices.get(coord.name()).now() : null,
-                            null,
-                            null
-                    )
+                    TxStateMeta.builder(commit ? COMMITTED : ABORTED)
+                            .txCoordinatorId(coordinatorId)
+                            .commitPartitionId(tx.commitPartition())
+                            .commitTimestamp(commit ? testCluster.clockServices.get(coord.name()).now() : null)
+                            .build()
             );
         }
     }

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

@@ -763,14 +763,11 @@ private static <T extends Comparable<T>> void updateTrackerIgnoringTrackerClosed
     }
 
     private void replicaTouch(UUID txId, UUID txCoordinatorId, HybridTimestamp commitTimestamp, boolean full) {
-        txManager.updateTxMeta(txId, old -> new TxStateMeta(
-                full ? COMMITTED : PENDING,
-                txCoordinatorId,
-                old == null ? null : old.commitPartitionId(),
-                full ? commitTimestamp : null,
-                old == null ? null : old.tx(),
-                old == null ? null : old.isFinishedDueToTimeout()
-        ));
+        txManager.updateTxMeta(txId, old -> TxStateMeta.builder(old, full ? COMMITTED : PENDING)
+                .txCoordinatorId(txCoordinatorId)
+                .commitTimestamp(full ? commitTimestamp : null)
+                .build()
+        );
     }
 
     /**

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

@@ -37,6 +37,7 @@
 import static org.apache.ignite.internal.tx.TxState.FINISHING;
 import static org.apache.ignite.internal.tx.TxState.PENDING;
 import static org.apache.ignite.internal.tx.TxState.isFinalState;
+import static org.apache.ignite.internal.tx.TxStateMeta.builder;
 import static org.apache.ignite.internal.util.CollectionUtils.nullOrEmpty;
 import static org.apache.ignite.internal.util.CompletableFutures.allOfToList;
 import static org.apache.ignite.internal.util.CompletableFutures.emptyCollectionCompletedFuture;
@@ -471,14 +472,11 @@ private CompletableFuture<?> processRequest(ReplicaRequest request, ReplicaPrima
 
             // Saving state is not needed for full transactions.
             if (!req.full()) {
-                txManager.updateTxMeta(req.transactionId(), old -> new TxStateMeta(
-                        PENDING,
-                        req.coordinatorId(),
-                        req.commitPartitionId().asZonePartitionId(),
-                        null,
-                        old == null ? null : old.tx(),
-                        old == null ? null : old.isFinishedDueToTimeout()
-                ));
+                txManager.updateTxMeta(req.transactionId(), old -> builder(old, PENDING)
+                        .txCoordinatorId(req.coordinatorId())
+                        .commitPartitionId(req.commitPartitionId().asZonePartitionId())
+                        .txLabel(req.txLabel())
+                        .build());
             }
         }
 
@@ -590,14 +588,11 @@ private CompletableFuture<?> processOperationRequest(
             // We treat SCAN as 2pc and only switch to a 1pc mode if all table rows fit in the bucket and the transaction is implicit.
             // See `req.full() && (err != null || rows.size() < req.batchSize())` condition.
             // If they don't fit the bucket, the transaction is treated as 2pc.
-            txManager.updateTxMeta(req.transactionId(), old -> new TxStateMeta(
-                    PENDING,
-                    req.coordinatorId(),
-                    req.commitPartitionId().asZonePartitionId(),
-                    null,
-                    old == null ? null : old.tx(),
-                    old == null ? null : old.isFinishedDueToTimeout()
-            ));
+            txManager.updateTxMeta(req.transactionId(),  old -> builder(old, PENDING)
+                    .txCoordinatorId(req.coordinatorId())
+                    .commitPartitionId(req.commitPartitionId().asZonePartitionId())
+                    .txLabel(req.txLabel())
+                    .build());
 
             var opId = new OperationId(senderId, req.timestamp().longValue());