@Test public void testFileNotFound() throws IOException { final File file1 = new File("1.prov"); final List<File> files = new ArrayList<>(); files.add(file1); List<Long> eventIds = new ArrayList<>(); eventIds.add(1L); eventIds.add(5L); final RecordReaderFactory readerFactory = (file, logs, maxChars) -> { return RecordReaders.newRecordReader(file, logs, maxChars); }; final SelectiveRecordReaderEventIterator itr = new SelectiveRecordReaderEventIterator(files, readerFactory, eventIds, 65536); final Optional<ProvenanceEventRecord> firstRecordOption = itr.nextEvent(); assertFalse(firstRecordOption.isPresent()); }
@Override public Optional<ProvenanceEventRecord> nextEvent() throws IOException { if (closed) { throw new IOException("EventIterator is already closed"); } final long start = System.nanoTime(); try { while (idIterator.hasNext()) { final long eventId = idIterator.next(); final File fileForEvent = getFileForEventId(eventId); if (fileForEvent == null) { continue; } try { if (!fileForEvent.equals(currentFile)) { if (reader != null) { try { reader.close(); } catch (final Exception e) { logger.warn("Failed to close {}; some resources may not be cleaned up appropriately", reader); } } reader = readerFactory.newRecordReader(fileForEvent, Collections.emptyList(), maxAttributeChars); this.currentFile = fileForEvent; } final Optional<ProvenanceEventRecord> eventOption = reader.skipToEvent(eventId); if (eventOption.isPresent() && eventOption.get().getEventId() == eventId) { reader.nextRecord(); return eventOption; } } catch (final FileNotFoundException | EOFException e) { logger.warn("Failed to retrieve Event with ID {}", eventId, e); } } return Optional.empty(); } finally { final long ms = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); logger.trace("Took {} ms to read next event", ms); } }
SelectiveRecordReaderEventIterator implements EventIterator { @Override public Optional<ProvenanceEventRecord> nextEvent() throws IOException { if (closed) { throw new IOException("EventIterator is already closed"); } final long start = System.nanoTime(); try { while (idIterator.hasNext()) { final long eventId = idIterator.next(); final File fileForEvent = getFileForEventId(eventId); if (fileForEvent == null) { continue; } try { if (!fileForEvent.equals(currentFile)) { if (reader != null) { try { reader.close(); } catch (final Exception e) { logger.warn("Failed to close {}; some resources may not be cleaned up appropriately", reader); } } reader = readerFactory.newRecordReader(fileForEvent, Collections.emptyList(), maxAttributeChars); this.currentFile = fileForEvent; } final Optional<ProvenanceEventRecord> eventOption = reader.skipToEvent(eventId); if (eventOption.isPresent() && eventOption.get().getEventId() == eventId) { reader.nextRecord(); return eventOption; } } catch (final FileNotFoundException | EOFException e) { logger.warn("Failed to retrieve Event with ID {}", eventId, e); } } return Optional.empty(); } finally { final long ms = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); logger.trace("Took {} ms to read next event", ms); } } }
SelectiveRecordReaderEventIterator implements EventIterator { @Override public Optional<ProvenanceEventRecord> nextEvent() throws IOException { if (closed) { throw new IOException("EventIterator is already closed"); } final long start = System.nanoTime(); try { while (idIterator.hasNext()) { final long eventId = idIterator.next(); final File fileForEvent = getFileForEventId(eventId); if (fileForEvent == null) { continue; } try { if (!fileForEvent.equals(currentFile)) { if (reader != null) { try { reader.close(); } catch (final Exception e) { logger.warn("Failed to close {}; some resources may not be cleaned up appropriately", reader); } } reader = readerFactory.newRecordReader(fileForEvent, Collections.emptyList(), maxAttributeChars); this.currentFile = fileForEvent; } final Optional<ProvenanceEventRecord> eventOption = reader.skipToEvent(eventId); if (eventOption.isPresent() && eventOption.get().getEventId() == eventId) { reader.nextRecord(); return eventOption; } } catch (final FileNotFoundException | EOFException e) { logger.warn("Failed to retrieve Event with ID {}", eventId, e); } } return Optional.empty(); } finally { final long ms = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); logger.trace("Took {} ms to read next event", ms); } } SelectiveRecordReaderEventIterator(final List<File> filesToRead, final RecordReaderFactory readerFactory, final List<Long> eventIds, final int maxAttributeChars); }
SelectiveRecordReaderEventIterator implements EventIterator { @Override public Optional<ProvenanceEventRecord> nextEvent() throws IOException { if (closed) { throw new IOException("EventIterator is already closed"); } final long start = System.nanoTime(); try { while (idIterator.hasNext()) { final long eventId = idIterator.next(); final File fileForEvent = getFileForEventId(eventId); if (fileForEvent == null) { continue; } try { if (!fileForEvent.equals(currentFile)) { if (reader != null) { try { reader.close(); } catch (final Exception e) { logger.warn("Failed to close {}; some resources may not be cleaned up appropriately", reader); } } reader = readerFactory.newRecordReader(fileForEvent, Collections.emptyList(), maxAttributeChars); this.currentFile = fileForEvent; } final Optional<ProvenanceEventRecord> eventOption = reader.skipToEvent(eventId); if (eventOption.isPresent() && eventOption.get().getEventId() == eventId) { reader.nextRecord(); return eventOption; } } catch (final FileNotFoundException | EOFException e) { logger.warn("Failed to retrieve Event with ID {}", eventId, e); } } return Optional.empty(); } finally { final long ms = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); logger.trace("Took {} ms to read next event", ms); } } SelectiveRecordReaderEventIterator(final List<File> filesToRead, final RecordReaderFactory readerFactory, final List<Long> eventIds, final int maxAttributeChars); @Override void close(); @Override Optional<ProvenanceEventRecord> nextEvent(); }
SelectiveRecordReaderEventIterator implements EventIterator { @Override public Optional<ProvenanceEventRecord> nextEvent() throws IOException { if (closed) { throw new IOException("EventIterator is already closed"); } final long start = System.nanoTime(); try { while (idIterator.hasNext()) { final long eventId = idIterator.next(); final File fileForEvent = getFileForEventId(eventId); if (fileForEvent == null) { continue; } try { if (!fileForEvent.equals(currentFile)) { if (reader != null) { try { reader.close(); } catch (final Exception e) { logger.warn("Failed to close {}; some resources may not be cleaned up appropriately", reader); } } reader = readerFactory.newRecordReader(fileForEvent, Collections.emptyList(), maxAttributeChars); this.currentFile = fileForEvent; } final Optional<ProvenanceEventRecord> eventOption = reader.skipToEvent(eventId); if (eventOption.isPresent() && eventOption.get().getEventId() == eventId) { reader.nextRecord(); return eventOption; } } catch (final FileNotFoundException | EOFException e) { logger.warn("Failed to retrieve Event with ID {}", eventId, e); } } return Optional.empty(); } finally { final long ms = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); logger.trace("Took {} ms to read next event", ms); } } SelectiveRecordReaderEventIterator(final List<File> filesToRead, final RecordReaderFactory readerFactory, final List<Long> eventIds, final int maxAttributeChars); @Override void close(); @Override Optional<ProvenanceEventRecord> nextEvent();  }