HdfsState implements State { @Override public void beginCommit(Long txId) { if (txId <= lastSeenTxn.txnid) { LOG.info("txID {} is already processed, lastSeenTxn {}. Triggering recovery.", txId, lastSeenTxn); long start = System.currentTimeMillis(); options.recover(lastSeenTxn.dataFilePath, lastSeenTxn.offset); LOG.info("Recovery took {} ms.", System.currentTimeMillis() - start); } updateIndex(txId); } HdfsState(Options options); @Override void beginCommit(Long txId); @Override void commit(Long txId); void updateState(List<TridentTuple> tuples, TridentCollector tridentCollector); static final Logger LOG; }
@Test public void testIndexFileCreation() throws Exception { HdfsState state = createHdfsState(); state.beginCommit(1L); Collection<File> files = FileUtils.listFiles(new File(TEST_OUT_DIR), null, false); File hdfsIndexFile = Paths.get(TEST_OUT_DIR, INDEX_FILE_PREFIX + TEST_TOPOLOGY_NAME + ".0").toFile(); Assert.assertTrue(files.contains(hdfsIndexFile)); }