HdfsState implements State { void prepare(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions) { this.options.prepare(conf, partitionIndex, numPartitions); initLastTxn(conf, partitionIndex); } 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 testPrepare() throws Exception { HdfsState state = createHdfsState(); Collection<File> files = FileUtils.listFiles(new File(TEST_OUT_DIR), null, false); File hdfsDataFile = Paths.get(TEST_OUT_DIR, FILE_NAME_PREFIX + "0").toFile(); Assert.assertTrue(files.contains(hdfsDataFile)); }