/*
 * 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.tephra.hbase.txprune;

import com.google.common.collect.ImmutableSet;
import com.google.common.collect.ImmutableSortedMap;
import com.google.common.collect.ImmutableSortedSet;

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.Store;
import org.apache.hadoop.hbase.regionserver.StoreFile;
import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequest;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.tephra.TransactionContext;
import org.apache.tephra.TransactionManager;
import org.apache.tephra.TxConstants;
import org.apache.tephra.coprocessor.CacheSupplier;
import org.apache.tephra.coprocessor.TransactionStateCache;
import org.apache.tephra.hbase.AbstractHBaseTableTest;
import org.apache.tephra.hbase.TransactionAwareHTable;
import org.apache.tephra.hbase.coprocessor.TransactionProcessor;
import org.apache.tephra.inmemory.InMemoryTxSystemClient;
import org.apache.tephra.metrics.TxMetricsCollector;
import org.apache.tephra.persist.InMemoryTransactionStateStorage;
import org.apache.tephra.persist.TransactionSnapshot;
import org.apache.tephra.persist.TransactionStateStorage;
import org.apache.tephra.persist.TransactionVisibilityState;
import org.apache.tephra.txprune.TransactionPruningPlugin;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;

import java.io.IOException;
import java.util.Collections;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Test invalid list pruning
 */
public class InvalidListPruneTest extends AbstractHBaseTableTest {
  private static final byte[] family = Bytes.toBytes("f1");
  private static final byte[] qualifier = Bytes.toBytes("col1");
  private static final int MAX_ROWS = 1000;

  private static TableName txDataTable1;
  private static TableName pruneStateTable;
  private static DataJanitorState dataJanitorState;

  // Override AbstractHBaseTableTest.startMiniCluster to setup configuration
  @BeforeClass
  public static void startMiniCluster() throws Exception {
    // Setup the configuration to start HBase cluster with the invalid list pruning enabled
    conf = HBaseConfiguration.create();
    conf.setBoolean(TxConstants.TransactionPruning.PRUNE_ENABLE, true);
    // Flush prune data to table quickly, so that tests don't need have to wait long to see updates
    conf.setLong(TxConstants.TransactionPruning.PRUNE_FLUSH_INTERVAL, 0L);
    AbstractHBaseTableTest.startMiniCluster();

    TransactionStateStorage txStateStorage = new InMemoryTransactionStateStorage();
    TransactionManager txManager = new TransactionManager(conf, txStateStorage, new TxMetricsCollector());
    txManager.startAndWait();

    // Do some transactional data operations
    txDataTable1 = TableName.valueOf("invalidListPruneTestTable1");
    Table hTable = createTable(txDataTable1.getName(), new byte[][]{family}, false,
                                Collections.singletonList(TestTransactionProcessor.class.getName()));
    try (TransactionAwareHTable txTable = new TransactionAwareHTable(hTable, TxConstants.ConflictDetection.ROW)) {
      TransactionContext txContext = new TransactionContext(new InMemoryTxSystemClient(txManager), txTable);
      txContext.start();
      for (int i = 0; i < MAX_ROWS; ++i) {
        txTable.put(new Put(Bytes.toBytes(i)).addColumn(family, qualifier, Bytes.toBytes(i)));
      }
      txContext.finish();
    }

    testUtil.flush(txDataTable1);
    txManager.stopAndWait();

    pruneStateTable = TableName.valueOf(conf.get(TxConstants.TransactionPruning.PRUNE_STATE_TABLE,
                                                 TxConstants.TransactionPruning.DEFAULT_PRUNE_STATE_TABLE));
    dataJanitorState =
      new DataJanitorState(new DataJanitorState.TableSupplier() {
        @Override
        public Table get() throws IOException {
          return testUtil.getConnection().getTable(pruneStateTable);
        }
      });

  }

  @AfterClass
  public static void shutdownAfterClass() throws Exception {
    hBaseAdmin.disableTable(txDataTable1);
    hBaseAdmin.deleteTable(txDataTable1);
  }

  @Before
  public void beforeTest() throws Exception {
    createPruneStateTable();
    InMemoryTransactionStateCache.setTransactionSnapshot(null);
  }

