TransactionPruningService extends AbstractIdleService { public TransactionPruningService(Configuration conf, TransactionManager txManager) { this.conf = conf; this.txManager = txManager; this.pruneEnabled = conf.getBoolean(TxConstants.TransactionPruning.PRUNE_ENABLE, TxConstants.TransactionPruning.DEFAULT_PRUNE_ENABLE); this.scheduleInterval = conf.getLong(TxConstants.TransactionPruning.PRUNE_INTERVAL, TxConstants.TransactionPruning.DEFAULT_PRUNE_INTERVAL); } TransactionPruningService(Configuration conf, TransactionManager txManager); void pruneNow();  }
@Test public void testTransactionPruningService() throws Exception { Configuration conf = new Configuration(); conf.set(TxConstants.TransactionPruning.PLUGINS, "data.tx.txprune.plugin.mockPlugin1, data.tx.txprune.plugin.mockPlugin2"); conf.set("data.tx.txprune.plugin.mockPlugin1.class", "org.apache.tephra.txprune.TransactionPruningServiceTest$MockPlugin1"); conf.set("data.tx.txprune.plugin.mockPlugin2.class", "org.apache.tephra.txprune.TransactionPruningServiceTest$MockPlugin2"); conf.setBoolean(TxConstants.TransactionPruning.PRUNE_ENABLE, true); conf.setInt(TxConstants.Manager.CFG_TX_MAX_LIFETIME, 10); conf.setLong(TxConstants.TransactionPruning.PRUNE_GRACE_PERIOD, 0); long m = 1000; long n = m * TxConstants.MAX_TX_PER_MS; Iterator<Long> currentTime = Iterators.cycle(120L * m, 220L * m); Iterator<Transaction> txns = Iterators.cycle(new Transaction(100 * n, 110 * n, new long[]{40 * n, 50 * n, 60 * n, 70 * n}, new long[]{80 * n, 90 * n}, 80 * n), new Transaction(200 * n, 210 * n, new long[]{60 * n, 75 * n, 78 * n, 100 * n, 110 * n, 120 * n}, new long[]{80 * n, 90 * n}, 80 * n)); Iterator<Long> pruneUpperBoundsPlugin1 = Iterators.cycle(60L * n, 80L * n); Iterator<Long> pruneUpperBoundsPlugin2 = Iterators.cycle(70L * n, 77L * n); TestTransactionPruningRunnable.setCurrentTime(currentTime); MockTxManager.setTxIter(txns); MockPlugin1.setPruneUpperBoundIter(pruneUpperBoundsPlugin1); MockPlugin2.setPruneUpperBoundIter(pruneUpperBoundsPlugin2); MockTxManager mockTxManager = new MockTxManager(conf); TransactionPruningService pruningService = new TestTransactionPruningService(conf, mockTxManager); pruningService.startAndWait(); int pruneRuns = TestTransactionPruningRunnable.getRuns(); pruningService.pruneNow(); pruningService.pruneNow(); pruningService.pruneNow(); TestTransactionPruningRunnable.waitForRuns(pruneRuns + 3, 5, TimeUnit.MILLISECONDS); pruningService.stopAndWait(); Assert.assertEquals(ImmutableList.of(110L * n - 1, 210L * n - 1), limitTwo(MockPlugin1.getInactiveTransactionBoundList())); Assert.assertEquals(ImmutableList.of(110L * n - 1, 210L * n - 1), limitTwo(MockPlugin2.getInactiveTransactionBoundList())); Assert.assertEquals(ImmutableList.of(ImmutableSet.of(40L * n, 50L * n, 60L * n), ImmutableSet.of(60L * n, 75L * n)), limitTwo(MockTxManager.getPrunedInvalidsList())); Assert.assertEquals(ImmutableList.of(60L * n, 75L * n), limitTwo(MockPlugin1.getMaxPrunedInvalidList())); Assert.assertEquals(ImmutableList.of(60L * n, 75L * n), limitTwo(MockPlugin2.getMaxPrunedInvalidList())); }