/*
 * 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.table;

import static java.util.concurrent.CompletableFuture.allOf;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.stream.Collectors.toList;
import static org.apache.ignite.internal.lang.IgniteSystemProperties.colocationEnabled;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.assertThrowsWithCause;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.assertThrowsWithCode;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
import static org.apache.ignite.lang.ErrorGroups.Common.INTERNAL_ERR;
import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_ALREADY_FINISHED_ERR;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.contains;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Flow;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.network.ClusterNodeImpl;
import org.apache.ignite.internal.replicator.Replica;
import org.apache.ignite.internal.replicator.ReplicaImpl;
import org.apache.ignite.internal.replicator.ReplicaManager;
import org.apache.ignite.internal.replicator.ReplicaTestUtils;
import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.schema.SchemaRegistry;
import org.apache.ignite.internal.schema.marshaller.TupleMarshallerImpl;
import org.apache.ignite.internal.schema.row.Row;
import org.apache.ignite.internal.storage.RowId;
import org.apache.ignite.internal.storage.impl.TestMvPartitionStorage;
import org.apache.ignite.internal.storage.index.IndexStorage;
import org.apache.ignite.internal.storage.index.impl.TestHashIndexStorage;
import org.apache.ignite.internal.table.distributed.TableSchemaAwareIndexStorage;
import org.apache.ignite.internal.table.distributed.replicator.PartitionReplicaListener;
import org.apache.ignite.internal.testframework.IgniteTestUtils;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.tx.InternalTxOptions;
import org.apache.ignite.internal.tx.Lock;
import org.apache.ignite.internal.tx.LockException;
import org.apache.ignite.internal.tx.LockManager;
import org.apache.ignite.internal.tx.LockMode;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.internal.tx.TxPriority;
import org.apache.ignite.internal.tx.TxState;
import org.apache.ignite.internal.tx.TxStateMeta;
import org.apache.ignite.internal.tx.impl.IgniteTransactionsImpl;
import org.apache.ignite.internal.tx.impl.ReadWriteTransactionImpl;
import org.apache.ignite.internal.tx.impl.TxManagerImpl;
import org.apache.ignite.internal.tx.impl.VolatileTxStateMetaStorage;
import org.apache.ignite.internal.tx.storage.state.TxStatePartitionStorage;
import org.apache.ignite.internal.util.CollectionUtils;
import org.apache.ignite.internal.util.Lazy;
import org.apache.ignite.internal.util.Pair;
import org.apache.ignite.internal.util.PendingComparableValuesTracker;
import org.apache.ignite.lang.IgniteException;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.table.KeyValueView;
import org.apache.ignite.table.RecordView;
import org.apache.ignite.table.Tuple;
import org.apache.ignite.tx.Transaction;
import org.apache.ignite.tx.TransactionException;
import org.apache.ignite.tx.TransactionOptions;
import org.jetbrains.annotations.Nullable;
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;
import org.junit.jupiter.api.function.Executable;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;

/**
 * Common tx test scenarios set. TODO asch IGNITE-15928 validate zero locks after test commit.
 */
@ExtendWith({MockitoExtension.class, ConfigurationExtension.class})
@MockitoSettings(strictness = Strictness.LENIENT)
public abstract class TxAbstractTest extends TxInfrastructureTest {
    /**
     * The constructor.
     *
     * @param testInfo Test info.
     */
    public TxAbstractTest(TestInfo testInfo) {
        super(testInfo);
    }

    @Test
    public void testCommitRollbackSameTxDoesNotThrow() throws TransactionException {
        InternalTransaction tx = (InternalTransaction) igniteTransactions.begin();

        accounts.recordView().upsert(tx, makeValue(1, 100.));

        tx.commit();

        assertDoesNotThrow(tx::rollback, "Unexpected exception was thrown.");
        assertDoesNotThrow(tx::commit, "Unexpected exception was thrown.");
        assertDoesNotThrow(tx::rollback, "Unexpected exception was thrown.");
    }

    @Test
    public void testRollbackCommitSameTxDoesNotThrow() throws TransactionException {
        InternalTransaction tx = (InternalTransaction) igniteTransactions.begin();

        accounts.recordView().upsert(tx, makeValue(1, 100.));

        tx.rollback();

        assertDoesNotThrow(tx::commit, "Unexpected exception was thrown.");
        assertDoesNotThrow(tx::rollback, "Unexpected exception was thrown.");
        assertDoesNotThrow(tx::commit, "Unexpected exception was thrown.");
    }

    @Test
    public void testRepeatedCommitRollbackAfterUpdateWithException() {
        injectFailureOnNextOperation(accounts);

        InternalTransaction tx = (InternalTransaction) igniteTransactions.begin();

        CompletableFuture<Void> fut = accounts.recordView().upsertAsync(tx, makeValue(1, 100.));

        assertThrows(Exception.class, fut::join);

        tx.commitAsync().join();

        tx.rollbackAsync().join();

        tx.commitAsync().join();
    }

    @Test
    public void testRepeatedCommitRollbackAfterRollbackWithException() {
        InternalTransaction tx = (InternalTransaction) igniteTransactions.begin();

        accounts.recordView().upsert(tx, makeValue(1, 100.));

        injectFailureOnNextOperation(accounts);

        CompletableFuture<Void> fut = tx.rollbackAsync();
        assertThrows(Exception.class, fut::join);

        tx.commitAsync().join();

        tx.rollbackAsync().join();

        tx.commitAsync().join();
    }