  private void createPruneStateTable() throws Exception {
    Table table = createTable(pruneStateTable.getName(), new byte[][]{DataJanitorState.FAMILY}, false,
                               // Prune state table is a non-transactional table, hence no transaction co-processor
                               Collections.<String>emptyList());
    table.close();
  }

  @After
  public void afterTest() throws Exception {
    // Disable the data table so that prune writer thread gets stopped,
    // this makes sure that any cached value will not interfere with next test
    hBaseAdmin.disableTable(txDataTable1);
    deletePruneStateTable();
    // Enabling the table enables the prune writer thread again
    hBaseAdmin.enableTable(txDataTable1);
  }

  private void deletePruneStateTable() throws Exception {
    if (hBaseAdmin.tableExists(pruneStateTable)) {
      hBaseAdmin.disableTable(pruneStateTable);
      hBaseAdmin.deleteTable(pruneStateTable);
    }
  }

  @Test
  public void testRecordCompactionState() throws Exception {
    // No prune upper bound initially
    Assert.assertEquals(-1,
                        dataJanitorState.getPruneUpperBoundForRegion(getRegionName(txDataTable1, Bytes.toBytes(0))));

    // Create a new transaction snapshot
    InMemoryTransactionStateCache.setTransactionSnapshot(
      new TransactionSnapshot(100, 100, 100, ImmutableSet.of(50L),
                              ImmutableSortedMap.<Long, TransactionManager.InProgressTx>of()));
    // Run minor compaction
    testUtil.compact(txDataTable1, false);
    // Since the write to prune table happens async, we need to sleep a bit before checking the state of the table
    TimeUnit.SECONDS.sleep(2);
    // No prune upper bound after minor compaction too
    Assert.assertEquals(-1,
                        dataJanitorState.getPruneUpperBoundForRegion(getRegionName(txDataTable1, Bytes.toBytes(0))));

    // Run major compaction, and verify prune upper bound
    testUtil.compact(txDataTable1, true);
    // Since the write to prune table happens async, we need to sleep a bit before checking the state of the table
    TimeUnit.SECONDS.sleep(2);
    Assert.assertEquals(50,
                        dataJanitorState.getPruneUpperBoundForRegion(getRegionName(txDataTable1, Bytes.toBytes(0))));

    // Run major compaction again with same snapshot, prune upper bound should not change
    testUtil.compact(txDataTable1, true);
    // Since the write to prune table happens async, we need to sleep a bit before checking the state of the table
    TimeUnit.SECONDS.sleep(2);
    Assert.assertEquals(50,
                        dataJanitorState.getPruneUpperBoundForRegion(getRegionName(txDataTable1, Bytes.toBytes(0))));

    // Create a new transaction snapshot
    InMemoryTransactionStateCache.setTransactionSnapshot(
      new TransactionSnapshot(110, 111, 112, ImmutableSet.of(150L),
                              ImmutableSortedMap.of(105L, new TransactionManager.InProgressTx(
                                100, 30, TransactionManager.InProgressType.SHORT))));
    Assert.assertEquals(50,
                        dataJanitorState.getPruneUpperBoundForRegion(getRegionName(txDataTable1, Bytes.toBytes(0))));

    // Run major compaction again, now prune upper bound should change
    testUtil.compact(txDataTable1, true);
    // Since the write to prune table happens async, we need to sleep a bit before checking the state of the table
    TimeUnit.SECONDS.sleep(2);
    Assert.assertEquals(104,
                        dataJanitorState.getPruneUpperBoundForRegion(getRegionName(txDataTable1, Bytes.toBytes(0))));
  }

  @Test
  public void testRecordCompactionStateNoTable() throws Exception {
    // Create a new transaction snapshot
    InMemoryTransactionStateCache.setTransactionSnapshot(
      new TransactionSnapshot(100, 100, 100, ImmutableSet.of(50L),
                              ImmutableSortedMap.<Long, TransactionManager.InProgressTx>of()));
    // Run major compaction, and verify it completes
    long now = System.currentTimeMillis();
    testUtil.compact(txDataTable1, true);
    // Since the write to prune table happens async, we need to sleep a bit before checking the state of the table
    TimeUnit.SECONDS.sleep(2);
    long lastMajorCompactionTime = TestTransactionProcessor.lastMajorCompactionTime.get();
    Assert.assertTrue(String.format("Expected %d, but was %d", now, lastMajorCompactionTime),
                      lastMajorCompactionTime >= now);
  }

