@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);
    }
@Test
    public void testIndexFileCreation() {
        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();
        assertTrue(files.contains(hdfsIndexFile));
    }