/*
 * 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.TimeUnit.SECONDS;
import static org.apache.ignite.internal.SessionUtils.executeUpdate;
import static org.apache.ignite.internal.TestWrappers.unwrapTableImpl;
import static org.apache.ignite.internal.catalog.CatalogService.DEFAULT_STORAGE_PROFILE;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.bypassingThreadAssertions;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureExceptionMatcher.willThrow;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.apache.ignite.internal.tx.impl.ResourceVacuumManager.RESOURCE_VACUUM_INTERVAL_MILLISECONDS_PROPERTY;
import static org.apache.ignite.internal.util.ExceptionUtils.extractCodeFrom;
import static org.hamcrest.MatcherAssert.assertThat;
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.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Flow.Publisher;
import java.util.concurrent.Flow.Subscription;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Predicate;
import java.util.stream.IntStream;
import org.apache.ignite.InitParametersBuilder;
import org.apache.ignite.internal.ClusterPerTestIntegrationTest;
import org.apache.ignite.internal.app.IgniteImpl;
import org.apache.ignite.internal.lang.IgniteBiTuple;
import org.apache.ignite.internal.network.ClusterService;
import org.apache.ignite.internal.network.DefaultMessagingService;
import org.apache.ignite.internal.network.NetworkMessage;
import org.apache.ignite.internal.placementdriver.ReplicaMeta;
import org.apache.ignite.internal.placementdriver.message.PlacementDriverMessagesFactory;
import org.apache.ignite.internal.placementdriver.message.StopLeaseProlongationMessage;
import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.replicator.message.ErrorTimestampAwareReplicaResponse;
import org.apache.ignite.internal.replicator.message.TimestampAwareReplicaResponse;
import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.table.distributed.replication.request.ReadWriteSingleRowReplicaRequest;
import org.apache.ignite.internal.testframework.IgniteTestUtils;
import org.apache.ignite.internal.testframework.SystemPropertiesExtension;
import org.apache.ignite.internal.testframework.WithSystemProperty;
import org.apache.ignite.internal.testframework.flow.TestFlowUtils;
import org.apache.ignite.internal.tx.HybridTimestampTracker;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.tx.MismatchingTransactionOutcomeException;
import org.apache.ignite.internal.tx.TxMeta;
import org.apache.ignite.internal.tx.TxState;
import org.apache.ignite.internal.tx.TxStateMeta;
import org.apache.ignite.internal.tx.configuration.TransactionConfiguration;
import org.apache.ignite.internal.tx.message.FinishedTransactionsBatchMessage;
import org.apache.ignite.internal.tx.message.TxFinishReplicaRequest;
import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
import org.apache.ignite.internal.tx.message.TxStateCommitPartitionRequest;
import org.apache.ignite.internal.util.ExceptionUtils;
import org.apache.ignite.lang.ErrorGroups.Transactions;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.table.RecordView;
import org.apache.ignite.table.Table;
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.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;

/**
 * Abandoned transactions integration tests.
 */