  @Test
  public void testRecordCompactionStateNoTxSnapshot() throws Exception {
    // Test recording state without having a transaction snapshot to make sure we don't disrupt
    // major compaction in that case
    InMemoryTransactionStateCache.setTransactionSnapshot(null);
    // Run major compaction, and verify it completes
    long now = System.currentTimeMillis();
    testUtil.compact(txDataTable1, true);
    // Since the write to prune table happens async, we need to sleep a bit before checking the state of the table
    TimeUnit.SECONDS.sleep(2);
    long lastMajorCompactionTime = TestTransactionProcessor.lastMajorCompactionTime.get();
    Assert.assertTrue(String.format("Expected %d, but was %d", now, lastMajorCompactionTime),
                      lastMajorCompactionTime >= now);
  }

  @Test
  public void testPruneUpperBound() throws Exception {
    TransactionPruningPlugin transactionPruningPlugin = new TestTransactionPruningPlugin();
    transactionPruningPlugin.initialize(conf);

    try {
      // Run without a transaction snapshot first
      long now1 = 200;
      long inactiveTxTimeNow1 = 150 * TxConstants.MAX_TX_PER_MS;
      long expectedPruneUpperBound1 = -1;
      // fetch prune upper bound
      long pruneUpperBound1 = transactionPruningPlugin.fetchPruneUpperBound(now1, inactiveTxTimeNow1);
      Assert.assertEquals(expectedPruneUpperBound1, pruneUpperBound1);

      TimeRegions expectedRegions1 =
        new TimeRegions(now1,
                        ImmutableSortedSet.orderedBy(Bytes.BYTES_COMPARATOR)
                          .add(getRegionName(txDataTable1, Bytes.toBytes(0)))
                          .build());
      // Assert prune state is recorded correctly
      Assert.assertEquals(expectedRegions1, dataJanitorState.getRegionsOnOrBeforeTime(now1));
      Assert.assertEquals(expectedPruneUpperBound1,
                          dataJanitorState.getPruneUpperBoundForRegion(getRegionName(txDataTable1, Bytes.toBytes(0))));
      Assert.assertEquals(inactiveTxTimeNow1, dataJanitorState.getInactiveTransactionBoundForTime(now1));

      // Run prune complete
      transactionPruningPlugin.pruneComplete(now1, expectedPruneUpperBound1);

      // Assert prune state was cleaned up correctly based on the prune time
      Assert.assertEquals(expectedRegions1, dataJanitorState.getRegionsOnOrBeforeTime(now1));
      Assert.assertEquals(expectedPruneUpperBound1,
                          dataJanitorState.getPruneUpperBoundForRegion(getRegionName(txDataTable1, Bytes.toBytes(0))));
      Assert.assertEquals(inactiveTxTimeNow1, dataJanitorState.getInactiveTransactionBoundForTime(now1));

      // Create a new transaction snapshot, and run major compaction on txDataTable1
      // And run all assertions again
      long now2 = 300;
      long inactiveTxTimeNow2 = 250 * TxConstants.MAX_TX_PER_MS;
      long expectedPruneUpperBound2 = 200 * TxConstants.MAX_TX_PER_MS;
      InMemoryTransactionStateCache.setTransactionSnapshot(
        new TransactionSnapshot(expectedPruneUpperBound2, expectedPruneUpperBound2, expectedPruneUpperBound2,
                                ImmutableSet.of(expectedPruneUpperBound2),
                                ImmutableSortedMap.<Long, TransactionManager.InProgressTx>of()));
      TimeRegions expectedRegions2 =
        new TimeRegions(now2,
                        ImmutableSortedSet.orderedBy(Bytes.BYTES_COMPARATOR)
                          .add(getRegionName(txDataTable1, Bytes.toBytes(0)))
                          .build());
      testUtil.compact(txDataTable1, true);
      // Since the write to prune table happens async, we need to sleep a bit before checking the state of the table
      TimeUnit.SECONDS.sleep(2);
      long pruneUpperBound2 = transactionPruningPlugin.fetchPruneUpperBound(now2, inactiveTxTimeNow2);
      Assert.assertEquals(expectedPruneUpperBound2, pruneUpperBound2);

      Assert.assertEquals(expectedRegions2, dataJanitorState.getRegionsOnOrBeforeTime(now2));
      Assert.assertEquals(expectedPruneUpperBound2,
                          dataJanitorState.getPruneUpperBoundForRegion(getRegionName(txDataTable1, Bytes.toBytes(0))));
      Assert.assertEquals(inactiveTxTimeNow2, dataJanitorState.getInactiveTransactionBoundForTime(now2));
      Assert.assertEquals(expectedRegions1, dataJanitorState.getRegionsOnOrBeforeTime(now1));
      Assert.assertEquals(inactiveTxTimeNow1, dataJanitorState.getInactiveTransactionBoundForTime(now1));

      transactionPruningPlugin.pruneComplete(now2, pruneUpperBound2);
      Assert.assertEquals(expectedRegions2, dataJanitorState.getRegionsOnOrBeforeTime(now2));
      Assert.assertEquals(expectedPruneUpperBound2,
                          dataJanitorState.getPruneUpperBoundForRegion(getRegionName(txDataTable1, Bytes.toBytes(0))));
      Assert.assertEquals(inactiveTxTimeNow2, dataJanitorState.getInactiveTransactionBoundForTime(now2));
      Assert.assertNull(dataJanitorState.getRegionsOnOrBeforeTime(now1));
      Assert.assertEquals(expectedPruneUpperBound1, dataJanitorState.getInactiveTransactionBoundForTime(now1));

    } finally {
      transactionPruningPlugin.destroy();
    }
  }