    @Test
    public void testDeleteUpsertCommit() throws TransactionException {
        deleteUpsert().commit();

        assertEquals(200., accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));
    }

    @Test
    public void testDeleteUpsertRollback() throws TransactionException {
        deleteUpsert().rollback();

        assertEquals(100., accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));
    }

    protected InternalTransaction deleteUpsert() {
        accounts.recordView().upsert(null, makeValue(1, 100.));

        InternalTransaction tx = (InternalTransaction) igniteTransactions.begin();

        accounts.recordView().delete(tx, makeKey(1));

        assertNull(accounts.recordView().get(tx, makeKey(1)));

        accounts.recordView().upsert(tx, makeValue(1, 200.));

        return tx;
    }

    @Test
    public void testDeleteUpsertAllCommit() throws TransactionException {
        deleteUpsertAll().commit();

        assertEquals(200., accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));
        assertEquals(200., accounts.recordView().get(null, makeKey(2)).doubleValue("balance"));
    }

    @Test
    public void testDeleteUpsertAllRollback() throws TransactionException {
        deleteUpsertAll().rollback();

        var res1 = accounts.recordView().get(null, makeKey(1));

        // Here we try to catch the flaky NPE problem.
        checkIfNull(res1, transaction -> accounts.recordView().get(transaction, makeKey(1)));

        assertEquals(100., res1.doubleValue("balance"), "tuple =[" + res1 + "]");

        var res2 = accounts.recordView().get(null, makeKey(2));

        // Here we try to catch the flaky NPE problem.
        checkIfNull(res2, transaction -> accounts.recordView().get(transaction, makeKey(2)));

        assertEquals(100., res2.doubleValue("balance"), "tuple =[" + res2 + "]");
    }

    private InternalTransaction deleteUpsertAll() {
        List<Tuple> tuples = new ArrayList<>();
        tuples.add(makeValue(1, 100.));
        tuples.add(makeValue(2, 100.));

        accounts.recordView().upsertAll(null, tuples);

        var res1 = accounts.recordView().get(null, makeKey(1));

        // Here we try to catch the flaky NPE problem.
        checkIfNull(res1, transaction -> accounts.recordView().get(transaction, makeKey(1)));

        assertEquals(100., res1.doubleValue("balance"), "tuple =[" + res1 + "]");

        var res2 = accounts.recordView().get(null, makeKey(2));
        // Here we try to catch the flaky NPE problem.
        checkIfNull(res2, transaction -> accounts.recordView().get(transaction, makeKey(2)));

        assertEquals(100., res2.doubleValue("balance"), "tuple =[" + res2 + "]");

        InternalTransaction tx = (InternalTransaction) igniteTransactions.begin();

        tuples.clear();
        tuples.add(makeKey(1));
        tuples.add(makeKey(2));

        accounts.recordView().deleteAll(tx, tuples);

        tuples.clear();
        tuples.add(makeValue(1, 200.));
        tuples.add(makeValue(2, 200.));

        accounts.recordView().upsertAll(tx, tuples);

        return tx;
    }

    private void checkIfNull(Tuple res1, Function<Transaction, Tuple> retryFunc) {
        if (res1 != null) {
            return;
        }

        logger().error("Found null value after upsertAll.");

        printDebugInfo();

        var resExplicit = igniteTransactions.runInTransaction(retryFunc);

        logger().info("Same explicit call resulted in {} value", resExplicit);

        fail("Found null value after upsertAll.");
    }

    private void printDebugInfo() {
        logger().info("Primary node for accounts table is [node={}]", primaryNode(accounts));
        logger().info("Cluster times:");
        long latestTime = IgniteTestUtils.getFieldValue(txTestCluster.clientClock(), HybridClockImpl.class, "latestTime");
        logger().info("Client clock [time={}]", HybridTimestamp.hybridTimestamp(latestTime));

        txTestCluster.clocks().forEach((name, hybridClock) -> {
            long time = IgniteTestUtils.getFieldValue(hybridClock, HybridClockImpl.class, "latestTime");

            logger().info("Cluster clock [node={}, time={}]", name, HybridTimestamp.hybridTimestamp(time));
        });

        logger().info("Replica info:");
        txTestCluster.replicaManagers().forEach((name, replicaManager) -> {
            ConcurrentHashMap<ReplicationGroupId, CompletableFuture<Replica>> replicas =
                    IgniteTestUtils.getFieldValue(replicaManager, ReplicaManager.class, "replicas");

            replicas.forEach((replicationGroupId, replicaCompletableFuture) ->
                    printReplicaInfo(name, replicationGroupId, replicaCompletableFuture)
            );

            TxManager txManager = txTestCluster.txManagers().get(name);
            VolatileTxStateMetaStorage volatileTxState =
                    IgniteTestUtils.getFieldValue(txManager, TxManagerImpl.class, "txStateVolatileStorage");

            ConcurrentHashMap<UUID, TxStateMeta> txStateMap =
                    IgniteTestUtils.getFieldValue(volatileTxState, VolatileTxStateMetaStorage.class, "txStateMap");
            logger().info("Volatile tx state data [node={}, data={}]", name, txStateMap);
        });

        TxManager txManager = txTestCluster.clientTxManager();
        VolatileTxStateMetaStorage volatileTxState =
                IgniteTestUtils.getFieldValue(txManager, TxManagerImpl.class, "txStateVolatileStorage");

        ConcurrentHashMap<UUID, TxStateMeta> txStateMap =
                IgniteTestUtils.getFieldValue(volatileTxState, VolatileTxStateMetaStorage.class, "txStateMap");
        logger().info("Volatile client tx state data [node={}, data={}]", "client", txStateMap);
    }

    private void printReplicaInfo(String name, ReplicationGroupId replicationGroupId, CompletableFuture<Replica> replicaCompletableFuture) {
        if (!replicaCompletableFuture.isDone()) {
            logger().info("Replica is not ready [node={}, groupId={}]", name, replicationGroupId);

            return;
        }

        Replica replica;
        try {
            replica = replicaCompletableFuture.get();
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
        }

        PartitionReplicaListener listener = IgniteTestUtils.getFieldValue(replica, ReplicaImpl.class, "listener");
        TestMvPartitionStorage storage = IgniteTestUtils.getFieldValue(listener, PartitionReplicaListener.class, "mvDataStorage");
        Map<RowId, ?> map = IgniteTestUtils.getFieldValue(storage, TestMvPartitionStorage.class, "map");

        PendingComparableValuesTracker<HybridTimestamp, Void> safeTime =
                IgniteTestUtils.getFieldValue(listener, PartitionReplicaListener.class, "safeTime");

        logger().info("Partition data "
                        + "[node={}, groupId={}, data={}, lastAppliedIndex={}, lastAppliedTerm={}, leaseInfo={}, safeTime = {}]",
                name, replicationGroupId, map, storage.lastAppliedIndex(), storage.lastAppliedTerm(), storage.leaseInfo(),
                safeTime.current());

        Lazy<TableSchemaAwareIndexStorage> indexStorageLazy =
                IgniteTestUtils.getFieldValue(listener, PartitionReplicaListener.class, "pkIndexStorage");
        IndexStorage indexStorage = indexStorageLazy.get().storage();
        Map<RowId, ?> indexMap = IgniteTestUtils.getFieldValue(indexStorage, TestHashIndexStorage.class, "index");

        logger().info("Index data [node={}, groupId={}, data={}]", name, replicationGroupId, indexMap);

        TxStatePartitionStorage stateStorage = IgniteTestUtils.getFieldValue(listener, PartitionReplicaListener.class, "txStateStorage");

        logger().info("Tx state data [node={}, groupId={}, data={}]", name, replicationGroupId, stateStorage);
    }

    @Test
    public void testMixedPutGet() throws TransactionException {
        accounts.recordView().upsert(null, makeValue(1, BALANCE_1));

        igniteTransactions.runInTransaction(
                tx -> {
                    var txAcc = accounts.recordView();

                    txAcc.getAsync(tx, makeKey(1)).thenCompose(r ->
                            txAcc.upsertAsync(tx, makeValue(1, r.doubleValue("balance") + DELTA))).join();
                }
        );

        assertEquals(BALANCE_1 + DELTA, accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));
    }

    @Test
    public void testLockOrdering() throws InterruptedException {
        accounts.recordView().upsert(null, makeValue(1, 50.));

        InternalTransaction tx1 = (InternalTransaction) igniteTransactions.begin();
        InternalTransaction tx2 = (InternalTransaction) igniteTransactions.begin();
        InternalTransaction tx3 = (InternalTransaction) igniteTransactions.begin();
        InternalTransaction tx4 = (InternalTransaction) igniteTransactions.begin();

        assertTrue(tx3.id().compareTo(tx4.id()) < 0);
        assertTrue(tx2.id().compareTo(tx3.id()) < 0);
        assertTrue(tx1.id().compareTo(tx2.id()) < 0);

        RecordView<Tuple> acc0 = accounts.recordView();
        RecordView<Tuple> acc2 = accounts.recordView();
        RecordView<Tuple> acc3 = accounts.recordView();
        RecordView<Tuple> acc4 = accounts.recordView();

        acc0.upsert(tx4, makeValue(1, 100.));

        CompletableFuture<Void> fut = acc3.upsertAsync(tx2, makeValue(1, 300.));

        Thread.sleep(100);

        assertFalse(fut.isDone());

        CompletableFuture<Void> fut2 = acc4.upsertAsync(tx2, makeValue(1, 400.));

        Thread.sleep(100);

        assertFalse(fut2.isDone());

        CompletableFuture<Void> fut3 = acc2.upsertAsync(tx3, makeValue(1, 200.));

        assertFalse(fut3.isDone());
    }

    /**
     * Tests a transaction closure.
     */
    @Test
    public void testTxClosure() throws TransactionException {
        RecordView<Tuple> view = accounts.recordView();

        view.upsert(null, makeValue(1, BALANCE_1));
        view.upsert(null, makeValue(2, BALANCE_2));

        igniteTransactions.runInTransaction(tx -> {
            CompletableFuture<Tuple> read1 = view.getAsync(tx, makeKey(1));
            CompletableFuture<Tuple> read2 = view.getAsync(tx, makeKey(2));

            // TODO asch IGNITE-15938 must ensure a commit happens after all pending tx async ops.
            view.upsertAsync(tx, makeValue(1, read1.join().doubleValue("balance") - DELTA)).join();
            view.upsertAsync(tx, makeValue(2, read2.join().doubleValue("balance") + DELTA)).join();
        });

        assertEquals(BALANCE_1 - DELTA, view.get(null, makeKey(1)).doubleValue("balance"));
        assertEquals(BALANCE_2 + DELTA, view.get(null, makeKey(2)).doubleValue("balance"));

        assertEquals(3, clientTxManager().finished());
        assertEquals(0, clientTxManager().pending());
    }

    /**
     * Tests a transaction closure over key-value view.
     */
    @Test
    public void testTxClosureKeyValueView() throws TransactionException {
        accounts.recordView().upsert(null, makeValue(1, BALANCE_1));
        accounts.recordView().upsert(null, makeValue(2, BALANCE_2));

        igniteTransactions.runInTransaction(tx -> {
            KeyValueView<Tuple, Tuple> view = accounts.keyValueView();

            CompletableFuture<Tuple> read1 = view.getAsync(tx, makeKey(1));
            CompletableFuture<Tuple> read2 = view.getAsync(tx, makeKey(2));

            view.putAsync(tx, makeKey(1), makeValue(read1.join().doubleValue("balance") - DELTA)).join();
            view.putAsync(tx, makeKey(2), makeValue(read2.join().doubleValue("balance") + DELTA)).join();
        });

        assertEquals(BALANCE_1 - DELTA, accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));
        assertEquals(BALANCE_2 + DELTA, accounts.recordView().get(null, makeKey(2)).doubleValue("balance"));

        assertEquals(3, clientTxManager().finished());
        assertEquals(0, clientTxManager().pending());
    }

    /**
     * Tests positive transfer scenario.
     */
    @Test
    public void testTxClosureAsync() {
        double balance1 = 200.;
        double balance2 = 300.;
        double delta = 50.;
        Tuple ret = transferAsync(balance1, balance2, delta).join();

        RecordView<Tuple> view = accounts.recordView();

        assertEquals(balance1 - delta, view.get(null, makeKey(1)).doubleValue("balance"));
        assertEquals(balance2 + delta, view.get(null, makeKey(2)).doubleValue("balance"));
        assertEquals(balance1, ret.doubleValue("balance1"));
        assertEquals(balance2, ret.doubleValue("balance2"));
    }

    /**
     * Tests negative transfer scenario.
     */
    @Test
    @Disabled("https://issues.apache.org/jira/browse/IGNITE-17861")
    public void testTxClosureAbortAsync() {
        double balance1 = 10.;
        double balance2 = 300.;
        double delta = 50.;
        assertThrows(CompletionException.class, () -> transferAsync(balance1, balance2, delta).join());

        RecordView<Tuple> view = accounts.recordView();

        assertEquals(balance1, view.get(null, makeKey(1)).doubleValue("balance"));
        assertEquals(balance2, view.get(null, makeKey(2)).doubleValue("balance"));
    }

    /**
     * Tests uncaught exception in the closure.
     */
    @Test
    public void testTxClosureUncaughtExceptionAsync() {
        double balance = 10.;
        double delta = 50.;

        RecordView<Tuple> view = accounts.recordView();
        view.upsert(null, makeValue(1, balance));

        CompletableFuture<Double> fut0 = igniteTransactions.runInTransactionAsync(
                tx -> {
                        CompletableFuture<Double> fut = view.getAsync(tx, makeKey(1))
                                .thenCompose(val2 -> {
                                    double prev = val2.doubleValue("balance");
                                    return view.upsertAsync(tx, makeValue(1, delta + 20))
                                            .thenApply(ignored -> prev);
                                })
                                .whenComplete((res, ex) -> log.info("Test: tx operations in tx closures completed, ex=" + ex));

                        if (true) {
                            throw new IllegalArgumentException();
                        }

                        return fut;
                },
                new TransactionOptions().timeoutMillis(1000)
        );

        var err = assertThrows(CompletionException.class, fut0::join);

        try {
            assertInstanceOf(IllegalArgumentException.class, err.getCause());
        } catch (AssertionError e) {
            throw new AssertionError("Unexpected exception type", err);
        }

        assertEquals(balance, view.get(null, makeKey(1)).doubleValue("balance"));
    }

    /**
     * Tests uncaught exception in the chain.
     */
    @Test
    public void testTxClosureUncaughtExceptionInChainAsync() {
        RecordView<Tuple> view = accounts.recordView();

        CompletableFuture<Double> fut0 = igniteTransactions
                .runInTransactionAsync(
                        tx -> view.getAsync(tx, makeKey(2))
                            .thenCompose(val2 -> {
                                double prev = val2.doubleValue("balance"); // val2 is null - NPE is thrown here
                                return view.upsertAsync(tx, makeValue(1, 100)).thenApply(ignored -> prev);
                            }),
                        new TransactionOptions().timeoutMillis(1000)
                );

        var err = assertThrows(CompletionException.class, fut0::join);

        try {
            assertInstanceOf(NullPointerException.class, err.getCause());
        } catch (AssertionError e) {
            throw new AssertionError("Unexpected exception type", err);
        }
    }

    @Test
    public void testBatchPutConcurrently() {
        Transaction tx1 = igniteTransactions.begin();
        Transaction tx2 = igniteTransactions.begin();

        log.info("Tx " + tx2);
        log.info("Tx2 " + tx1);

        ArrayList<Tuple> rows = new ArrayList<>();
        ArrayList<Tuple> rows2 = new ArrayList<>();

        for (int i = 0; i < 1; i++) {
            rows.add(makeValue(i, i * 100.));
            rows2.add(makeValue(i, 2 * i * 100.));
        }

        var table = accounts.recordView();
        var table2 = accounts.recordView();

        table2.upsertAll(tx1, rows2);

        Exception err = assertThrows(Exception.class, () -> table.upsertAll(tx2, rows));

        assertTrue(err.getMessage().contains("Failed to acquire a lock"), err.getMessage());

        tx1.commit();
    }

    @Test
    public void testBatchReadPutConcurrently() throws InterruptedException {
        InternalTransaction tx1 = (InternalTransaction) igniteTransactions.begin();
        InternalTransaction tx2 = (InternalTransaction) igniteTransactions.begin();

        log.info("Tx1 " + tx1);
        log.info("Tx2 " + tx2);

        var table = accounts.recordView();
        var table2 = accounts.recordView();

        ArrayList<Tuple> keys = new ArrayList<>();
        ArrayList<Tuple> keys2 = new ArrayList<>();

        for (int i = 0; i < 1; i++) {
            keys.add(makeKey(i));
            keys2.add(makeKey(i));
        }

        table2.getAll(tx1, keys);
        table2.getAll(tx2, keys2);

        ArrayList<Tuple> rows = new ArrayList<>();
        ArrayList<Tuple> rows2 = new ArrayList<>();

        for (int i = 0; i < 1; i++) {
            rows.add(makeValue(i, i * 100.));
            rows2.add(makeValue(i, 2 * i * 100.));
        }

        var futUpd2 = table2.upsertAllAsync(tx1, rows2);

        assertTrue(waitForCondition(() -> {
            boolean lockUpgraded = false;

            for (Iterator<Lock> it = txManager(accounts).lockManager().locks(tx1.id()); it.hasNext(); ) {
                Lock lock = it.next();

                lockUpgraded = txManager(accounts).lockManager().waiter(lock.lockKey(), tx1.id()).intendedLockMode() == LockMode.X;

                if (lockUpgraded) {
                    break;
                }
            }

            return lockUpgraded;
        }, 3000));

        assertFalse(futUpd2.isDone());

        assertThrowsWithCause(() -> table.upsertAll(tx2, rows), LockException.class);
    }

    /**
     * Tests an asynchronous transaction.
     */
    @Test
    public void testTxAsync() {
        accounts.recordView().upsert(null, makeValue(1, BALANCE_1));
        accounts.recordView().upsert(null, makeValue(2, BALANCE_2));

        igniteTransactions.beginAsync()
                .thenCompose(tx -> accounts.recordView().getAsync(tx, makeKey(1))
                        .thenCombine(accounts.recordView().getAsync(tx, makeKey(2)), (v1, v2) -> new Pair<>(v1, v2))
                        .thenCompose(pair -> allOf(
                                        accounts.recordView().upsertAsync(
                                                tx, makeValue(1, pair.getFirst().doubleValue("balance") - DELTA)),
                                        accounts.recordView().upsertAsync(
                                                tx, makeValue(2, pair.getSecond().doubleValue("balance") + DELTA))
                                )
                                        .thenApply(ignored -> tx)
                        )
                ).thenCompose(Transaction::commitAsync).join();

        assertEquals(BALANCE_1 - DELTA, accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));
        assertEquals(BALANCE_2 + DELTA, accounts.recordView().get(null, makeKey(2)).doubleValue("balance"));
    }

    /**
     * Tests an asynchronous transaction over key-value view.
     */
    @Test
    public void testTxAsyncKeyValueView() {
        accounts.recordView().upsert(null, makeValue(1, BALANCE_1));
        accounts.recordView().upsert(null, makeValue(2, BALANCE_2));

        igniteTransactions.beginAsync()
                .thenCompose(tx -> accounts.keyValueView().getAsync(tx, makeKey(1))
                        .thenCombine(accounts.recordView().getAsync(tx, makeKey(2)), (v1, v2) -> new Pair<>(v1, v2))
                        .thenCompose(pair -> allOf(
                                        accounts.keyValueView().putAsync(
                                                tx, makeKey(1), makeValue(pair.getFirst().doubleValue("balance") - DELTA)),
                                        accounts.keyValueView().putAsync(
                                                tx, makeKey(2), makeValue(pair.getSecond().doubleValue("balance") + DELTA))
                                )
                                        .thenApply(ignored -> tx)
                        )
                ).thenCompose(Transaction::commitAsync).join();

        assertEquals(BALANCE_1 - DELTA, accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));
        assertEquals(BALANCE_2 + DELTA, accounts.recordView().get(null, makeKey(2)).doubleValue("balance"));
    }

    @Test
    public void testSimpleConflict() {
        accounts.recordView().upsert(null, makeValue(1, 100.));

        Transaction tx1 = igniteTransactions.begin();
        Transaction tx2 = igniteTransactions.begin();

        var table = accounts.recordView();
        var table2 = accounts.recordView();

        double val = table.get(tx2, makeKey(1)).doubleValue("balance");
        table2.get(tx1, makeKey(1)).doubleValue("balance");

        try {
            table.upsert(tx2, makeValue(1, val + 1));

            fail();
        } catch (Exception e) {
            // Expected.
        }

        table2.upsert(tx1, makeValue(1, val + 1));

        tx1.commit();

        tx2.commit();

        assertEquals(101., accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));
    }

    @Test
    public void testCommit() throws TransactionException {
        InternalTransaction tx = (InternalTransaction) igniteTransactions.begin();

        Tuple key = makeKey(1);

        var table = accounts.recordView();

        table.upsert(tx, makeValue(1, 100.));

        assertEquals(100., table.get(tx, key).doubleValue("balance"));

        table.upsert(tx, makeValue(1, 200.));

        assertEquals(200., table.get(tx, key).doubleValue("balance"));

        tx.commit();

        assertEquals(200., accounts.recordView().get(null, key).doubleValue("balance"));
    }

    @Test
    public void testAbort() throws TransactionException {
        InternalTransaction tx = (InternalTransaction) igniteTransactions.begin();

        Tuple key = makeKey(1);

        var table = accounts.recordView();

        table.upsert(tx, makeValue(1, 100.));

        assertEquals(100., table.get(tx, key).doubleValue("balance"));

        table.upsert(tx, makeValue(1, 200.));

        assertEquals(200., table.get(tx, key).doubleValue("balance"));

        tx.rollback();

        assertNull(accounts.recordView().get(null, key));
    }

    @Test
    public void testAbortNoUpdate() throws TransactionException {
        accounts.recordView().upsert(null, makeValue(1, 100.));

        InternalTransaction tx = (InternalTransaction) igniteTransactions.begin();

        tx.rollback();

        assertEquals(100., accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));
    }

    @Test
    public void testConcurrent() throws TransactionException {
        Transaction tx = igniteTransactions.begin();
        Transaction tx2 = igniteTransactions.begin();

        Tuple key = makeKey(1);
        Tuple val = makeValue(1, 100.);

        accounts.recordView().upsert(null, val);

        var table = accounts.recordView();
        var table2 = accounts.recordView();

        assertEquals(100., table.get(tx, key).doubleValue("balance"));
        assertEquals(100., table2.get(tx2, key).doubleValue("balance"));

        tx.commit();
        tx2.commit();
    }

    /**
     * Tests if a lost update is not happening on concurrent increment.
     */
    @Test
    public void testIncrement() throws TransactionException {
        Transaction tx1 = igniteTransactions.begin();
        Transaction tx2 = igniteTransactions.begin();

        Tuple key = makeKey(1);
        Tuple val = makeValue(1, 100.);

        accounts.recordView().upsert(null, val); // Creates implicit transaction.

        var table = accounts.recordView();
        var table2 = accounts.recordView();

        // Read in tx2
        double valTx = table.get(tx2, key).doubleValue("balance");

        // Read in tx1
        double valTx2 = table2.get(tx1, key).doubleValue("balance");

        // Write in tx2 (out of order)
        // TODO asch IGNITE-15937 fix exception model.
        Exception err = assertThrows(Exception.class, () -> table.upsert(tx2, makeValue(1, valTx + 1)));

        assertTrue(err.getMessage().contains("Failed to acquire a lock"), err.getMessage());

        // Write in tx1
        table2.upsert(tx1, makeValue(1, valTx2 + 1));

        tx1.commit();

        assertEquals(101., accounts.recordView().get(null, key).doubleValue("balance"));
    }

    @Test
    public void testAbortWithValue() throws TransactionException {
        accounts.recordView().upsert(null, makeValue(0, 100.));

        assertEquals(100., accounts.recordView().get(null, makeKey(0)).doubleValue("balance"));

        InternalTransaction tx = (InternalTransaction) igniteTransactions.begin();
        var table = accounts.recordView();
        table.upsert(tx, makeValue(0, 200.));
        assertEquals(200., table.get(tx, makeKey(0)).doubleValue("balance"));
        tx.rollback();

        assertEquals(100., accounts.recordView().get(null, makeKey(0)).doubleValue("balance"));
    }

    @Test
    public void testInsert() throws TransactionException {
        assertNull(accounts.recordView().get(null, makeKey(1)));

        Transaction tx = igniteTransactions.begin();

        var table = accounts.recordView();

        assertTrue(table.insert(tx, makeValue(1, 100.)));
        assertFalse(table.insert(tx, makeValue(1, 200.)));
        assertEquals(100., table.get(tx, makeKey(1)).doubleValue("balance"));

        tx.commit();

        assertEquals(100., accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));

        assertTrue(accounts.recordView().insert(null, makeValue(2, 200.)));
        assertEquals(200., accounts.recordView().get(null, makeKey(2)).doubleValue("balance"));

        Transaction tx2 = igniteTransactions.begin();

        table = accounts.recordView();

        assertTrue(table.insert(tx2, makeValue(3, 100.)));
        assertFalse(table.insert(tx2, makeValue(3, 200.)));
        assertEquals(100., table.get(tx2, makeKey(3)).doubleValue("balance"));

        tx2.rollback();

        assertEquals(100., accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));
        assertEquals(200., accounts.recordView().get(null, makeKey(2)).doubleValue("balance"));
        assertNull(accounts.recordView().get(null, makeKey(3)));
    }

    @Test
    public void testDelete() throws TransactionException {
        Tuple key = makeKey(1);

        assertFalse(accounts.recordView().delete(null, key));
        assertNull(accounts.recordView().get(null, key));

        accounts.recordView().upsert(null, makeValue(1, 100.));

        igniteTransactions.runInTransaction(tx -> {
            assertNotNull(accounts.recordView().get(tx, key));
            assertTrue(accounts.recordView().delete(tx, key));
            assertNull(accounts.recordView().get(tx, key));
        });

        assertNull(accounts.recordView().get(null, key));
        accounts.recordView().upsert(null, makeValue(1, 100.));
        assertNotNull(accounts.recordView().get(null, key));

        Tuple key2 = makeKey(2);

        accounts.recordView().upsert(null, makeValue(2, 100.));

        assertThrows(
                RuntimeException.class,
                () -> igniteTransactions.runInTransaction(
                        (Consumer<Transaction>) tx -> {
                                assertNotNull(accounts.recordView().get(tx, key2));
                                assertTrue(accounts.recordView().delete(tx, key2));
                                assertNull(accounts.recordView().get(tx, key2));
                                throw new RuntimeException(); // Triggers rollback.
                        },
                        new TransactionOptions().timeoutMillis(1000)
                )
        );

        assertNotNull(accounts.recordView().get(null, key2));
        assertTrue(accounts.recordView().delete(null, key2));
        assertNull(accounts.recordView().get(null, key2));
    }

    @Test
    public void testGetAll() {
        List<Tuple> keys = List.of(makeKey(1), makeKey(2));

        Collection<Tuple> ret = accounts.recordView().getAll(null, keys);

        assertThat(ret, contains(null, null));

        accounts.recordView().upsert(null, makeValue(1, 100.));
        accounts.recordView().upsert(null, makeValue(2, 200.));

        ret = new ArrayList<>(accounts.recordView().getAll(null, keys));

        validateBalance(ret, 100., 200.);
    }

    /**
     * Tests if a transaction is rolled back if one of the batch keys can't be locked.
     */
    @Test
    public void testGetAllAbort() throws TransactionException {
        List<Tuple> keys = List.of(makeKey(1), makeKey(2));

        accounts.recordView().upsertAll(null, List.of(makeValue(1, 100.), makeValue(2, 200.)));

        Transaction tx = igniteTransactions.begin();

        RecordView<Tuple> txAcc = accounts.recordView();

        txAcc.upsert(tx, makeValue(1, 300.));
        validateBalance(txAcc.getAll(tx, keys), 300., 200.);

        tx.rollback();

        validateBalance(accounts.recordView().getAll(null, keys), 100., 200.);
    }

    /**
     * Tests if a transaction is rolled back if one of the batch keys can't be locked.
     */
    @Test
    public void testGetAllConflict() throws Exception {
        accounts.recordView().upsertAll(null, List.of(makeValue(1, 100.), makeValue(2, 200.)));

        InternalTransaction tx1 = (InternalTransaction) igniteTransactions.begin();
        InternalTransaction tx2 = (InternalTransaction) igniteTransactions.begin();

        RecordView<Tuple> txAcc = accounts.recordView();
        RecordView<Tuple> txAcc2 = accounts.recordView();

        txAcc2.upsert(tx1, makeValue(1, 300.));
        txAcc.upsert(tx2, makeValue(2, 400.));

        Exception err = assertThrows(Exception.class, () -> txAcc.getAll(tx2, List.of(makeKey(2), makeKey(1))));
        assertTrue(err.getMessage().contains("Failed to acquire a lock"), err.getMessage());

        validateBalance(txAcc2.getAll(tx1, List.of(makeKey(2), makeKey(1))), 200., 300.);
        validateBalance(txAcc2.getAll(tx1, List.of(makeKey(1), makeKey(2))), 300., 200.);

        assertTrue(waitForCondition(() -> TxState.ABORTED == tx2.state(), 5_000), tx2.state().toString());

        tx1.commit();

        validateBalance(accounts.recordView().getAll(null, List.of(makeKey(2), makeKey(1))), 200., 300.);
    }

    @Test
    public void testPutAll() throws TransactionException {
        igniteTransactions.runInTransaction(tx -> {
            accounts.recordView().upsertAll(tx, List.of(makeValue(1, 100.), makeValue(2, 200.)));
        });

        validateBalance(accounts.recordView().getAll(null, List.of(makeKey(2), makeKey(1))), 200., 100.);

        assertThrows(IgniteException.class, () -> igniteTransactions.runInTransaction(tx -> {
            accounts.recordView().upsertAll(tx, List.of(makeValue(3, 300.), makeValue(4, 400.)));
            if (true) {
                throw new IgniteException(INTERNAL_ERR, "Test error");
            }
        }, new TransactionOptions().timeoutMillis(1000)));

        assertNull(accounts.recordView().get(null, makeKey(3)));
        assertNull(accounts.recordView().get(null, makeKey(4)));
    }

    @Test
    public void testInsertAll() throws TransactionException {
        accounts.recordView().upsertAll(null, List.of(makeValue(1, 100.), makeValue(2, 200.)));

        igniteTransactions.runInTransaction(
                tx -> {
                    Collection<Tuple> res = accounts.recordView().insertAll(
                            tx,
                            List.of(makeValue(1, 200.), makeValue(3, 300.))
                    );

                    assertEquals(1, res.size());
                });

        validateBalance(
                accounts.recordView().getAll(
                        null,
                        List.of(makeKey(1), makeKey(2), makeKey(3))
                ),
                100., 200., 300.
        );
    }

    @Test
    public void testDeleteAll() throws TransactionException {
        accounts.recordView().upsertAll(null, List.of(makeValue(1, 100.), makeValue(2, 200.)));

        igniteTransactions.runInTransaction(tx -> {
            Collection<Tuple> res = accounts.recordView().deleteAll(tx, List.of(makeKey(1), makeKey(2), makeKey(3)));

            assertEquals(1, res.size());
        });

        assertNull(accounts.recordView().get(null, makeKey(1)));
        assertNull(accounts.recordView().get(null, makeKey(2)));
    }

    @Test
    public void testReplace() throws TransactionException {
        accounts.recordView().upsert(null, makeValue(1, 100.));

        igniteTransactions.runInTransaction(tx -> {
            assertFalse(accounts.recordView().replace(tx, makeValue(2, 200.)));
            assertTrue(accounts.recordView().replace(tx, makeValue(1, 200.)));
        });

        validateBalance(accounts.recordView().getAll(null, List.of(makeKey(1))), 200.);
    }

    @Test
    public void testGetAndReplace() throws TransactionException {
        accounts.recordView().upsert(null, makeValue(1, 100.));

        igniteTransactions.runInTransaction(tx -> {
            assertNull(accounts.recordView().getAndReplace(tx, makeValue(2, 200.)));
            assertNotNull(accounts.recordView().getAndReplace(tx, makeValue(1, 200.)));
        });

        validateBalance(accounts.recordView().getAll(null, List.of(makeKey(1))), 200.);
    }

    @Test
    public void testDeleteExact() throws TransactionException {
        accounts.recordView().upsert(null, makeValue(1, 100.));

        igniteTransactions.runInTransaction(tx -> {
            assertFalse(accounts.recordView().deleteExact(tx, makeValue(1, 200.)));
            assertTrue(accounts.recordView().deleteExact(tx, makeValue(1, 100.)));
        });

        Tuple actual = accounts.recordView().get(null, makeKey(1));

        assertNull(actual);
    }

    @Test
    public void testDeleteAllExact() throws TransactionException {
        accounts.recordView().upsertAll(null, List.of(makeValue(1, 100.), makeValue(2, 200.)));

        igniteTransactions.runInTransaction(
                tx -> {
                    Collection<Tuple> res = accounts.recordView().deleteAllExact(
                            tx,
                            List.of(makeValue(1, 200.), makeValue(2, 200.), makeValue(3, 300.))
                    );

                    assertEquals(2, res.size());
                });

        assertNotNull(accounts.recordView().get(null, makeKey(1)));
        assertNull(accounts.recordView().get(null, makeKey(2)));
    }

    @Test
    public void testGetAndPut() throws TransactionException {
        accounts.recordView().upsert(null, makeValue(1, 100.));

        igniteTransactions.runInTransaction(tx -> {
            assertNotNull(accounts.recordView().getAndUpsert(tx, makeValue(1, 200.)));
            assertNull(accounts.recordView().getAndUpsert(tx, makeValue(2, 200.)));
        });

        validateBalance(accounts.recordView().getAll(null, List.of(makeKey(1), makeKey(2))), 200., 200.);
    }

    @Test
    public void testGetAndDelete() throws TransactionException {
        accounts.recordView().upsert(null, makeValue(1, 100.));

        igniteTransactions.runInTransaction(tx -> {
            assertEquals(100., accounts.recordView().getAndDelete(tx, makeKey(1)).doubleValue("balance"));
        });

        assertNull(accounts.recordView().get(null, makeKey(1)));
    }

    @Test
    public void testRollbackUpgradedLock() throws Exception {
        accounts.recordView().upsert(null, makeValue(1, 100.));

        InternalTransaction tx = (InternalTransaction) igniteTransactions.begin();
        InternalTransaction tx2 = (InternalTransaction) igniteTransactions.begin();

        var table = accounts.recordView();
        var table2 = accounts.recordView();

        double v0 = table.get(tx, makeKey(1)).doubleValue("balance");
        double v1 = table2.get(tx2, makeKey(1)).doubleValue("balance");

        assertEquals(v0, v1);

        // Try to upgrade a lock.
        table2.upsertAsync(tx2, makeValue(1, v0 + 10));
        Thread.sleep(300); // Give some time to update lock queue TODO asch IGNITE-15928

        tx2.rollback();
    }

    @Test
    @Disabled("https://issues.apache.org/jira/browse/IGNITE-15938") // TODO asch IGNITE-15938
    public void testUpgradedLockInvalidation() throws Exception {
        accounts.recordView().upsert(null, makeValue(1, 100.));

        InternalTransaction tx = (InternalTransaction) igniteTransactions.begin();
        InternalTransaction tx2 = (InternalTransaction) igniteTransactions.begin();

        var table = accounts.recordView();
        var table2 = accounts.recordView();

        double v0 = table.get(tx, makeKey(1)).doubleValue("balance");
        double v1 = table2.get(tx2, makeKey(1)).doubleValue("balance");

        assertEquals(v0, v1);

        // Try to upgrade a lock.
        table2.upsertAsync(tx2, makeValue(1, v0 + 10));
        Thread.sleep(300); // Give some time to update lock queue TODO asch IGNITE-15928

        table.upsert(tx, makeValue(1, v0 + 20));

        tx.commit();
        assertThrows(Exception.class, () -> tx2.commit());
    }

    @Test
    public void testReorder() {
        accounts.recordView().upsert(null, makeValue(1, 100.));

        InternalTransaction tx1 = (InternalTransaction) igniteTransactions.begin();
        InternalTransaction tx2 = (InternalTransaction) igniteTransactions.begin();
        InternalTransaction tx3 = (InternalTransaction) igniteTransactions.begin();

        var table = accounts.recordView();
        var table2 = accounts.recordView();
        var table3 = accounts.recordView();

        double v0 = table.get(tx3, makeKey(1)).doubleValue("balance");

        table.upsertAsync(tx3, makeValue(1, v0 + 10));

        CompletableFuture<Tuple> fut = table2.getAsync(tx2, makeKey(1));
        assertFalse(fut.isDone());

        CompletableFuture<Tuple> fut2 = table3.getAsync(tx1, makeKey(1));
        assertFalse(fut2.isDone());
    }

    @Test
    public void testCrossTable() throws Exception {
        customers.recordView().upsert(null, makeValue(1, "test"));
        accounts.recordView().upsert(null, makeValue(1, 100.));

        assertEquals("test", customers.recordView().get(null, makeKey(1)).stringValue("name"));
        assertEquals(100., accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));

        InternalTransaction tx = (InternalTransaction) igniteTransactions.begin();

        var txCust = customers.recordView();
        var txAcc = accounts.recordView();

        txCust.upsert(tx, makeValue(1, "test2"));
        txAcc.upsert(tx, makeValue(1, 200.));

        Tuple txValCust = txCust.get(tx, makeKey(1));
        assertEquals("test2", txValCust.stringValue("name"));

        txValCust.set("accountNumber", 2L);
        txValCust.set("name", "test3");

        Tuple txValAcc = txAcc.get(tx, makeKey(1));
        assertEquals(200., txValAcc.doubleValue("balance"));

        txValAcc.set("accountNumber", 2L);
        txValAcc.set("balance", 300.);

        tx.commit();

        assertEquals("test2", customers.recordView().get(null, makeKey(1)).stringValue("name"));
        assertEquals(200., accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));

        assertTrue(waitForCondition(() -> lockManager(accounts).isEmpty(), 10_000));
        assertTrue(waitForCondition(() -> lockManager(customers).isEmpty(), 10_000));
    }

    @Test
    public void testTwoTables() throws Exception {
        customers.recordView().upsert(null, makeValue(1, "test"));
        accounts.recordView().upsert(null, makeValue(1, 100.));

        assertEquals("test", customers.recordView().get(null, makeKey(1)).stringValue("name"));
        assertEquals(100., accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));

        InternalTransaction tx = (InternalTransaction) igniteTransactions.begin();
        InternalTransaction tx2 = (InternalTransaction) igniteTransactions.begin();

        var txCust = customers.recordView();
        var txAcc = accounts.recordView();

        txCust.upsert(tx, makeValue(1, "test2"));
        txAcc.upsert(tx, makeValue(1, 200.));

        Tuple txValCust = txCust.get(tx, makeKey(1));
        assertEquals("test2", txValCust.stringValue("name"));

        txValCust.set("accountNumber", 2L);
        txValCust.set("name", "test3");

        Tuple txValAcc = txAcc.get(tx, makeKey(1));
        assertEquals(200., txValAcc.doubleValue("balance"));

        txValAcc.set("accountNumber", 2L);
        txValAcc.set("balance", 300.);

        tx.commit();
        tx2.commit();

        assertEquals("test2", customers.recordView().get(null, makeKey(1)).stringValue("name"));
        assertEquals(200., accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));

        assertTrue(waitForCondition(() -> lockManager(accounts).isEmpty(), 10_000));
    }

    @Test
    public void testCrossTableKeyValueView() throws Exception {
        customers.recordView().upsert(null, makeValue(1L, "test"));
        accounts.recordView().upsert(null, makeValue(1L, 100.));

        assertEquals("test", customers.recordView().get(null, makeKey(1)).stringValue("name"));
        assertEquals(100., accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));

        InternalTransaction tx = (InternalTransaction) igniteTransactions.begin();
        InternalTransaction tx2 = (InternalTransaction) igniteTransactions.begin();

        var txCust = customers.keyValueView();
        var txAcc = accounts.keyValueView();

        txCust.put(tx, makeKey(1), makeValue("test2"));
        txAcc.put(tx, makeKey(1), makeValue(200.));

        Tuple txValCust = txCust.get(tx, makeKey(1));
        assertEquals("test2", txValCust.stringValue("name"));

        txValCust.set("accountNumber", 2L);
        txValCust.set("name", "test3");

        Tuple txValAcc = txAcc.get(tx, makeKey(1));
        assertEquals(200., txValAcc.doubleValue("balance"));

        txValAcc.set("accountNumber", 2L);
        txValAcc.set("balance", 300.);

        tx.commit();
        tx2.commit();

        assertEquals("test2", customers.recordView().get(null, makeKey(1)).stringValue("name"));
        assertEquals(200., accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));

        assertTrue(waitForCondition(() -> lockManager(accounts).isEmpty(), 10_000));
    }

    @Test
    public void testCrossTableAsync() throws Exception {
        customers.recordView().upsert(null, makeValue(1, "test"));
        accounts.recordView().upsert(null, makeValue(1, 100.));

        igniteTransactions.beginAsync()
                .thenCompose(
                        tx -> accounts.recordView().upsertAsync(tx, makeValue(1, 200.))
                                .thenCombine(customers.recordView().upsertAsync(tx, makeValue(1, "test2")), (v1, v2) -> tx)
                )
                .thenCompose(Transaction::commitAsync).join();

        assertEquals("test2", customers.recordView().get(null, makeKey(1)).stringValue("name"));
        assertEquals(200., accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));

        assertTrue(waitForCondition(() -> lockManager(accounts).isEmpty(), 10_000));
    }

    @Test
    public void testCrossTableAsyncRollback() throws Exception {
        customers.recordView().upsert(null, makeValue(1, "test"));
        accounts.recordView().upsert(null, makeValue(1, 100.));

        igniteTransactions.beginAsync()
                .thenCompose(
                        tx -> accounts.recordView().upsertAsync(tx, makeValue(1, 200.))
                                .thenCombine(customers.recordView().upsertAsync(tx, makeValue(1, "test2")), (v1, v2) -> tx)
                )
                .thenCompose(Transaction::rollbackAsync).join();

        assertEquals("test", customers.recordView().get(null, makeKey(1)).stringValue("name"));
        assertEquals(100., accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));

        assertTrue(waitForCondition(() -> lockManager(accounts).isEmpty(), 10_000));
    }

    @Test
    public void testCrossTableAsyncKeyValueView() throws Exception {
        customers.recordView().upsert(null, makeValue(1, "test"));
        accounts.recordView().upsert(null, makeValue(1, 100.));

        igniteTransactions.beginAsync()
                .thenCompose(
                        tx -> accounts.keyValueView().putAsync(tx, makeKey(1), makeValue(200.))
                                .thenCombine(customers.keyValueView().putAsync(tx, makeKey(1), makeValue("test2")),
                                        (v1, v2) -> tx)
                )
                .thenCompose(Transaction::commitAsync).join();

        assertEquals("test2", customers.recordView().get(null, makeKey(1)).stringValue("name"));
        assertEquals(200., accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));

        assertTrue(waitForCondition(() -> lockManager(accounts).isEmpty(), 10_000));
    }

    @Test
    public void testCrossTableAsyncKeyValueViewRollback() throws Exception {
        customers.recordView().upsert(null, makeValue(1, "test"));
        accounts.recordView().upsert(null, makeValue(1, 100.));

        igniteTransactions.beginAsync()
                .thenCompose(
                        tx -> accounts.keyValueView().putAsync(tx, makeKey(1), makeValue(200.))
                                .thenCombine(customers.keyValueView().putAsync(tx, makeKey(1), makeValue("test2")),
                                        (v1, v2) -> tx)
                )
                .thenCompose(Transaction::rollbackAsync).join();

        assertEquals("test", customers.recordView().get(null, makeKey(1)).stringValue("name"));
        assertEquals(100., accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));

        assertTrue(waitForCondition(() -> lockManager(accounts).isEmpty(), 10_000));
    }

    @Test
    public void testBalance() throws InterruptedException {
        doTestSingleKeyMultithreaded(5_000, false);
    }

    @Test
    public void testLockedTooLong() {
        // TODO asch IGNITE-15936 if lock can't be acquired until timeout tx should be rolled back.
    }

    @Test
    public void testScan() {
        doTestScan(null);
    }

    @Test
    public void testScanExplicit() {
        igniteTransactions.runInTransaction(this::doTestScan);
    }

    /**
     * Do scan in test.
     *
     * @param tx The transaction.
     */
    private void doTestScan(@Nullable Transaction tx) {
        accounts.recordView().upsertAll(tx, List.of(makeValue(1, 100.), makeValue(2, 200.)));

        CompletableFuture<List<Tuple>> scanFut = scan(accounts.internalTable(), tx == null ? null : (InternalTransaction) tx);

        var rows = scanFut.join();

        Map<Long, Tuple> map = new HashMap<>();

        for (Tuple row : rows) {
            map.put(row.longValue("accountNumber"), row);
        }

        assertEquals(100., map.get(1L).doubleValue("balance"));
        assertEquals(200., map.get(2L).doubleValue("balance"));

        // Attempt to overwrite.
        accounts.recordView().upsertAll(tx, List.of(makeValue(1, 300.), makeValue(2, 400.)));
    }

    /**
     * Scans {@code 0} partition of a table in a specific transaction or implicit one.
     *
     * @param internalTable Internal table to scanning.
     * @param internalTx Internal transaction of {@code null}.
     * @return Future to scanning result.
     */
    private CompletableFuture<List<Tuple>> scan(InternalTable internalTable, @Nullable InternalTransaction internalTx) {
        Flow.Publisher<BinaryRow> pub = internalTx != null && internalTx.isReadOnly()
                ?
                internalTable.scan(
                        0,
                        internalTx.id(),
                        internalTx.readTimestamp(),
                        ReplicaTestUtils.leaderAssignment(
                                txTestCluster.replicaManagers().get(txTestCluster.localNodeName()),
                                txTestCluster.clusterServices().get(txTestCluster.localNodeName()).topologyService(),
                                colocationEnabled() ? internalTable.zoneId() : internalTable.tableId(),
                                0
                        ),
                        internalTx.coordinatorId()
                )
                : internalTable.scan(0, internalTx);

        List<Tuple> rows = new ArrayList<>();

        var fut = new CompletableFuture<List<Tuple>>();

        pub.subscribe(new Flow.Subscriber<>() {
            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                subscription.request(3);
            }

            @Override
            public void onNext(BinaryRow item) {
                SchemaRegistry registry = accounts.schemaView();
                Row row = registry.resolve(item, registry.lastKnownSchema());

                rows.add(TableRow.tuple(row));
            }

            @Override
            public void onError(Throwable throwable) {
                // No-op.
            }

            @Override
            public void onComplete() {
                fut.complete(rows);
            }
        });

        return fut;
    }

    @Test
    public void testComplexImplicit() {
        doTestComplex(accounts.recordView(), null);
    }

    @Test
    public void testComplexExplicit() throws TransactionException {
        doTestComplex(accounts.recordView(), igniteTransactions.begin());
    }

    @Test
    public void testComplexImplicitKeyValueView() {
        doTestComplexKeyValue(accounts.keyValueView(), null);
    }

    @Test
    public void testComplexExplicitKeyValueView() throws TransactionException {
        doTestComplexKeyValue(accounts.keyValueView(), igniteTransactions.begin());
    }

    /**
     * Checks operation over tuple record view. The scenario was moved from ITDistributedTableTest.
     *
     * @param view Record view.
     * @param tx Transaction or {@code null} for implicit one.
     */
    private void doTestComplex(RecordView<Tuple> view, @Nullable Transaction tx) {
        final int keysCnt = 10;

        long start = System.nanoTime();

        for (long i = 0; i < keysCnt; i++) {
            view.insert(tx, makeValue(i, i + 2.));
        }

        long dur = (long) ((System.nanoTime() - start) / 1000 / 1000.);

        log.info("Inserted={}, time={}ms  avg={} tps={}", keysCnt, dur, dur / keysCnt, 1000 / (dur / (float) keysCnt));

        for (long i = 0; i < keysCnt; i++) {
            Tuple entry = view.get(tx, makeKey(i));

            assertEquals(i + 2., entry.doubleValue("balance"));
        }

        for (int i = 0; i < keysCnt; i++) {
            view.upsert(tx, makeValue(i, i + 5.));

            Tuple entry = view.get(tx, makeKey(i));

            assertEquals(i + 5., entry.doubleValue("balance"));
        }

        HashSet<Tuple> keys = new HashSet<>();

        for (long i = 0; i < keysCnt; i++) {
            keys.add(makeKey(i));
        }

        Collection<Tuple> entries = view.getAll(tx, keys);

        assertEquals(keysCnt, entries.size());

        for (long i = 0; i < keysCnt; i++) {
            boolean res = view.replace(tx, makeValue(i, i + 5.), makeValue(i, i + 2.));

            assertTrue(res, "Failed to replace for idx=" + i);
        }

        for (long i = 0; i < keysCnt; i++) {
            boolean res = view.delete(tx, makeKey(i));

            assertTrue(res);

            Tuple entry = view.get(tx, makeKey(i));

            assertNull(entry);
        }

        ArrayList<Tuple> batch = new ArrayList<>(keysCnt);

        for (long i = 0; i < keysCnt; i++) {
            batch.add(makeValue(i, i + 2.));
        }

        view.upsertAll(tx, batch);

        for (long i = 0; i < keysCnt; i++) {
            Tuple entry = view.get(tx, makeKey(i));

            assertEquals(i + 2., entry.doubleValue("balance"));
        }

        view.deleteAll(tx, keys);

        for (Tuple key : keys) {
            Tuple entry = view.get(tx, key);

            assertNull(entry);
        }

        if (tx != null) {
            tx.commit();
        }
    }

    /**
     * Checks operation over tuple key value view. The scenario was moved from ITDistributedTableTest.
     *
     * @param view Table view.
     * @param tx Transaction or {@code null} for implicit one.
     */
    public void doTestComplexKeyValue(KeyValueView<Tuple, Tuple> view, @Nullable Transaction tx) {
        final int keysCnt = 10;

        for (long i = 0; i < keysCnt; i++) {
            view.put(tx, makeKey(i), makeValue(i + 2.));
        }

        for (long i = 0; i < keysCnt; i++) {
            Tuple entry = view.get(tx, makeKey(i));

            assertEquals(i + 2., entry.doubleValue("balance"));
        }

        for (int i = 0; i < keysCnt; i++) {
            view.put(tx, makeKey(i), makeValue(i + 5.));

            Tuple entry = view.get(tx, makeKey(i));

            assertEquals(i + 5., entry.doubleValue("balance"));
        }

        HashSet<Tuple> keys = new HashSet<>();

        for (long i = 0; i < keysCnt; i++) {
            keys.add(makeKey(i));
        }

        Map<Tuple, Tuple> entries = view.getAll(tx, keys);

        assertEquals(keysCnt, entries.size());

        for (long i = 0; i < keysCnt; i++) {
            boolean res = view.replace(tx, makeKey(i), makeValue(i + 5.), makeValue(i + 2.));

            assertTrue(res, "Failed to replace for idx=" + i);
        }

        for (long i = 0; i < keysCnt; i++) {
            boolean res = view.remove(tx, makeKey(i));

            assertTrue(res);

            Tuple entry = view.get(tx, makeKey(i));

            assertNull(entry);
        }

        Map<Tuple, Tuple> batch = new LinkedHashMap<>(keysCnt);

        for (long i = 0; i < keysCnt; i++) {
            batch.put(makeKey(i), makeValue(i + 2.));
        }

        view.putAll(tx, batch);

        for (long i = 0; i < keysCnt; i++) {
            Tuple entry = view.get(tx, makeKey(i));

            assertEquals(i + 2., entry.doubleValue("balance"));
        }

        view.removeAll(tx, keys);

        for (Tuple key : keys) {
            Tuple entry = view.get(tx, key);

            assertNull(entry);
        }

        if (tx != null) {
            tx.commit();
        }
    }

    /**
     * Performs a test.
     *
     * @param duration The duration.
     * @param verbose Verbose mode.
     * @throws InterruptedException If interrupted while waiting.
     */
    private void doTestSingleKeyMultithreaded(long duration, boolean verbose) throws InterruptedException {
        int threadsCnt = Runtime.getRuntime().availableProcessors() * 2;

        Thread[] threads = new Thread[threadsCnt];

        final int accountsCount = threads.length * 10;

        final double initial = 1000;
        final double total = accountsCount * initial;

        for (int i = 0; i < accountsCount; i++) {
            accounts.recordView().upsert(null, makeValue(i, 1000));
        }

        double total0 = 0;

        for (long i = 0; i < accountsCount; i++) {
            double balance = accounts.recordView().get(null, makeKey(i)).doubleValue("balance");

            total0 += balance;
        }

        assertEquals(total, total0, "Total amount invariant is not preserved");

        CyclicBarrier startBar = new CyclicBarrier(threads.length, () -> log.info("Before test"));

        LongAdder ops = new LongAdder();
        LongAdder fails = new LongAdder();

        AtomicBoolean stop = new AtomicBoolean();

        Random r = new Random();

        AtomicReference<Throwable> firstErr = new AtomicReference<>();

        for (int i = 0; i < threads.length; i++) {
            threads[i] = new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        startBar.await();
                    } catch (Exception e) {
                        fail();
                    }

                    while (!stop.get() && firstErr.get() == null) {
                        InternalTransaction tx = clientTxManager().beginExplicitRw(timestampTracker, InternalTxOptions.defaults());

                        var table = accounts.recordView();

                        try {
                            long acc1 = r.nextInt(accountsCount);

                            double amount = 100 + r.nextInt(500);

                            if (verbose) {
                                log.info("op=tryGet ts={} id={}", tx.id(), acc1);
                            }

                            double val0 = table.get(tx, makeKey(acc1)).doubleValue("balance");

                            long acc2 = acc1;

                            while (acc1 == acc2) {
                                acc2 = r.nextInt(accountsCount);
                            }

                            if (verbose) {
                                log.info("op=tryGet ts={} id={}", tx.id(), acc2);
                            }

                            double val1 = table.get(tx, makeKey(acc2)).doubleValue("balance");

                            if (verbose) {
                                log.info("op=tryPut ts={} id={}", tx.id(), acc1);
                            }

                            table.upsert(tx, makeValue(acc1, val0 - amount));

                            if (verbose) {
                                log.info("op=tryPut ts={} id={}", tx.id(), acc2);
                            }

                            table.upsert(tx, makeValue(acc2, val1 + amount));

                            tx.commit();

                            ops.increment();
                        } catch (Exception e) {
                            assertTrue(e.getMessage().contains("Failed to acquire a lock"), e.getMessage());

                            tx.rollback();

                            fails.increment();
                        }
                    }
                }
            });

            threads[i].setName("Worker-" + i);
            threads[i].setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
                @Override
                public void uncaughtException(Thread t, Throwable e) {
                    firstErr.compareAndExchange(null, e);
                }
            });
            threads[i].start();
        }

        Thread.sleep(duration);

        stop.set(true);

        for (Thread thread : threads) {
            thread.join(3_000);
        }

        if (firstErr.get() != null) {
            throw new IgniteException(INTERNAL_ERR, firstErr.get());
        }

        log.info("After test ops={} fails={}", ops.sum(), fails.sum());

        total0 = 0;

        for (long i = 0; i < accountsCount; i++) {
            double balance = accounts.recordView().get(null, makeKey(i)).doubleValue("balance");

            total0 += balance;
        }

        assertEquals(total, total0, "Total amount invariant is not preserved");
    }

    /**
     * Get a lock manager on a partition leader.
     *
     * @param t The table.
     * @return Lock manager.
     */
    protected LockManager lockManager(TableViewInternal t) {
        return txManager(t).lockManager();
    }

    /**
     * Validates balances.
     *
     * @param rows Rows.
     * @param expected Expected values.
     */
    protected static void validateBalance(Collection<Tuple> rows, @Nullable Double... expected) {
        assertThat(
                rows.stream().map(tuple -> tuple == null ? null : tuple.doubleValue("balance")).collect(toList()),
                contains(expected)
        );
    }

    /**
     * Transfers money between accounts.
     *
     * @param balance1 First account initial balance.
     * @param balance2 Second account initial balance.
     * @param delta Delta.
     * @return The future holding tuple with previous balances.
     */
    private CompletableFuture<Tuple> transferAsync(double balance1, double balance2, double delta) {
        RecordView<Tuple> view = accounts.recordView();

        view.upsert(null, makeValue(1, balance1));
        view.upsert(null, makeValue(2, balance2));

        return igniteTransactions.runInTransactionAsync(tx -> {
            // Attempt to withdraw from first account.
            CompletableFuture<Double> fut1 = view.getAsync(tx, makeKey(1))
                    .thenCompose(val1 -> {
                        double prev = val1.doubleValue("balance");
                        double balance = prev - delta;

                        if (balance < 0) {
                            return tx.rollbackAsync().thenApply(ignored -> null);
                        }

                        return view.upsertAsync(tx, makeValue(1, balance)).thenApply(ignored -> prev);
                    });

            // Optimistically deposit to second account.
            CompletableFuture<Double> fut2 = view.getAsync(tx, makeKey(2))
                    .thenCompose(val2 -> {
                        double prev = val2.doubleValue("balance");
                        return view.upsertAsync(tx, makeValue(2, prev + delta)).thenApply(ignored -> prev);
                    });

            return fut1.thenCompose(val1 -> fut2.thenCompose(val2 ->
                    completedFuture(Tuple.create().set("balance1", val1).set("balance2", val2))));
        });
    }

    @Test
    public void testReadOnlyGet() {
        accounts.recordView().upsert(null, makeValue(1, 100.));

        Transaction readOnlyTx = igniteTransactions.begin(new TransactionOptions().readOnly(true));
        assertEquals(100., accounts.recordView().get(readOnlyTx, makeKey(1)).doubleValue("balance"));
    }

    @Test
    public void testReadOnlyScan() throws Exception {
        accounts.recordView().upsert(null, makeValue(1, 100.));
        accounts.recordView().upsert(null, makeValue(2, 500.));

        // Pending tx
        Transaction tx = igniteTransactions.begin();
        accounts.recordView().upsert(tx, makeValue(1, 300.));
        accounts.recordView().delete(tx, makeKey(2));

        InternalTransaction readOnlyTx = (InternalTransaction) igniteTransactions.begin(new TransactionOptions().readOnly(true));

        CompletableFuture<List<Tuple>> roBeforeCommitTxFut = scan(accounts.internalTable(), readOnlyTx);

        var roBeforeCommitTxRows = roBeforeCommitTxFut.get(10, TimeUnit.SECONDS);

        assertEquals(2, roBeforeCommitTxRows.size());

        for (Tuple row : roBeforeCommitTxRows) {
            if (row.longValue("accountNumber") == 1) {
                assertEquals(100., row.doubleValue("balance"));
            } else {
                assertEquals(2, row.longValue("accountNumber"));
                assertEquals(500., row.doubleValue("balance"));
            }
        }

        // Commit pending tx.
        tx.commit();

        // Same read-only transaction.
        roBeforeCommitTxFut = scan(accounts.internalTable(), readOnlyTx);

        roBeforeCommitTxRows = roBeforeCommitTxFut.get(10, TimeUnit.SECONDS);

        assertEquals(2, roBeforeCommitTxRows.size());

        for (Tuple row : roBeforeCommitTxRows) {
            if (row.longValue("accountNumber") == 1) {
                assertEquals(100., row.doubleValue("balance"));
            } else {
                assertEquals(2, row.longValue("accountNumber"));
                assertEquals(500., row.doubleValue("balance"));
            }
        }

        // New read-only transaction.
        InternalTransaction readOnlyTx2 = (InternalTransaction) igniteTransactions.begin(new TransactionOptions().readOnly(true));

        CompletableFuture<List<Tuple>> roAfterCommitTxFut = scan(accounts.internalTable(), readOnlyTx2);

        var roAfterCommitTxRows = roAfterCommitTxFut.get(10, TimeUnit.SECONDS);

        assertEquals(1, roAfterCommitTxRows.size());

        for (Tuple row : roAfterCommitTxRows) {
            assertEquals(1, row.longValue("accountNumber"));
            assertEquals(300., row.doubleValue("balance"));
        }
    }

    @Test
    public void testReadOnlyGetWriteIntentResolutionUpdate() {
        accounts.recordView().upsert(null, makeValue(1, 100.));

        // Pending tx
        Transaction tx = igniteTransactions.begin();
        accounts.recordView().upsert(tx, makeValue(1, 300.));

        // Update
        Transaction readOnlyTx = igniteTransactions.begin(new TransactionOptions().readOnly(true));
        assertEquals(100., accounts.recordView().get(readOnlyTx, makeKey(1)).doubleValue("balance"));

        // Commit pending tx.
        tx.commit();

        // Same read-only transaction.
        assertEquals(100., accounts.recordView().get(readOnlyTx, makeKey(1)).doubleValue("balance"));

        // New read-only transaction.
        Transaction readOnlyTx2 = igniteTransactions.begin(new TransactionOptions().readOnly(true));
        assertEquals(300., accounts.recordView().get(readOnlyTx2, makeKey(1)).doubleValue("balance"));
    }

    @Test
    public void testReadOnlyGetWriteIntentResolutionRemove() {
        accounts.recordView().upsert(null, makeValue(1, 100.));

        // Pending tx
        Transaction tx = igniteTransactions.begin();
        accounts.recordView().delete(tx, makeKey(1));

        // Remove.
        Transaction readOnlyTx = igniteTransactions.begin(new TransactionOptions().readOnly(true));
        assertEquals(100., accounts.recordView().get(readOnlyTx, makeKey(1)).doubleValue("balance"));

        // Commit pending tx.
        tx.commit();

        // Same read-only transaction.
        assertEquals(100., accounts.recordView().get(readOnlyTx, makeKey(1)).doubleValue("balance"));

        // New read-only transaction.
        Transaction readOnlyTx2 = igniteTransactions.begin(new TransactionOptions().readOnly(true));
        Tuple row = accounts.recordView().get(readOnlyTx2, makeKey(1));
        assertNull(row);
    }

    @Test
    public void testReadOnlyGetAll() {
        accounts.recordView().upsert(null, makeValue(1, 100.));
        accounts.recordView().upsert(null, makeValue(2, 200.));
        accounts.recordView().upsert(null, makeValue(3, 300.));

        Transaction readOnlyTx = igniteTransactions.begin(new TransactionOptions().readOnly(true));
        Collection<Tuple> retrievedKeys = accounts.recordView().getAll(readOnlyTx, List.of(makeKey(1), makeKey(2)));
        validateBalance(retrievedKeys, 100., 200.);
    }

    @Test
    public void testReadOnlyPendingWriteIntentSkippedCombined() {
        accounts.recordView().upsert(null, makeValue(1, 100.));
        accounts.recordView().upsert(null, makeValue(2, 200.));

        // Pending tx
        Transaction tx = igniteTransactions.begin();
        accounts.recordView().delete(tx, makeKey(1));
        accounts.recordView().upsert(tx, makeValue(2, 300.));

        Transaction readOnlyTx = igniteTransactions.begin(new TransactionOptions().readOnly(true));
        Collection<Tuple> retrievedKeys = accounts.recordView().getAll(readOnlyTx, List.of(makeKey(1), makeKey(2)));
        validateBalance(retrievedKeys, 100., 200.);

        // Commit pending tx.
        tx.commit();

        Collection<Tuple> retrievedKeys2 = accounts.recordView().getAll(readOnlyTx, List.of(makeKey(1), makeKey(2)));
        validateBalance(retrievedKeys2, 100., 200.);

        Transaction readOnlyTx2 = igniteTransactions.begin(new TransactionOptions().readOnly(true));
        Collection<Tuple> retrievedKeys3 = accounts.recordView().getAll(readOnlyTx2, List.of(makeKey(1), makeKey(2)));
        validateBalance(retrievedKeys3, null, 300.);
    }

    @Test
    public void testTransactionAlreadyCommitted() {
        testTransactionAlreadyFinished(true, true, (transaction, uuid) -> {
            transaction.commit();

            log.info("Committed transaction {}", uuid);
        });
    }

    @Test
    public void testTransactionAlreadyRolledback() {
        testTransactionAlreadyFinished(false, true, (transaction, uuid) -> {
            transaction.rollback();

            log.info("Rolled back transaction {}", uuid);
        });
    }

    @Test
    public void testReadOnlyTransactionAlreadyFinished() throws InterruptedException {
        Transaction tx = igniteTransactions.begin(new TransactionOptions().readOnly(true));

        tx.rollback();

        RecordView accountsRv = accounts.recordView();

        assertThrowsTxFinishedException(() -> accountsRv.get(tx, makeKey(1)));
        assertThrowsTxFinishedException(() -> accountsRv.contains(tx, makeKey(1)));
        assertThrowsTxFinishedException(() -> accountsRv.getAll(tx, List.of(makeKey(1))));
        assertThrowsTxFinishedException(() -> accountsRv.containsAll(tx, List.of(makeKey(1))));
        assertAsyncThrowsTxFinishedException(() -> accountsRv.getAsync(tx, makeKey(1)));
        assertAsyncThrowsTxFinishedException(() -> accountsRv.containsAsync(tx, makeKey(1)));
        assertAsyncThrowsTxFinishedException(() -> accountsRv.getAllAsync(tx, List.of(makeKey(1))));
        assertAsyncThrowsTxFinishedException(() -> accountsRv.containsAllAsync(tx, List.of(makeKey(1))));

        KeyValueView accountsKv = accounts.keyValueView();

        assertThrowsTxFinishedException(() -> accountsKv.get(tx, makeKey(1)));
        assertThrowsTxFinishedException(() -> accountsKv.getNullable(tx, makeKey(1)));
        assertThrowsTxFinishedException(() -> accountsKv.getOrDefault(tx, makeKey(1), makeValue(1, BALANCE_1)));
        assertThrowsTxFinishedException(() -> accountsKv.contains(tx, makeKey(1)));
        assertThrowsTxFinishedException(() -> accountsKv.getAll(tx, List.of(makeKey(1))));
        assertThrowsTxFinishedException(() -> accountsKv.containsAll(tx, List.of(makeKey(1))));
        assertAsyncThrowsTxFinishedException(() -> accountsKv.getAsync(tx, makeKey(1)));
        assertAsyncThrowsTxFinishedException(() -> accountsKv.getNullableAsync(tx, makeKey(1)));
        assertAsyncThrowsTxFinishedException(() -> accountsKv.getOrDefaultAsync(tx, makeKey(1), makeValue(1, BALANCE_1)));
        assertAsyncThrowsTxFinishedException(() -> accountsKv.getAllAsync(tx, List.of(makeKey(1))));
        assertAsyncThrowsTxFinishedException(() -> accountsKv.containsAsync(tx, makeKey(1)));
        assertAsyncThrowsTxFinishedException(() -> accountsKv.containsAllAsync(tx, List.of(makeKey(1))));

        InternalTransaction internalTx = (InternalTransaction) tx;

        assertThrowsTxFinishedException(() -> {
            Flow.Publisher<BinaryRow> pub = accounts.internalTable().scan(
                    0,
                    internalTx.id(),
                    internalTx.readTimestamp(),
                    new ClusterNodeImpl(UUID.randomUUID(), "node", new NetworkAddress("localhost", 123)),
                    internalTx.coordinatorId()
            );

            AtomicReference<Throwable> errorRef = new AtomicReference<>();

            pub.subscribe(new SingleRequestSubscriber<>(errorRef));

            assertTrue(waitForCondition(() -> errorRef.get() != null, 1_000));

            throw errorRef.get();
        });

        assertThrowsTxFinishedException(() -> {
            Flow.Publisher<BinaryRow> pub = accounts.internalTable().lookup(
                    0,
                    internalTx.id(),
                    internalTx.readTimestamp(),
                    new ClusterNodeImpl(UUID.randomUUID(), "node", new NetworkAddress("localhost", 123)),
                    0,
                    // Binary tuple is null for testing purposes, assuming that it wouldn't be processed anyway.
                    null,
                    null,
                    internalTx.coordinatorId()
            );

            AtomicReference<Throwable> errorRef = new AtomicReference<>();

            pub.subscribe(new SingleRequestSubscriber<>(errorRef));

            assertTrue(waitForCondition(() -> errorRef.get() != null, 1_000));

            throw errorRef.get();
        });
    }

    private void assertThrowsTxFinishedException(Executable run) {
        assertThrowsWithCode(TransactionException.class, TX_ALREADY_FINISHED_ERR, run, "Transaction is already finished");
    }

    private void assertAsyncThrowsTxFinishedException(Supplier<CompletableFuture<?>> run) {
        assertThrowsTxFinishedException(() -> {
            try {
                run.get().join();
            } catch (CompletionException e) {
                throw e.getCause();
            } catch (Exception e) {
                throw e;
            }
        });
    }

    @Test
    public void testImplicit() {
        accounts.recordView().upsert(null, makeValue(1, BALANCE_1));
        assertEquals(BALANCE_1, accounts.recordView().get(null, makeKey(1)).doubleValue("balance"));
    }

    @Test
    public void testWriteIntentResolutionFallbackToCommitPartitionPath() {
        accounts.recordView().upsert(null, makeValue(1, 100.));

        // Pending tx
        Transaction tx = igniteTransactions.begin();
        accounts.recordView().delete(tx, makeKey(1));

        // Imitate the restart of the client node, which is a tx coordinator, in order to make its volatile state unavailable.
        // Now coordinator path of the write intent resolution has no effect, and we should fallback to commit partition path.
        UUID txId = ((ReadWriteTransactionImpl) tx).id();

        for (TxManager txManager : txManagers()) {
            txManager.updateTxMeta(txId, old -> old == null ? null : new TxStateMeta(
                    old.txState(),
                    new UUID(1, 2),
                    old.commitPartitionId(),
                    old.commitTimestamp(),
                    old == null ? null : old.tx(),
                    old == null ? null : old.isFinishedDueToTimeout()
            ));
        }

        // Read-only.
        Transaction readOnlyTx = igniteTransactions.begin(new TransactionOptions().readOnly(true));
        assertEquals(100., accounts.recordView().get(readOnlyTx, makeKey(1)).doubleValue("balance"));
    }

    @Test
    public void testSingleGet() {
        var accountRecordsView = accounts.recordView();

        accountRecordsView.upsert(null, makeValue(1, 100.));

        Transaction tx1 = igniteTransactions.begin();
        Transaction tx2 = igniteTransactions.begin();

        accountRecordsView.upsert(tx1, makeValue(1, 200.));

        assertThrows(TransactionException.class, () -> accountRecordsView.get(tx2, makeKey(1)));

        assertEquals(100., accountRecordsView.get(null, makeKey(1)).doubleValue("balance"));

        tx1.commit();

        assertEquals(200., accountRecordsView.get(null, makeKey(1)).doubleValue("balance"));
    }

    @Test
    public void testBatchSinglePartitionGet() {
        var accountRecordsView = accounts.recordView();

        SchemaRegistry schemaRegistry = accounts.schemaView();
        var marshaller = new TupleMarshallerImpl(schemaRegistry.lastKnownSchema());

        int partId = accounts.internalTable().partitionId(marshaller.marshalKey(makeKey(0)));

        ArrayList<Integer> keys = new ArrayList<>(10);
        keys.add(0);

        for (int i = 1; i < 10_000 && keys.size() < 10; i++) {
            var p = accounts.internalTable().partitionId(marshaller.marshalKey(makeKey(i)));

            if (p == partId) {
                keys.add(i);
            }
        }

        log.info("A batch of keys for a single partition is found [partId={}, keys{}]", partId, keys);

        accountRecordsView.upsertAll(null, keys.stream().map(k -> makeValue(k, 100.)).collect(toList()));

        Transaction tx1 = igniteTransactions.begin();
        Transaction tx2 = igniteTransactions.begin();

        accountRecordsView.upsertAll(tx1, keys.stream().map(k -> makeValue(k, 200.)).collect(toList()));

        assertThrows(TransactionException.class,
                () -> accountRecordsView.getAll(tx2, keys.stream().map(k -> makeKey(k)).collect(toList())));

        for (Tuple tuple : accountRecordsView.getAll(null, keys.stream().map(k -> makeKey(k)).collect(toList()))) {
            assertEquals(100., tuple.doubleValue("balance"));
        }

        tx1.commit();

        for (Tuple tuple : accountRecordsView.getAll(null, keys.stream().map(k -> makeKey(k)).collect(toList()))) {
            assertEquals(200., tuple.doubleValue("balance"));
        }
    }

    @Test
    public void testYoungerTransactionWithHigherPriorityWaitsForOlderTransactionCommit() {
        IgniteTransactionsImpl igniteTransactionsImpl = (IgniteTransactionsImpl) igniteTransactions;

        KeyValueView<Long, String> keyValueView = customers.keyValueView(Long.class, String.class);

        // Init data.
        keyValueView.put(null, 1L, "init");

        // Start low priority transaction.
        Transaction oldLowTx = igniteTransactionsImpl.beginWithPriority(false, TxPriority.LOW);

        // Update data.
        keyValueView.put(oldLowTx, 1L, "low");

        // Start normal priority transaction.
        Transaction youngNormalTx = igniteTransactionsImpl.beginWithPriority(false, TxPriority.NORMAL);

        // Try to update the same key with normal priority.
        CompletableFuture<String> objectCompletableFuture = CompletableFuture.supplyAsync(
                () -> keyValueView.getAndPut(youngNormalTx, 1L, "normal")
        );

        // Commit low priority transaction.
        oldLowTx.commit();

        // Wait for normal priority transaction to update the key.
        assertThat(objectCompletableFuture, willBe("low"));

        // Commit normal priority transaction.
        youngNormalTx.commit();

        // Check that normal priority transaction has updated the key.
        assertEquals("normal", keyValueView.get(null, 1L));
    }

    @ParameterizedTest
    @EnumSource(TxPriority.class)
    public void testYoungerTransactionThrowsExceptionIfKeyLockedByOlderTransactionWithSamePriority(TxPriority priority) {
        IgniteTransactionsImpl igniteTransactionsImpl = (IgniteTransactionsImpl) igniteTransactions;

        KeyValueView<Long, String> keyValueView = customers.keyValueView(Long.class, String.class);

        // Init data.
        keyValueView.put(null, 1L, "init");

        // Start the first transaction.
        Transaction oldNormalTx = igniteTransactionsImpl.beginWithPriority(false, priority);

        // Update data with the first transaction.
        keyValueView.put(oldNormalTx, 1L, "low");

        // Start the second transaction with the same priority.
        Transaction youngNormalTx = igniteTransactionsImpl.beginWithPriority(false, priority);

        // Try to update the same key with the second normal priority transaction.
        assertThrows(TransactionException.class, () -> keyValueView.put(youngNormalTx, 1L, "normal"));
    }

    @Test
    public void testIgniteTransactionsAndReadTimestamp() {
        Transaction readWriteTx = igniteTransactions.begin();
        assertFalse(readWriteTx.isReadOnly());
        assertNull(((InternalTransaction) readWriteTx).readTimestamp());

        Transaction readOnlyTx = igniteTransactions.begin(new TransactionOptions().readOnly(true));
        assertTrue(readOnlyTx.isReadOnly());
        assertNotNull(((InternalTransaction) readOnlyTx).readTimestamp());

        readWriteTx.commit();

        Transaction readOnlyTx2 = igniteTransactions.begin(new TransactionOptions().readOnly(true));
        readOnlyTx2.rollback();
    }

    /**
     * Checks operations that act after a transaction is committed, are finished with exception.
     *
     * @param commit True when transaction is committed, false the transaction is rolled back.
     * @param checkLocks Whether to check locks after.
     * @param finisher Finishing closure.
     */
    protected void testTransactionAlreadyFinished(
            boolean commit,
            boolean checkLocks,
            BiConsumer<Transaction, UUID> finisher
    ) {
        Transaction tx = igniteTransactions.begin();

        var txId = ((ReadWriteTransactionImpl) tx).id();

        log.info("Started transaction {}", txId);

        var accountsRv = accounts.recordView();

        accountsRv.upsert(tx, makeValue(1, 100.));
        accountsRv.upsert(tx, makeValue(2, 200.));

        Collection<Tuple> res = accountsRv.getAll(tx, List.of(makeKey(1), makeKey(2)));

        validateBalance(res, 100., 200.);

        finisher.accept(tx, txId);

        assertThrowsWithCode(TransactionException.class, TX_ALREADY_FINISHED_ERR,
                () -> accountsRv.get(tx, makeKey(1)), "Transaction is already finished");

        assertThrowsWithCode(TransactionException.class, TX_ALREADY_FINISHED_ERR,
                () -> accountsRv.delete(tx, makeKey(1)), "Transaction is already finished");

        assertThrowsWithCode(TransactionException.class, TX_ALREADY_FINISHED_ERR,
                () -> accountsRv.get(tx, makeKey(2)), "Transaction is already finished");

        assertThrowsWithCode(TransactionException.class, TX_ALREADY_FINISHED_ERR,
                () -> accountsRv.upsert(tx, makeValue(2, 300.)), "Transaction is already finished");

        if (checkLocks) {
            assertTrue(CollectionUtils.nullOrEmpty(txManager(accounts).lockManager().locks(txId)));
        }

        if (commit) {
            res = accountsRv.getAll(null, List.of(makeKey(1), makeKey(2)));

            validateBalance(res, 100., 200.);
        } else {
            res = accountsRv.getAll(null, List.of(makeKey(1), makeKey(2)));

            assertThat(res, contains(null, null));
        }
    }

    private static class SingleRequestSubscriber<T> implements Flow.Subscriber<T> {
        private final AtomicReference<Throwable> errorRef;

        private SingleRequestSubscriber(AtomicReference<Throwable> errorRef) {
            this.errorRef = errorRef;
        }

        @Override
        public void onSubscribe(Flow.Subscription s) {
            s.request(1);
        }

        @Override
        public void onNext(T item) {
        }

        @Override
        public void onError(Throwable throwable) {
            errorRef.set(throwable);
        }

        @Override
        public void onComplete() {
        }
    }
}
