@Test public void testGetEventsWithMinIdAndCount() throws IOException { final RepositoryConfiguration config = createConfig(); config.setMaxEventFileCount(100); final PartitionedWriteAheadEventStore store = new PartitionedWriteAheadEventStore(config, writerFactory, readerFactory, EventReporter.NO_OP, new EventFileManager()); store.initialize(); final int numEvents = 50_000; final List<ProvenanceEventRecord> events = new ArrayList<>(numEvents); for (int i = 0; i < numEvents; i++) { final ProvenanceEventRecord event = createEvent(); store.addEvents(Collections.singleton(event)); if (i < 1000) { events.add(event); } } assertTrue(store.getEvents(-1000L, 1000).isEmpty()); assertEquals(events, store.getEvents(0, events.size())); assertEquals(events, store.getEvents(-30, events.size())); assertEquals(events.subList(10, events.size()), store.getEvents(10L, events.size() - 10)); assertTrue(store.getEvents(numEvents, 100).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);  }