  @Test
  public void testPruneEmptyTable() throws Exception {
    // Make sure that empty tables do not block the progress of pruning

    // Create an empty table
    TableName txEmptyTable = TableName.valueOf("emptyPruneTestTable");
    Table emptyHTable = createTable(txEmptyTable.getName(), new byte[][]{family}, false,
                                     Collections.singletonList(TestTransactionProcessor.class.getName()));

    TransactionPruningPlugin transactionPruningPlugin = new TestTransactionPruningPlugin();
    transactionPruningPlugin.initialize(conf);

    try {
      long now1 = System.currentTimeMillis();
      long inactiveTxTimeNow1 = (now1 - 150) * TxConstants.MAX_TX_PER_MS;
      long noPruneUpperBound = -1;
      long expectedPruneUpperBound1 = (now1 - 200) * TxConstants.MAX_TX_PER_MS;
      InMemoryTransactionStateCache.setTransactionSnapshot(
        new TransactionSnapshot(expectedPruneUpperBound1, expectedPruneUpperBound1, expectedPruneUpperBound1,
                                ImmutableSet.of(expectedPruneUpperBound1),
                                ImmutableSortedMap.<Long, TransactionManager.InProgressTx>of()));
      testUtil.compact(txEmptyTable, true);
      testUtil.compact(txDataTable1, true);
      // Since the write to prune table happens async, we need to sleep a bit before checking the state of the table
      TimeUnit.SECONDS.sleep(2);

      // fetch prune upper bound, there should be no prune upper bound since txEmptyTable cannot be compacted
      long pruneUpperBound1 = transactionPruningPlugin.fetchPruneUpperBound(now1, inactiveTxTimeNow1);
      Assert.assertEquals(noPruneUpperBound, pruneUpperBound1);
      transactionPruningPlugin.pruneComplete(now1, noPruneUpperBound);

      // Now flush the empty table, this will record the table region as empty, and then pruning will continue
      hBaseAdmin.flush(txEmptyTable);
      // Since the write to prune table happens async, we need to sleep a bit before checking the state of the table
      TimeUnit.SECONDS.sleep(2);

      // fetch prune upper bound, again, this time it should work
      pruneUpperBound1 = transactionPruningPlugin.fetchPruneUpperBound(now1, inactiveTxTimeNow1);
      Assert.assertEquals(expectedPruneUpperBound1, pruneUpperBound1);
      transactionPruningPlugin.pruneComplete(now1, expectedPruneUpperBound1);

      // Now add some data to the empty table
      // (adding data non-transactionally is okay too, we just need some data for the compaction to run)
      emptyHTable.put(new Put(Bytes.toBytes(1)).addColumn(family, qualifier, Bytes.toBytes(1)));
      emptyHTable.close();

      // Now run another compaction on txDataTable1 with an updated tx snapshot
      long now2 = System.currentTimeMillis();
      long inactiveTxTimeNow2 = (now2 - 150) * TxConstants.MAX_TX_PER_MS;
      long expectedPruneUpperBound2 = (now2 - 200) * TxConstants.MAX_TX_PER_MS;
      InMemoryTransactionStateCache.setTransactionSnapshot(
        new TransactionSnapshot(expectedPruneUpperBound2, expectedPruneUpperBound2, expectedPruneUpperBound2,
                                ImmutableSet.of(expectedPruneUpperBound2),
                                ImmutableSortedMap.<Long, TransactionManager.InProgressTx>of()));
      testUtil.flush(txEmptyTable);
      testUtil.compact(txDataTable1, true);
      // Since the write to prune table happens async, we need to sleep a bit before checking the state of the table
      TimeUnit.SECONDS.sleep(2);

      // Running a prune now should still return min(inactiveTxTimeNow1, expectedPruneUpperBound1) since
      // txEmptyTable is no longer empty. This information is returned since the txEmptyTable was recorded as being
      // empty in the previous run with inactiveTxTimeNow1
      long pruneUpperBound2 = transactionPruningPlugin.fetchPruneUpperBound(now2, inactiveTxTimeNow2);
      Assert.assertEquals(inactiveTxTimeNow1, pruneUpperBound2);
      transactionPruningPlugin.pruneComplete(now2, expectedPruneUpperBound1);

      // However, after compacting txEmptyTable we should get the latest upper bound
      testUtil.flush(txEmptyTable);
      testUtil.compact(txEmptyTable, true);
      // Since the write to prune table happens async, we need to sleep a bit before checking the state of the table
      TimeUnit.SECONDS.sleep(2);
      pruneUpperBound2 = transactionPruningPlugin.fetchPruneUpperBound(now2, inactiveTxTimeNow2);
      Assert.assertEquals(expectedPruneUpperBound2, pruneUpperBound2);
      transactionPruningPlugin.pruneComplete(now2, expectedPruneUpperBound2);
    } finally {
      transactionPruningPlugin.destroy();
      hBaseAdmin.disableTable(txEmptyTable);
      hBaseAdmin.deleteTable(txEmptyTable);
    }
  }

