@Test() public void testMultipleWritesThenGetAllInSingleRead() throws IOException { final PartitionedWriteAheadEventStore store = new PartitionedWriteAheadEventStore(createConfig(), writerFactory, readerFactory, EventReporter.NO_OP, new EventFileManager()); store.initialize(); assertEquals(-1, store.getMaxEventId()); final int numEvents = 20; final List<ProvenanceEventRecord> events = new ArrayList<>(numEvents); for (int i = 0; i < numEvents; i++) { final ProvenanceEventRecord event = createEvent(); store.addEvents(Collections.singleton(event)); assertEquals(i, store.getMaxEventId()); events.add(event); } List<ProvenanceEventRecord> eventsRead = store.getEvents(0L, numEvents, null, EventTransformer.EMPTY_TRANSFORMER); assertNotNull(eventsRead); assertEquals(numEvents, eventsRead.size()); for (int i = 0; i < numEvents; i++) { final ProvenanceEventRecord read = eventsRead.get(i); assertEquals(events.get(i), read); } eventsRead = store.getEvents(-1000, 1000, null, EventTransformer.EMPTY_TRANSFORMER); assertNotNull(eventsRead); assertTrue(eventsRead.isEmpty()); eventsRead = store.getEvents(10, 0, null, EventTransformer.EMPTY_TRANSFORMER); assertNotNull(eventsRead); assertTrue(eventsRead.isEmpty()); eventsRead = store.getEvents(10, 1, null, EventTransformer.EMPTY_TRANSFORMER); assertNotNull(eventsRead); assertFalse(eventsRead.isEmpty()); assertEquals(1, eventsRead.size()); assertEquals(events.get(10), eventsRead.get(0)); eventsRead = store.getEvents(20, 1000, null, EventTransformer.EMPTY_TRANSFORMER); assertNotNull(eventsRead); assertTrue(eventsRead.isEmpty()); }
@Override public void initialize() throws IOException { if (repoConfig.isCompressOnRollover()) { for (int i = 0; i < repoConfig.getIndexThreadPoolSize(); i++) { final EventFileCompressor compressor = new EventFileCompressor(filesToCompress, fileManager); compressionExecutor.submit(compressor); fileCompressors.add(compressor); } } super.initialize(); }
PartitionedWriteAheadEventStore extends PartitionedEventStore { @Override public void initialize() throws IOException { if (repoConfig.isCompressOnRollover()) { for (int i = 0; i < repoConfig.getIndexThreadPoolSize(); i++) { final EventFileCompressor compressor = new EventFileCompressor(filesToCompress, fileManager); compressionExecutor.submit(compressor); fileCompressors.add(compressor); } } super.initialize(); } }
PartitionedWriteAheadEventStore extends PartitionedEventStore { @Override public void initialize() throws IOException { if (repoConfig.isCompressOnRollover()) { for (int i = 0; i < repoConfig.getIndexThreadPoolSize(); i++) { final EventFileCompressor compressor = new EventFileCompressor(filesToCompress, fileManager); compressionExecutor.submit(compressor); fileCompressors.add(compressor); } } super.initialize(); } PartitionedWriteAheadEventStore(final RepositoryConfiguration repoConfig, final RecordWriterFactory recordWriterFactory,
        final RecordReaderFactory recordReaderFactory, final EventReporter eventReporter, final EventFileManager fileManager); }
PartitionedWriteAheadEventStore extends PartitionedEventStore { @Override public void initialize() throws IOException { if (repoConfig.isCompressOnRollover()) { for (int i = 0; i < repoConfig.getIndexThreadPoolSize(); i++) { final EventFileCompressor compressor = new EventFileCompressor(filesToCompress, fileManager); compressionExecutor.submit(compressor); fileCompressors.add(compressor); } } super.initialize(); } PartitionedWriteAheadEventStore(final RepositoryConfiguration repoConfig, final RecordWriterFactory recordWriterFactory,
        final RecordReaderFactory recordReaderFactory, final EventReporter eventReporter, final EventFileManager fileManager); @Override void initialize(); @Override void close(); @Override void reindexLatestEvents(final EventIndex eventIndex); }
PartitionedWriteAheadEventStore extends PartitionedEventStore { @Override public void initialize() throws IOException { if (repoConfig.isCompressOnRollover()) { for (int i = 0; i < repoConfig.getIndexThreadPoolSize(); i++) { final EventFileCompressor compressor = new EventFileCompressor(filesToCompress, fileManager); compressionExecutor.submit(compressor); fileCompressors.add(compressor); } } super.initialize(); } PartitionedWriteAheadEventStore(final RepositoryConfiguration repoConfig, final RecordWriterFactory recordWriterFactory,
        final RecordReaderFactory recordReaderFactory, final EventReporter eventReporter, final EventFileManager fileManager); @Override void initialize(); @Override void close(); @Override void reindexLatestEvents(final EventIndex eventIndex);  }