IGNITE-24963

modules/api/src/main/java/org/apache/ignite/tx/RunInTransactionInternalImpl.java

@@ -66,6 +66,8 @@ static <T> T runInTransactionInternal(
             try {
                 ret = clo.apply(tx);
 
+                tx.commit(); // Commit is retriable.
+
                 break;
             } catch (Exception ex) {
                 addSuppressedToList(suppressed, ex);
@@ -98,19 +100,6 @@ static <T> T runInTransactionInternal(
             }
         }
 
-        try {
-            tx.commit();
-        } catch (Exception e) {
-            try {
-                // Try to rollback tx in case if it's not finished. Retry is not needed here due to the durable finish.
-                tx.rollback();
-            } catch (Exception re) {
-                e.addSuppressed(re);
-            }
-
-            throw e;
-        }
-
         return ret;
     }
 
@@ -158,6 +147,7 @@ static <T> CompletableFuture<T> runInTransactionAsyncInternal(
                 .thenCompose(tx -> {
                     try {
                         return clo.apply(tx)
+                                .thenCompose(res -> tx.commitAsync().thenApply(ignored -> res))
                                 .handle((res, e) -> {
                                     if (e != null) {
                                         return handleClosureException(
@@ -173,30 +163,11 @@ static <T> CompletableFuture<T> runInTransactionAsyncInternal(
                                     } else {
                                         return completedFuture(res);
                                     }
-                                })
-                                .thenCompose(identity())
-                                .thenApply(res -> new TxWithVal<>(tx, res));
+                                }).thenCompose(identity());
                     } catch (Exception e) {
-                        return handleClosureException(igniteTransactions, tx, clo, txOptions, startTimestamp, initialTimeout, sup, e)
-                                .thenApply(res -> new TxWithVal<>(tx, res));
+                        return handleClosureException(igniteTransactions, tx, clo, txOptions, startTimestamp, initialTimeout, sup, e);
                     }
-                })
-                // Transaction commit with rollback on failure, without retries.
-                // Transaction rollback on closure failure is implemented in closure retry logic.
-                .thenCompose(txWithVal ->
-                        txWithVal.tx.commitAsync()
-                                .handle((ignored, e) -> {
-                                    if (e == null) {
-                                        return completedFuture(null);
-                                    } else {
-                                        return txWithVal.tx.rollbackAsync()
-                                                // Rethrow commit exception.
-                                                .handle((ign, re) -> sneakyThrow(e));
-                                    }
-                                })
-                                .thenCompose(fut -> fut)
-                                .thenApply(ignored -> txWithVal.val)
-                );
+                });
     }
 
     private static <T> CompletableFuture<T> handleClosureException(
@@ -347,14 +318,4 @@ private static long calcRemainingTime(long initialTimeout, long startTimestamp)
     private static <E extends Throwable> E sneakyThrow(Throwable e) throws E {
         throw (E) e;
     }
-
-    private static class TxWithVal<T> {
-        private final Transaction tx;
-        private final T val;
-
-        private TxWithVal(Transaction tx, T val) {
-            this.tx = tx;
-            this.val = val;
-        }
-    }
 }

modules/api/src/test/java/org/apache/ignite/tx/RunInTransactionRetryTest.java

@@ -88,8 +88,8 @@ public void testRetries(
         }
 
         boolean requiresEventualSuccess = closureFailureCount < Integer.MAX_VALUE
-                // Commit failure can't be retried.
-                && commitFailureCount == 0
+                && commitFailureCount < Integer.MAX_VALUE
+                && (commitFailureCount == 0 || rollbackFailureCount < Integer.MAX_VALUE)
                 // Rollbacks should be retried until success or timeout, so the rollback must succeed before closure retry.
                 && (closureFailureCount == 0 || rollbackFailureCount < Integer.MAX_VALUE);
 

modules/core/src/testFixtures/java/org/apache/ignite/internal/testframework/IgniteTestUtils.java

@@ -24,6 +24,7 @@
 import static org.apache.ignite.internal.util.IgniteUtils.deleteIfExists;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.is;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.fail;