  @Test
  public void testPruneTransientTable() throws Exception {
    // Make sure that transient tables do not block the progress of pruning

    // Create a temp table
    TableName txTempTable = TableName.valueOf("tempTable");
    createTable(txTempTable.getName(), new byte[][]{family}, false,
                Collections.singletonList(TestTransactionProcessor.class.getName()));

    TableName txDataTable2 = null;

    TransactionPruningPlugin transactionPruningPlugin = new TestTransactionPruningPlugin();
    transactionPruningPlugin.initialize(conf);

    try {
      long now1 = System.currentTimeMillis();
      long inactiveTxTimeNow1 = (now1 - 150) * TxConstants.MAX_TX_PER_MS;
      long noPruneUpperBound = -1;
      long expectedPruneUpperBound1 = (now1 - 200) * TxConstants.MAX_TX_PER_MS;
      InMemoryTransactionStateCache.setTransactionSnapshot(
        new TransactionSnapshot(expectedPruneUpperBound1, expectedPruneUpperBound1, expectedPruneUpperBound1,
                                ImmutableSet.of(expectedPruneUpperBound1),
                                ImmutableSortedMap.<Long, TransactionManager.InProgressTx>of()));

      // fetch prune upper bound, there should be no prune upper bound since nothing has been compacted yet.
      // This run is only to store the initial set of regions
      long pruneUpperBound1 = transactionPruningPlugin.fetchPruneUpperBound(now1, inactiveTxTimeNow1);
      Assert.assertEquals(noPruneUpperBound, pruneUpperBound1);
      transactionPruningPlugin.pruneComplete(now1, noPruneUpperBound);

      // Now delete the transient table
      hBaseAdmin.disableTable(txTempTable);
      hBaseAdmin.deleteTable(txTempTable);
      txTempTable = null;

      // Compact the data table now
      testUtil.compact(txDataTable1, true);
      // Since the write to prune table happens async, we need to sleep a bit before checking the state of the table
      TimeUnit.SECONDS.sleep(2);

      // Create a new table that will not be compacted
      txDataTable2 = TableName.valueOf("invalidListPruneTestTable2");
      createTable(txDataTable2.getName(), new byte[][]{family}, false,
                  Collections.singletonList(TestTransactionProcessor.class.getName()));

      // fetch prune upper bound, there should be a prune upper bound even though txTempTable does not exist anymore,
      // and txDataTable2 has not been compacted/flushed yet
      long now2 = System.currentTimeMillis();
      long inactiveTxTimeNow2 = (now1 - 150) * TxConstants.MAX_TX_PER_MS;
      long pruneUpperBound2 = transactionPruningPlugin.fetchPruneUpperBound(now2, inactiveTxTimeNow2);
      Assert.assertEquals(expectedPruneUpperBound1, pruneUpperBound2);
      transactionPruningPlugin.pruneComplete(now2, expectedPruneUpperBound1);
    } finally {
      transactionPruningPlugin.destroy();
      if (txDataTable2 != null) {
        hBaseAdmin.disableTable(txDataTable2);
        hBaseAdmin.deleteTable(txDataTable2);
      }
      if (txTempTable != null) {
        hBaseAdmin.disableTable(txTempTable);
        hBaseAdmin.deleteTable(txTempTable);
      }
    }
  }