@ExtendWith(SystemPropertiesExtension.class)
@WithSystemProperty(key = RESOURCE_VACUUM_INTERVAL_MILLISECONDS_PROPERTY, value = "500")
public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
    private static final PlacementDriverMessagesFactory PLACEMENT_DRIVER_MESSAGES_FACTORY = new PlacementDriverMessagesFactory();

    /** Table name. */
    private static final String TABLE_NAME = "test_table";

    private static final int PART_ID = 0;

    @BeforeEach
    @Override
    public void setup(TestInfo testInfo) throws Exception {
        super.setup(testInfo);

        String zoneSql = "create zone test_zone with partitions=1, replicas=3, storage_profiles='" + DEFAULT_STORAGE_PROFILE + "'";
        String sql = "create table " + TABLE_NAME + " (key int primary key, val varchar(20)) with primary_zone='TEST_ZONE'";

        cluster.doInSession(0, session -> {
            executeUpdate(zoneSql, session);
            executeUpdate(sql, session);
        });
    }

    @Override
    protected void customizeInitParameters(InitParametersBuilder builder) {
        super.customizeInitParameters(builder);

        builder.clusterConfiguration("{\n"
                + "  \"transaction\": {\n"
                + "  \"abandonedCheckTs\": 600000\n"
                + "  \"attemptsObtainLock\": 0\n"
                + "  }\n"
                + "}\n");
    }

    @Test
    public void testMultipleRecoveryRequestsIssued() throws Exception {
        TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));

        var tblReplicationGrp = new TablePartitionId(tbl.tableId(), PART_ID);

        String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);

        IgniteImpl commitPartNode = findNodeByName(leaseholder);

        log.info("Transaction commit partition is determined [node={}].", commitPartNode.name());

        IgniteImpl txCrdNode = nonPrimaryNode(leaseholder);

        log.info("Transaction coordinator is chosen [node={}].", txCrdNode.name());

        Transaction oldRwTx = node(0).transactions().begin();

        UUID orphanTxId = startTransactionAndStopNode(txCrdNode);

        CompletableFuture<UUID> recoveryTxMsgCaptureFut = new CompletableFuture<>();
        AtomicInteger msgCount = new AtomicInteger();

        commitPartNode.dropMessages((nodeName, msg) -> {
            if (msg instanceof TxRecoveryMessage) {
                var recoveryTxMsg = (TxRecoveryMessage) msg;

                recoveryTxMsgCaptureFut.complete(recoveryTxMsg.txId());

                // Drop only the first recovery to emulate a lost message.
                // Another one should be issued eventually.
                return msgCount.incrementAndGet() == 1;
            }

            return false;
        });

        runConflictingTransaction(node(0), oldRwTx);
        runConflictingTransaction(node(0), node(0).transactions().begin());

        assertThat(recoveryTxMsgCaptureFut, willCompleteSuccessfully());

        assertEquals(orphanTxId, recoveryTxMsgCaptureFut.join());
        assertEquals(1, msgCount.get());

        node(0).clusterConfiguration().getConfiguration(TransactionConfiguration.KEY).change(transactionChange ->
                transactionChange.changeAbandonedCheckTs(1));

        assertTrue(waitForCondition(() -> {
            runConflictingTransaction(node(0), node(0).transactions().begin());

            return msgCount.get() > 1;
        }, 10_000));
    }

    @Test
    public void testAbandonedTxIsAborted() throws Exception {
        TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));

        var tblReplicationGrp = new TablePartitionId(tbl.tableId(), PART_ID);

        String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);

        IgniteImpl commitPartNode = findNodeByName(leaseholder);

        log.info("Transaction commit partition is determined [node={}].", commitPartNode.name());

        IgniteImpl txCrdNode = nonPrimaryNode(leaseholder);

        log.info("Transaction coordinator is chosen [node={}].", txCrdNode.name());

        UUID orphanTxId = startTransactionAndStopNode(txCrdNode);

        CompletableFuture<UUID> recoveryTxMsgCaptureFut = new CompletableFuture<>();
        AtomicInteger msgCount = new AtomicInteger();

        commitPartNode.dropMessages((nodeName, msg) -> {
            if (msg instanceof TxRecoveryMessage) {
                var recoveryTxMsg = (TxRecoveryMessage) msg;

                recoveryTxMsgCaptureFut.complete(recoveryTxMsg.txId());

                msgCount.incrementAndGet();
            }

            return false;
        });

        runConflictingTransaction(node(0), node(0).transactions().begin());

        assertThat(recoveryTxMsgCaptureFut, willCompleteSuccessfully());

        assertEquals(orphanTxId, recoveryTxMsgCaptureFut.join());
        assertEquals(1, msgCount.get());

        assertTrue(waitForCondition(() -> txStoredState(commitPartNode, orphanTxId) == TxState.ABORTED, 10_000));
    }

    @Test
    public void testWriteIntentRecoverNoCoordinator() throws Exception {
        TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));

        var tblReplicationGrp = new TablePartitionId(tbl.tableId(), PART_ID);

        String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);

        IgniteImpl commitPartNode = findNodeByName(leaseholder);

        log.info("Transaction commit partition is determined [node={}].", commitPartNode.name());

        IgniteImpl txCrdNode = nonPrimaryNode(leaseholder);

        log.info("Transaction coordinator is chosen [node={}].", txCrdNode.name());

        UUID orphanTxId = startTransactionAndStopNode(txCrdNode);

        AtomicInteger msgCount = new AtomicInteger();

        IgniteImpl roCoordNode = node(0);

        log.info("RO Transaction coordinator is chosen [node={}].", roCoordNode.name());

        commitPartNode.dropMessages((nodeName, msg) -> {
            if (msg instanceof TxStateCommitPartitionRequest) {
                msgCount.incrementAndGet();

                assertEquals(TxState.ABANDONED, txVolatileState(commitPartNode, orphanTxId));
            }

            return false;
        });

        Transaction recoveryTxReadOnly = roCoordNode.transactions().begin(new TransactionOptions().readOnly(true));

        runReadOnlyTransaction(roCoordNode, recoveryTxReadOnly);

        assertEquals(1, msgCount.get());

        assertTrue(waitForCondition(() -> txStoredState(commitPartNode, orphanTxId) == TxState.ABORTED, 10_000));
    }

    /**
     * Coordinator is alive, no recovery expected.
     */
    @Test
    public void testWriteIntentNoRecovery() throws Exception {
        TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));

        var tblReplicationGrp = new TablePartitionId(tbl.tableId(), PART_ID);

        String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);

        IgniteImpl commitPartNode = findNodeByName(leaseholder);

        log.info("Transaction commit partition is determined [node={}].", commitPartNode.name());

        IgniteImpl txCrdNode = nonPrimaryNode(leaseholder);

        log.info("Transaction coordinator is chosen [node={}].", txCrdNode.name());

        Transaction rwTransaction = createRwTransaction(txCrdNode);

        AtomicInteger msgCount = new AtomicInteger();

        IgniteImpl roCoordNode = node(0);

        log.info("RO Transaction coordinator is chosen [node={}].", roCoordNode.name());

        commitPartNode.dropMessages((nodeName, msg) -> {
            if (msg instanceof TxRecoveryMessage) {
                msgCount.incrementAndGet();
            }

            return false;
        });

        Transaction recoveryTxReadOnly = roCoordNode.transactions().begin(new TransactionOptions().readOnly(true));

        runReadOnlyTransaction(roCoordNode, recoveryTxReadOnly);

        assertEquals(0, msgCount.get());

        rwTransaction.commit();

        UUID rwId = ((InternalTransaction) rwTransaction).id();

        assertTrue(waitForCondition(() -> txStoredState(commitPartNode, rwId) == TxState.COMMITTED, 10_000));
    }

    @Test
    public void testWriteIntentRecoveryAndLockConflict() throws Exception {
        TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));

        var tblReplicationGrp = new TablePartitionId(tbl.tableId(), PART_ID);

        String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);

        IgniteImpl commitPartNode = findNodeByName(leaseholder);

        log.info("Transaction commit partition is determined [node={}].", commitPartNode.name());

        IgniteImpl txCrdNode = nonPrimaryNode(leaseholder);

        log.info("Transaction coordinator is chosen [node={}].", txCrdNode.name());

        UUID orphanTxId = startTransactionAndStopNode(txCrdNode);

        AtomicInteger stateMsgCount = new AtomicInteger();
        AtomicInteger recoveryMsgCount = new AtomicInteger();

        IgniteImpl roCoordNode = node(0);

        log.info("RO Transaction coordinator is chosen [node={}].", roCoordNode.name());

        CompletableFuture<UUID> txMsgCaptureFut = new CompletableFuture<>();

        commitPartNode.dropMessages((nodeName, msg) -> {
            if (msg instanceof TxStateCommitPartitionRequest) {
                stateMsgCount.incrementAndGet();

                assertEquals(TxState.ABANDONED, txVolatileState(commitPartNode, orphanTxId));

                txMsgCaptureFut.complete(((TxStateCommitPartitionRequest) msg).txId());
            }

            if (msg instanceof TxRecoveryMessage) {
                recoveryMsgCount.incrementAndGet();
            }

            return false;
        });

        Transaction recoveryTxReadOnly = roCoordNode.transactions().begin(new TransactionOptions().readOnly(true));

        RecordView view = roCoordNode.tables().table(TABLE_NAME).recordView();

        view.getAsync(recoveryTxReadOnly, Tuple.create().set("key", 42));

        assertThat(txMsgCaptureFut, willCompleteSuccessfully());

        runConflictingTransaction(commitPartNode, commitPartNode.transactions().begin());

        assertEquals(1, stateMsgCount.get());

        assertEquals(0, recoveryMsgCount.get());

        assertTrue(waitForCondition(() -> txStoredState(commitPartNode, orphanTxId) == TxState.ABORTED, 10_000));
    }

    /**
     * Coordinator sends a commit message and dies. The message eventually reaches the commit partition and gets executed.
     * The expected outcome of the transaction is COMMIT.
     */
    @Test
    public void testSendCommitAndDie() throws Exception {
        TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));

        var tblReplicationGrp = new TablePartitionId(tbl.tableId(), PART_ID);

        String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);

        IgniteImpl commitPartNode = findNodeByName(leaseholder);

        log.info("Transaction commit partition is determined [node={}].", commitPartNode.name());

        IgniteImpl txCrdNode = nonPrimaryNode(leaseholder);

        log.info("Transaction coordinator is chosen [node={}].", txCrdNode.name());

        InternalTransaction orphanTx = (InternalTransaction) createRwTransaction(txCrdNode);

        CompletableFuture<TxFinishReplicaRequest> finishRequestCaptureFut = new CompletableFuture<>();
        AtomicReference<String> targetName = new AtomicReference<>();

        // Intercept the commit message, prevent it form being sent. We will kill this node anyway.
        txCrdNode.dropMessages((nodeName, msg) -> {
            if (msg instanceof TxFinishReplicaRequest) {
                var finishTxMsg = (TxFinishReplicaRequest) msg;

                finishRequestCaptureFut.complete(finishTxMsg);
                targetName.set(nodeName);

                return true;
            }

            return false;
        });

        // Initiate commit.
        orphanTx.commitAsync();

        assertThat(finishRequestCaptureFut, willCompleteSuccessfully());

        // Stop old coordinator.
        String txCrdNodeId = txCrdNode.id();

        txCrdNode.stop();

        assertTrue(waitForCondition(
                () -> node(0).clusterNodes().stream().filter(n -> txCrdNodeId.equals(n.id())).count() == 0,
                10_000)
        );

        // The state on the commit partition is still PENDING.
        assertEquals(TxState.PENDING, txVolatileState(commitPartNode, orphanTx.id()));

        // Continue the COMMIT message flow.
        CompletableFuture<NetworkMessage> finishRequest =
                bypassingThreadAssertions(() -> messaging(commitPartNode).invoke(targetName.get(), finishRequestCaptureFut.join(), 3000));

        assertThat(finishRequest, willCompleteSuccessfully());

        // The conflicting transaction should see an already committed TX.
        runRwTransactionNoError(node(0), node(0).transactions().begin());

        assertTrue(waitForCondition(() -> txStoredState(commitPartNode, orphanTx.id()) == TxState.COMMITTED, 10_000));
    }

    /**
     * Coordinator sends a commit message and dies. Another tx initiates recovery and aborts this transaction.
     * The commit message eventually reaches the commit partition and gets executed but the outcome is ABORTED.
     */
    @Test
    public void testCommitAndDieRecoveryFirst() throws Exception {
        TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));

        var tblReplicationGrp = new TablePartitionId(tbl.tableId(), PART_ID);

        String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);

        IgniteImpl commitPartNode = findNodeByName(leaseholder);

        log.info("Transaction commit partition is determined [node={}].", commitPartNode.name());

        IgniteImpl txCrdNode = nonPrimaryNode(leaseholder);

        log.info("Transaction coordinator is chosen [node={}].", txCrdNode.name());

        InternalTransaction orphanTx = (InternalTransaction) createRwTransaction(txCrdNode);

        CompletableFuture<TxFinishReplicaRequest> finishRequestCaptureFut = new CompletableFuture<>();
        AtomicReference<String> targetName = new AtomicReference<>();

        // Intercept the commit message, prevent it form being sent. We will kill this node anyway.
        txCrdNode.dropMessages((nodeName, msg) -> {
            if (msg instanceof TxFinishReplicaRequest) {
                var finishTxMsg = (TxFinishReplicaRequest) msg;

                finishRequestCaptureFut.complete(finishTxMsg);
                targetName.set(nodeName);

                return true;
            }

            return false;
        });

        // Initiate commit.
        orphanTx.commitAsync();

        assertThat(finishRequestCaptureFut, willCompleteSuccessfully());

        // Stop old coordinator.
        String txCrdNodeId = txCrdNode.id();

        txCrdNode.stop();

        assertTrue(waitForCondition(
                () -> node(0).clusterNodes().stream().filter(n -> txCrdNodeId.equals(n.id())).count() == 0,
                10_000)
        );

        // The state on the commit partition is still PENDING.
        assertEquals(TxState.PENDING, txVolatileState(commitPartNode, orphanTx.id()));

        IgniteImpl newTxCoord = node(0);

        runRwTransactionNoError(newTxCoord, newTxCoord.transactions().begin());

        assertTrue(waitForCondition(() -> txStoredState(commitPartNode, orphanTx.id()) == TxState.ABORTED, 10_000));

        CompletableFuture<NetworkMessage> commitRequest =
                bypassingThreadAssertions(() -> messaging(commitPartNode).invoke(targetName.get(), finishRequestCaptureFut.join(), 3000));

        assertThat(commitRequest, willCompleteSuccessfully());

        NetworkMessage response = commitRequest.join();

        assertInstanceOf(ErrorTimestampAwareReplicaResponse.class, response);

        ErrorTimestampAwareReplicaResponse errorResponse = (ErrorTimestampAwareReplicaResponse) response;

        assertInstanceOf(MismatchingTransactionOutcomeException.class, ExceptionUtils.unwrapCause(errorResponse.throwable()));

        assertEquals(TxState.ABORTED, txStoredState(commitPartNode, orphanTx.id()));
    }

    @Test
    public void testRecoveryIsTriggeredOnce() throws Exception {
        TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));

        var tblReplicationGrp = new TablePartitionId(tbl.tableId(), PART_ID);

        String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);

        IgniteImpl commitPartNode = findNodeByName(leaseholder);

        log.info("Transaction commit partition is determined [node={}].", commitPartNode.name());

        IgniteImpl txCrdNode = nonPrimaryNode(leaseholder);

        log.info("Transaction coordinator is chosen [node={}].", txCrdNode.name());

        UUID orphanTxId = startTransactionAndStopNode(txCrdNode);

        log.info("Orphan tx [id={}]", orphanTxId);

        CompletableFuture<UUID> recoveryTxMsgCaptureFut = new CompletableFuture<>();
        AtomicInteger msgCount = new AtomicInteger();

        commitPartNode.dropMessages((nodeName, msg) -> {
            if (msg instanceof TxRecoveryMessage) {
                var recoveryTxMsg = (TxRecoveryMessage) msg;

                recoveryTxMsgCaptureFut.complete(recoveryTxMsg.txId());

                msgCount.incrementAndGet();
            }

            return false;
        });

        IgniteImpl newCoordNode = node(0);

        log.info("New transaction coordinator is chosen [node={}].", newCoordNode.name());

        // Run RW transaction.
        Transaction rwTx1 = commitPartNode.transactions().begin();

        UUID rwTx1Id = ((InternalTransaction) rwTx1).id();

        log.info("First concurrent tx [id={}]", rwTx1Id);

        runConflictingTransaction(commitPartNode, rwTx1);

        Transaction rwTx2 = newCoordNode.transactions().begin();

        UUID rwTx2Id = ((InternalTransaction) rwTx2).id();

        log.info("Second concurrent tx [id={}]", rwTx2Id);

        runRwTransactionNoError(newCoordNode, rwTx2);

        assertThat(recoveryTxMsgCaptureFut, willCompleteSuccessfully());

        assertEquals(orphanTxId, recoveryTxMsgCaptureFut.join());
        assertEquals(1, msgCount.get());

        rwTx2.commit();

        assertTrue(waitForCondition(() -> txStoredState(commitPartNode, orphanTxId) == TxState.ABORTED, 10_000));
        assertTrue(waitForCondition(() -> txStoredState(commitPartNode, rwTx1Id) == TxState.ABORTED, 10_000));

        Transaction rwTx3 = newCoordNode.transactions().begin();

        log.info("Start RW tx {}", ((InternalTransaction) rwTx3).id());

        runRwTransactionNoError(newCoordNode, rwTx3);

        rwTx3.commit();

        assertTrue(waitForCondition(() -> txStoredState(commitPartNode, ((InternalTransaction) rwTx3).id()) == TxState.COMMITTED, 10_000));
    }

    @Test
    public void testFinishAlreadyFinishedTx() throws Exception {
        TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));

        var tblReplicationGrp = new TablePartitionId(tbl.tableId(), PART_ID);

        String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);

        IgniteImpl commitPartNode = findNodeByName(leaseholder);

        log.info("Transaction commit partition is determined [node={}].", commitPartNode.name());

        IgniteImpl txCrdNode = nonPrimaryNode(leaseholder);

        log.info("Transaction coordinator is chosen [node={}].", txCrdNode.name());

        Transaction rwTx1 = createRwTransaction(txCrdNode);

        rwTx1.commit();

        UUID rwTx1Id = ((InternalTransaction) rwTx1).id();

        assertTrue(waitForCondition(() -> txStoredState(commitPartNode, rwTx1Id) == TxState.COMMITTED, 10_000));

        IgniteImpl txCrdNode2 = node(0);

        CompletableFuture<Void> finish2 = txCrdNode2.txManager().finish(
                new HybridTimestampTracker(),
                ((InternalTransaction) rwTx1).commitPartition(),
                false,
                Map.of(((InternalTransaction) rwTx1).commitPartition(), new IgniteBiTuple<>(txCrdNode2.node(), 0L)),
                rwTx1Id
        );

        assertThat(finish2, willThrow(MismatchingTransactionOutcomeException.class));
    }

    @Test
    public void testPrimaryFailureRightAfterCommitMsg() throws Exception {
        TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));

        var tblReplicationGrp = new TablePartitionId(tbl.tableId(), PART_ID);

        String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);

        IgniteImpl commitPartNode = findNodeByName(leaseholder);

        log.info("Transaction commit partition is determined [node={}].", commitPartNode.name());

        IgniteImpl txCrdNode = nonPrimaryNode(leaseholder);

        log.info("Transaction coordinator node is determined [node={}].", txCrdNode.name());

        Transaction rwTx1 = createRwTransaction(txCrdNode);

        CompletableFuture<?> commitMsgSentFut = new CompletableFuture<>();
        CompletableFuture<?> cancelLeaseFuture = new CompletableFuture<>();

        txCrdNode.dropMessages((nodeName, msg) -> {
            if (msg instanceof TxFinishReplicaRequest) {
                boolean isFirst = !commitMsgSentFut.isDone();

                if (isFirst) {
                    commitMsgSentFut.complete(null);

                    return true;
                } else {
                    cancelLeaseFuture.join();

                    return false;
                }
            }

            return false;
        });

        CompletableFuture<Void> commitFut = rwTx1.commitAsync();

        assertThat(commitMsgSentFut, willCompleteSuccessfully());

        cancelLease(commitPartNode, tblReplicationGrp);

        waitAndGetLeaseholder(txCrdNode, tblReplicationGrp);

        cancelLeaseFuture.complete(null);

        assertThat(commitFut, willCompleteSuccessfully());

        RecordView<Tuple> view = txCrdNode.tables().table(TABLE_NAME).recordView();

        var rec = view.get(null, Tuple.create().set("key", 42));

        assertNotNull(rec);
        assertEquals((Integer) 42, rec.value("key"));
        assertEquals("val1", rec.value("val"));
    }

    @Test
    public void testPrimaryFailureWhileInflightInProgress() throws Exception {
        TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));

        var tblReplicationGrp = new TablePartitionId(tbl.tableId(), PART_ID);

        String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);

        IgniteImpl commitPartNode = findNodeByName(leaseholder);

        log.info("Transaction commit partition is determined [node={}].", commitPartNode.name());

        IgniteImpl txCrdNode = nonPrimaryNode(leaseholder);

        log.info("Transaction coordinator node is determined [node={}].", txCrdNode.name());

        Transaction rwTx1 = createRwTransaction(txCrdNode);

        txCrdNode.dropMessages((nodeName, msg) -> {
            if (msg instanceof ReadWriteSingleRowReplicaRequest) {
                return true;
            }

            return false;
        });

        assertThrows(TransactionException.class, () -> {
            RecordView<Tuple> view = txCrdNode.tables().table(TABLE_NAME).recordView();
            view.upsert(rwTx1, Tuple.create().set("key", 1).set("val", "val1"));
        });

        CompletableFuture<Void> commitFut = rwTx1.commitAsync();

        commitPartNode.stop();

        assertThat(commitFut, willCompleteSuccessfully());
    }

    @Test
    public void testPrimaryFailureWhileInflightInProgressAfterFirstResponse() throws Exception {
        TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));

        var tblReplicationGrp = new TablePartitionId(tbl.tableId(), PART_ID);

        String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);

        IgniteImpl commitPartNode = findNodeByName(leaseholder);

        log.info("Transaction commit partition is determined [node={}].", commitPartNode.name());

        IgniteImpl txCrdNode = nonPrimaryNode(leaseholder);

        log.info("Transaction coordinator node is determined [node={}].", txCrdNode.name());

        CompletableFuture<?> firstResponseSent = new CompletableFuture<>();

        commitPartNode.dropMessages((nodeName, msg) -> {
            if (msg instanceof TimestampAwareReplicaResponse) {
                TimestampAwareReplicaResponse response = (TimestampAwareReplicaResponse) msg;

                if (response.result() == null) {
                    firstResponseSent.complete(null);
                }

                // This means this is the second response that finishes an in-flight future.
                if (response.result() instanceof UUID) {
                    return true;
                }
            }

            return false;
        });

        Transaction rwTx1 = createRwTransaction(txCrdNode);

        CompletableFuture<Void> commitFut = rwTx1.commitAsync();

        assertThat(firstResponseSent, willCompleteSuccessfully());

        cancelLease(commitPartNode, tblReplicationGrp);

        assertThat(commitFut, willThrow(MismatchingTransactionOutcomeException.class, 30, SECONDS));

        RecordView<Tuple> view = txCrdNode.tables().table(TABLE_NAME).recordView();

        var rec = view.get(null, Tuple.create().set("key", 42));

        assertNull(rec);
    }

    @Test
    public void testTsRecoveryForCursor() throws Exception {
        TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));

        preloadData(tbl, 10);

        var tblReplicationGrp = new TablePartitionId(tbl.tableId(), PART_ID);

        String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);

        IgniteImpl commitPartNode = findNodeByName(leaseholder);

        log.info("Transaction commit partition is determined [node={}].", commitPartNode.name());

        IgniteImpl txCrdNode = nonPrimaryNode(leaseholder);

        log.info("Transaction coordinator is chosen [node={}].", txCrdNode.name());

        startTransactionWithCursorAndStopNode(txCrdNode, commitPartNode);

        IgniteImpl newCoordNode = node(0);

        log.info("New transaction coordinator is chosen [node={}].", newCoordNode.name());

        CompletableFuture<Void> txMsgCaptureFut = new CompletableFuture<>();

        commitPartNode.dropMessages((nodeName, msg) -> {
            if (msg instanceof TxRecoveryMessage) {
                txMsgCaptureFut.complete(null);
            }

            return false;
        });

        Transaction tx = newCoordNode.transactions().begin();

        RecordView view = newCoordNode.tables().table(TABLE_NAME).recordView();

        var opFut = view.upsertAsync(tx, Tuple.create().set("key", 42).set("val", "new"));

        try {
            opFut.get();
        } catch (Exception ex) {
            log.info("Expected conflict that have to start recovery: " + ex.getMessage());
        }

        assertThat(txMsgCaptureFut, willCompleteSuccessfully());
    }

    /**
     * Starts read-write/read only transaction, creates a cursor and checks that the cursor is canceled after tx coordinator leaves.
     *
     * @param readOnly Whether the tx is read only.
     */
    @ParameterizedTest
    @ValueSource(booleans = {true, false})
    public void testCursorCleanup(boolean readOnly) throws Exception {
        TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));

        var tblReplicationGrp = new TablePartitionId(tbl.tableId(), PART_ID);

        String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);

        IgniteImpl targetNode = findNodeByName(leaseholder);

        log.info("Transaction target node is determined [node={}].", targetNode.name());

        IgniteImpl txCrdNode = nonPrimaryNode(leaseholder);

        log.info("Transaction coordinator is chosen [node={}].", txCrdNode.name());

        IgniteImpl thirdNode = findNode(0, initialNodes(), node -> node != txCrdNode && node != targetNode);

        log.info("Another tx coordinator is chosen [node={}].", thirdNode.name());

        // Preload data from coordinator node to adjust the observable timestamp.
        preloadData(txCrdNode.tables().table(TABLE_NAME), 10);

        // Creating a cursor that should remain because it is created from the node that will remain in the cluster.
        InternalTransaction rwTx = (InternalTransaction) thirdNode.transactions().begin();
        scanSingleEntryAndLeaveCursorOpen(targetNode, unwrapTableImpl(thirdNode.tables().table(TABLE_NAME)), rwTx);

        // Creating a cursor that should be closed because tx coordinator leaves topology.
        InternalTransaction tx = (InternalTransaction) txCrdNode.transactions().begin(new TransactionOptions().readOnly(readOnly));
        startTransactionWithCursorAndStopNode(txCrdNode, targetNode, tx);

        // Checking that just one cursor is remaining.
        assertTrue(waitForCondition(() -> targetNode.resourcesRegistry().resources().size() == 1, 3000));
    }

    private static void preloadData(Table table, int entries) {
        RecordView<Tuple> view = table.recordView();

        for (int i = 0; i < entries; i++) {
            view.upsert(null, Tuple.create().set("key", i).set("val", "preload"));
        }
    }

    private UUID startTransactionWithCursorAndStopNode(IgniteImpl txCrdNode, IgniteImpl targetNode) throws Exception {
        InternalTransaction rwTx = (InternalTransaction) txCrdNode.transactions().begin();

        startTransactionWithCursorAndStopNode(txCrdNode, targetNode, rwTx);

        return rwTx.id();
    }

    /**
     * Starts a scan procedure and leaves it incomplete, then stops the coordinator node.
     *
     * @param txCrdNode Tx coordinator node.
     * @param targetNode Node where the cursor should be created.
     * @param tx Transaction.
     */
    private void startTransactionWithCursorAndStopNode(IgniteImpl txCrdNode, IgniteImpl targetNode, InternalTransaction tx)
            throws Exception {
        scanSingleEntryAndLeaveCursorOpen(targetNode, unwrapTableImpl(txCrdNode.tables().table(TABLE_NAME)), tx);

        String txCrdNodeId = txCrdNode.id();

        txCrdNode.stop();

        assertTrue(waitForCondition(
                () -> node(0).clusterNodes().stream().filter(n -> txCrdNodeId.equals(n.id())).count() == 0,
                10_000)
        );
    }

    /**
     * Starts a scan procedure for a specific transaction and reads only the first line from the cursor.
     *
     * @param targetNode Node where the cursor should be created.
     * @param tbl Scanned table.
     * @param tx Transaction.
     * @throws Exception If failed.
     */
    private void scanSingleEntryAndLeaveCursorOpen(IgniteImpl targetNode, TableViewInternal tbl, InternalTransaction tx)
            throws Exception {
        int initialCursorsCount = targetNode.resourcesRegistry().resources().size();

        Publisher<BinaryRow> publisher;
        if (tx.isReadOnly()) {
            String primary = waitAndGetLeaseholder(node(0), new TablePartitionId(tbl.tableId(), PART_ID));

            ClusterNode primaryNode = node(0).clusterNodes().stream().filter(node -> node.name().equals(primary)).findAny().get();

            publisher = tbl.internalTable().scan(PART_ID, tx.id(), tx.readTimestamp(), primaryNode, tx.coordinatorId());
        } else {
            publisher = tbl.internalTable().scan(PART_ID, tx);
        }

        List<BinaryRow> scannedRows = new ArrayList<>();
        CompletableFuture<Void> scanned = new CompletableFuture<>();

        Subscription subscription = TestFlowUtils.subscribeToPublisher(scannedRows, publisher, scanned);

        subscription.request(1);

        assertTrue(waitForCondition(() -> scannedRows.size() == 1, 10_000));

        assertFalse(scanned.isDone());

        assertEquals(initialCursorsCount + 1, targetNode.resourcesRegistry().resources().size());
    }

    @Test
    public void testCursorsClosedAfterTxClose() throws Exception {
        TableImpl tbl = unwrapTableImpl(node(0).tables().table(TABLE_NAME));

        var tblReplicationGrp = new TablePartitionId(tbl.tableId(), PART_ID);

        String leaseholder = waitAndGetPrimaryReplica(node(0), tblReplicationGrp).getLeaseholder();

        IgniteImpl txExecNode = findNodeByName(leaseholder);

        log.info("Transaction will be executed on [node={}].", txExecNode.name());

        IgniteImpl txCrdNode = findNode(1, initialNodes(), n -> !leaseholder.equals(n.name()));

        log.info("Transaction coordinator is [node={}].", txCrdNode.name());

        // Ensure there are no open cursors.
        assertEquals(0, txExecNode.resourcesRegistry().resources().size());

        preloadData(txCrdNode.tables().table(TABLE_NAME), 10);

        CompletableFuture<Void> txRequestCaptureFut = new CompletableFuture<>();

        InternalTransaction roTx = (InternalTransaction) txCrdNode.transactions().begin(new TransactionOptions().readOnly(true));

        log.info("Run scan in RO [txId={}].", roTx.id());

        txCrdNode.dropMessages((nodeName, msg) -> {
            if (msg instanceof FinishedTransactionsBatchMessage) {
                Collection<UUID> transactions = ((FinishedTransactionsBatchMessage) msg).transactions();

                if (transactions.contains(roTx.id())) {
                    // Caught the request containing the finished tx.
                    txRequestCaptureFut.complete(null);
                } else {
                    log.info("Received FinishedTransactionsBatchMessage without tx [txId={}]", roTx.id());
                }
            }

            return false;
        });

        scanSingleEntryAndLeaveCursorOpen(txExecNode, unwrapTableImpl(txCrdNode.tables().table(TABLE_NAME)), roTx);

        // After the RO scan there should be one open cursor.
        assertEquals(1, txExecNode.resourcesRegistry().resources().size());

        roTx.commit();

        // Wait for the cursor cleanup message to arrive.
        assertThat(txRequestCaptureFut, willCompleteSuccessfully());

        // Now check that the cursor is closed.
        assertTrue(waitForCondition(
                () -> txExecNode.resourcesRegistry().resources().isEmpty(),
                10_000)
        );
    }

    private DefaultMessagingService messaging(IgniteImpl node) {
        ClusterService coordinatorService = IgniteTestUtils.getFieldValue(node, IgniteImpl.class, "clusterSvc");

        return (DefaultMessagingService) coordinatorService.messagingService();
    }

    private @Nullable TxState txVolatileState(IgniteImpl node, UUID txId) {
        TxStateMeta txMeta = node.txManager().stateMeta(txId);

        return txMeta == null ? null : txMeta.txState();
    }

    private static @Nullable TxState txStoredState(IgniteImpl node, UUID txId) {
        TxMeta txMeta = txStoredMeta(node, txId);

        return txMeta == null ? null : txMeta.txState();
    }

    private static @Nullable TxMeta txStoredMeta(IgniteImpl node, UUID txId) {
        InternalTable internalTable = unwrapTableImpl(node.tables().table(TABLE_NAME)).internalTable();

        return bypassingThreadAssertions(() -> internalTable.txStateStorage().getTxStateStorage(0).get(txId));
    }

    /**
     * Runs a transaction that was expectedly finished with the lock conflict exception.
     *
     * @param node Transaction coordinator node.
     * @param rwTx A transaction to create a lock conflict with an abandoned one.
     */
    private void runConflictingTransaction(IgniteImpl node, Transaction rwTx) {
        RecordView view = node.tables().table(TABLE_NAME).recordView();

        try {
            view.upsert(rwTx, Tuple.create().set("key", 42).set("val", "val2"));

            fail("Lock conflict have to be detected.");
        } catch (Exception e) {
            assertEquals(Transactions.ACQUIRE_LOCK_ERR, extractCodeFrom(e));

            log.info("Expected lock conflict.", e);
        }
    }

    private void runRwTransactionNoError(IgniteImpl node, Transaction rwTx) {
        RecordView view = node.tables().table(TABLE_NAME).recordView();

        try {
            view.upsert(rwTx, Tuple.create().set("key", 42).set("val", "val2"));
        } catch (Exception e) {
            assertEquals(Transactions.ACQUIRE_LOCK_ERR, extractCodeFrom(e));

            log.info("Expected lock conflict.", e);
        }
    }

    /**
     * Runs a RO transaction to trigger recovery on write intent resolutioin.
     *
     * @param node Transaction coordinator node.
     * @param roTx A transaction to resolve write intents from the abandoned TX.
     */
    private void runReadOnlyTransaction(IgniteImpl node, Transaction roTx) {
        RecordView view = node.tables().table(TABLE_NAME).recordView();

        try {
            view.get(roTx, Tuple.create().set("key", 42));
        } catch (Exception e) {
            assertEquals(Transactions.ACQUIRE_LOCK_ERR, extractCodeFrom(e));

            log.info("Expected lock conflict.", e);
        }
    }

    /**
     * Starts the transaction, takes a lock, and stops the transaction coordinator. The stopped node leaves the transaction in the pending
     * state.
     *
     * @param node Transaction coordinator node.
     * @return Transaction id.
     * @throws InterruptedException If interrupted.
     */
    private UUID startTransactionAndStopNode(IgniteImpl node) throws InterruptedException {
        Transaction rwTx1 = createRwTransaction(node);

        String txCrdNodeId = node.id();

        node.stop();

        assertTrue(waitForCondition(
                () -> node(0).clusterNodes().stream().filter(n -> txCrdNodeId.equals(n.id())).count() == 0,
                10_000)
        );
        return ((InternalTransaction) rwTx1).id();
    }

    /**
     * Creates RW the transaction.
     *
     * @param node Transaction coordinator node.
     * @return Transaction id.
     */
    private Transaction createRwTransaction(IgniteImpl node) {
        RecordView<Tuple> view = node.tables().table(TABLE_NAME).recordView();

        Transaction rwTx1 = node.transactions().begin();

        view.upsert(rwTx1, Tuple.create().set("key", 42).set("val", "val1"));

        return rwTx1;
    }

    private IgniteImpl findNode(int startRange, int endRange, Predicate<IgniteImpl> filter) {
        return IntStream.range(startRange, endRange)
                .mapToObj(this::node)
                .filter(filter::test)
                .findFirst()
                .get();
    }

    private IgniteImpl findNodeByName(String leaseholder) {
        return findNode(0, initialNodes(), n -> leaseholder.equals(n.name()));
    }

    private IgniteImpl nonPrimaryNode(String leaseholder) {
        return findNode(1, initialNodes(), n -> !leaseholder.equals(n.name()));
    }

    private static ReplicaMeta waitAndGetPrimaryReplica(IgniteImpl node, ReplicationGroupId tblReplicationGrp) {
        CompletableFuture<ReplicaMeta> primaryReplicaFut = node.placementDriver().awaitPrimaryReplica(
                tblReplicationGrp,
                node.clock().now(),
                10,
                SECONDS
        );

        assertThat(primaryReplicaFut, willCompleteSuccessfully());

        return primaryReplicaFut.join();
    }

    private static String waitAndGetLeaseholder(IgniteImpl node, ReplicationGroupId tblReplicationGrp) {
        return waitAndGetPrimaryReplica(node, tblReplicationGrp).getLeaseholder();
    }

    private void cancelLease(IgniteImpl leaseholder, ReplicationGroupId groupId) {
        StopLeaseProlongationMessage msg = PLACEMENT_DRIVER_MESSAGES_FACTORY
                .stopLeaseProlongationMessage()
                .groupId(groupId)
                .build();

        // Just sent it to all nodes to not determine the exact placement driver active actor.
        runningNodes().forEach(node -> leaseholder.sendFakeMessage(node.name(), msg));
    }
}