@@ -702,6 +703,19 @@ public static boolean waitForCondition(BooleanSupplier cond, long sleepMillis, l
         return false;
     }
 
+    /**
+     * Ensure the future is not completed for a duration.
+     *
+     * @param future The future.
+     * @param durationMillis Milliseconds to check for condition.
+     */
+    public static void ensureFutureNotCompleted(CompletableFuture<?> future, long durationMillis) {
+        Awaitility.await()
+                .pollInterval(10, TimeUnit.MILLISECONDS)
+                .during(durationMillis, TimeUnit.MILLISECONDS)
+                .until(future::isDone, is(false));
+    }
+
     /**
      * Returns random byte array.
      *

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

@@ -130,6 +130,8 @@ public TxFinishReplicaRequestHandler(
      * @return future result of the operation.
      */
     public CompletableFuture<TransactionResult> handle(TxFinishReplicaRequest request) {
+        //LOG.info("DBG: handle finish " + request.txId() + " commit=" + request.commit());
+
         Map<ZonePartitionId, PartitionEnlistment> enlistedGroups = asReplicationGroupIdToPartitionMap(request.groups());
 
         UUID txId = request.txId();
@@ -223,10 +225,13 @@ private CompletableFuture<TransactionResult> finishAndCleanup(
         List<EnlistedPartitionGroup> enlistedPartitionGroups = enlistedPartitions.entrySet().stream()
                 .map(entry -> new EnlistedPartitionGroup(entry.getKey(), entry.getValue().tableIds()))
                 .collect(toList());
+        // LOG.info("DBG: finishTx " + txId + " " + commit + " " + enlistedPartitionGroups.size());
         return finishTransaction(enlistedPartitionGroups, txId, commit, commitTimestamp)
-                .thenCompose(txResult ->
-                    txManager.cleanup(replicationGroupId, enlistedPartitions, commit, commitTimestamp, txId)
-                            .thenApply(v -> txResult)
+                .thenCompose(txResult -> {
+                            //LOG.info("DBG: done finishTx " + txId);
+                            return txManager.cleanup(replicationGroupId, enlistedPartitions, commit, commitTimestamp, txId)
+                                    .thenApply(v -> txResult);
+                        }
                 );
     }
 

modules/runner/src/integrationTest/java/org/apache/ignite/internal/benchmark/TpccBenchmarkNodeRunner.java

@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.benchmark;
+
+import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
+import static org.apache.ignite.internal.catalog.CatalogService.DEFAULT_STORAGE_PROFILE;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.getAllResultSet;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+import java.io.File;
+import java.nio.file.Path;
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteServer;
+import org.apache.ignite.InitParameters;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.failure.handlers.configuration.StopNodeOrHaltFailureHandlerConfigurationSchema;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.testframework.TestIgnitionManager;
+import org.apache.ignite.sql.IgniteSql;
+import org.apache.ignite.sql.ResultSet;
+import org.apache.ignite.sql.SqlRow;
+import org.apache.ignite.sql.Statement;
+import org.apache.ignite.sql.Statement.StatementBuilder;
+import org.apache.ignite.tx.Transaction;
+import org.intellij.lang.annotations.Language;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Extendable class to start a dedicated cluster node for TPC-C benchmark.
+ */
+public class TpccBenchmarkNodeRunner {
+    private static final int BASE_PORT = 3344;
+    private static final int BASE_CLIENT_PORT = 10800;
+    private static final int BASE_REST_PORT = 10300;
+
+    private static final List<IgniteServer> igniteServers = new ArrayList<>();
+
+    protected static Ignite publicIgnite;
+    protected static IgniteImpl igniteImpl;
+
+    public static void main(String[] args) throws Exception {
+        TpccBenchmarkNodeRunner runner = new TpccBenchmarkNodeRunner();
+        runner.startCluster();
+    }
+
+    public IgniteImpl node(int idx) {
+        return unwrapIgniteImpl(igniteServers.get(idx).api());
+    }
+
+    private void startCluster() throws Exception {
+        Path workDir = workDir();
+
+        String connectNodeAddr = "\"localhost:" + BASE_PORT + '\"';
+
+        @Language("HOCON")
+        String configTemplate = "ignite {\n"
+                + "  \"network\": {\n"
+                + "    \"port\":{},\n"
+                + "    \"nodeFinder\":{\n"
+                + "      \"netClusterNodes\": [ {} ]\n"
+                + "    }\n"
+                + "  },\n"
+                + "  storage.profiles: {"
+                + "        " + DEFAULT_STORAGE_PROFILE + ".engine: aipersist, "
+                + "        " + DEFAULT_STORAGE_PROFILE + ".sizeBytes: " + pageMemorySize() + " "
+                + "  },\n"
+                + "  clientConnector: { port:{} },\n"
+                + "  clientConnector.sendServerExceptionStackTraceToClient: true\n"
+                + "  rest.port: {},\n"
+                + "  raft.fsync = " + fsync() + ",\n"
+                + "  system.partitionsLogPath = \"" + logPath() + "\",\n"
+                + "  failureHandler.handler: {\n"
+                + "      type: \"" + StopNodeOrHaltFailureHandlerConfigurationSchema.TYPE + "\",\n"
+                + "      tryStop: true,\n"
+                + "      timeoutMillis: 60000,\n" // 1 minute for graceful shutdown
+                + "  },\n"
+                + "}";
+
+        for (int i = 0; i < nodes(); i++) {
+            int port = BASE_PORT + i;
+            String nodeName = nodeName(port);
+
+            String config = IgniteStringFormatter.format(configTemplate, port, connectNodeAddr,
+                    BASE_CLIENT_PORT + i, BASE_REST_PORT + i);
+
+            igniteServers.add(TestIgnitionManager.startWithProductionDefaults(nodeName, config, workDir.resolve(nodeName)));
+        }
+
+        String metaStorageNodeName = nodeName(BASE_PORT);
+
+        InitParameters initParameters = InitParameters.builder()
+                .metaStorageNodeNames(metaStorageNodeName)
+                .clusterName("cluster")
+                .clusterConfiguration(clusterConfiguration())
+                .build();
+
+        TestIgnitionManager.init(igniteServers.get(0), initParameters);
+
+        for (IgniteServer node : igniteServers) {
+            assertThat(node.waitForInitAsync(), willCompleteSuccessfully());
+
+            if (publicIgnite == null) {
+                publicIgnite = node.api();
+                igniteImpl = unwrapIgniteImpl(publicIgnite);
+            }
+        }
+    }
+
+    @Nullable
+    protected String clusterConfiguration() {
+        return "ignite {}";
+    }
+
+    protected static String nodeName(int port) {
+        return "node_" + port;
+    }
+
+    protected Path workDir() throws Exception {
+        return new File("c:/work/tpcc").toPath();
+    }
+
+    protected int pageMemorySize() {
+        return 2073741824;
+    }
+
+    protected String logPath() {
+        return "";
+    }
+
+    protected boolean fsync() {
+        return false;
+    }
+
+    protected int nodes() {
+        return 1;
+    }
+
+    protected void dumpWarehouse() {
+        final String query = "select * from warehouse";
+        System.out.println("Executing the query: ");
+        List<List<Object>> rows = sql(publicIgnite, null, null, null, query);
+        for (List<Object> row : rows) {
+            System.out.println("Row: " + row);
+        }
+    }
+
+    protected static List<List<Object>> sql(Ignite node, @Nullable Transaction tx, @Nullable String schema, @Nullable ZoneId zoneId,
+            String query, Object... args) {
+        IgniteSql sql = node.sql();
+        StatementBuilder builder = sql.statementBuilder()
+                .query(query);
+
+        if (zoneId != null) {
+            builder.timeZoneId(zoneId);
+        }
+
+        if (schema != null) {
+            builder.defaultSchema(schema);
+        }
+
+        Statement statement = builder.build();
+        try (ResultSet<SqlRow> rs = sql.execute(tx, statement, args)) {
+            return getAllResultSet(rs);
+        }
+    }
+}