  private byte[] getRegionName(TableName dataTable, byte[] row) throws IOException {
    HRegionLocation regionLocation =
      testUtil.getConnection().getRegionLocator(dataTable).getRegionLocation(row);
    return regionLocation.getRegionInfo().getRegionName();
  }

  /**
   * A transaction co-processor that uses in-memory {@link TransactionSnapshot} for testing
   */
  @SuppressWarnings("WeakerAccess")
  public static class TestTransactionProcessor extends TransactionProcessor {
    private static final AtomicLong lastMajorCompactionTime = new AtomicLong(-1);

    @Override
    protected CacheSupplier<TransactionStateCache> getTransactionStateCacheSupplier(RegionCoprocessorEnvironment env) {
      return new CacheSupplier<TransactionStateCache>() {
        @Override
        public TransactionStateCache get() {
          return new InMemoryTransactionStateCache();
        }

        @Override
        public void release() {
          // no-op
        }
      };
    }

        @Override
        public void postCompact(
                org.apache.hadoop.hbase.coprocessor.ObserverContext<RegionCoprocessorEnvironment> c,
                Store store, StoreFile resultFile,
                org.apache.hadoop.hbase.regionserver.compactions.CompactionLifeCycleTracker tracker,
                CompactionRequest request) throws IOException {
            super.postCompact(c, store, resultFile, tracker, request);
            lastMajorCompactionTime.set(System.currentTimeMillis());
        }
    }

  /**
   * Used to supply in-memory {@link TransactionSnapshot} to {@link TestTransactionProcessor} for testing
   */
  @SuppressWarnings("WeakerAccess")
  public static class InMemoryTransactionStateCache extends TransactionStateCache {
    private static TransactionVisibilityState transactionSnapshot;

    public static void setTransactionSnapshot(TransactionVisibilityState transactionSnapshot) {
      InMemoryTransactionStateCache.transactionSnapshot = transactionSnapshot;
    }

    @Override
    protected void startUp() throws Exception {
      // Nothing to do
    }

    @Override
    protected void shutDown() throws Exception {
      // Nothing to do
    }

    @Override
    public TransactionVisibilityState getLatestState() {
      return transactionSnapshot;
    }
  }

  @SuppressWarnings("WeakerAccess")
  public static class TestTransactionPruningPlugin extends HBaseTransactionPruningPlugin {
    @Override
    protected boolean isTransactionalTable(HTableDescriptor tableDescriptor) {
      return tableDescriptor.hasCoprocessor(TestTransactionProcessor.class.getName());
    }
  }
}