@Test public void testMergeJournals() throws IOException, InterruptedException { final RepositoryConfiguration config = createConfiguration(); config.setMaxEventFileLife(3, TimeUnit.SECONDS); repo = new PersistentProvenanceRepository(config, DEFAULT_ROLLOVER_MILLIS); repo.initialize(getEventReporter(), null, null, IdentifierLookup.EMPTY); final Map<String, String> attributes = new HashMap<>(); final ProvenanceEventBuilder builder = new StandardProvenanceEventRecord.Builder(); builder.setEventTime(System.currentTimeMillis()); builder.setEventType(ProvenanceEventType.RECEIVE); builder.setTransitUri("nifi: attributes.put("uuid", "12345678-0000-0000-0000-012345678912"); builder.fromFlowFile(createFlowFile(3L, 3000L, attributes)); builder.setComponentId("1234"); builder.setComponentType("dummy processor"); final ProvenanceEventRecord record = builder.build(); final ExecutorService exec = Executors.newFixedThreadPool(10); for (int i = 0; i < 10000; i++) { exec.submit(new Runnable() { @Override public void run() { repo.registerEvent(record); } }); } repo.waitForRollover(); final File storageDir = config.getStorageDirectories().values().iterator().next(); long counter = 0; for (final File file : storageDir.listFiles()) { if (file.isFile()) { try (RecordReader reader = RecordReaders.newRecordReader(file, null, 2048)) { ProvenanceEventRecord r = null; while ((r = reader.nextRecord()) != null) { assertEquals(counter++, r.getEventId()); } } } } assertEquals(10000, counter); }
File mergeJournals(final List<File> journalFiles, final File suggestedMergeFile, final EventReporter eventReporter) throws IOException { logger.debug("Merging {} to {}", journalFiles, suggestedMergeFile); if (this.closed.get()) { logger.info("Provenance Repository has been closed; will not merge journal files to {}", suggestedMergeFile); return null; } if (journalFiles.isEmpty()) { logger.debug("Couldn't merge journals: Journal Files is empty; won't merge journals"); return null; } Collections.sort(journalFiles, new Comparator<File>() { @Override public int compare(final File o1, final File o2) { final String suffix1 = LuceneUtil.substringAfterLast(o1.getName(), "."); final String suffix2 = LuceneUtil.substringAfterLast(o2.getName(), "."); try { final int journalIndex1 = Integer.parseInt(suffix1); final int journalIndex2 = Integer.parseInt(suffix2); return Integer.compare(journalIndex1, journalIndex2); } catch (final NumberFormatException nfe) { return o1.getName().compareTo(o2.getName()); } } }); final List<File> availableFiles = filterUnavailableFiles(journalFiles); final int numAvailableFiles = availableFiles.size(); if (numAvailableFiles > 0) { if (suggestedMergeFile.exists()) { logger.warn("Merged Journal File {} already exists; however, all partial journal files also exist " + "so assuming that the merge did not finish. Repeating procedure in order to ensure consistency."); final DeleteIndexAction deleteAction = new DeleteIndexAction(this, indexConfig, getIndexManager()); try { deleteAction.execute(suggestedMergeFile); } catch (final Exception e) { logger.warn("Failed to delete records from Journal File {} from the index; this could potentially result in duplicates. Failure was due to {}", suggestedMergeFile, e.toString()); if (logger.isDebugEnabled()) { logger.warn("", e); } } if (!suggestedMergeFile.delete()) { logger.error("Failed to delete partially written Provenance Journal File {}. This may result in events from this journal " + "file not being able to be displayed. This file should be deleted manually.", suggestedMergeFile); } final File tocFile = TocUtil.getTocFile(suggestedMergeFile); if (tocFile.exists() && !tocFile.delete()) { logger.error("Failed to delete .toc file {}; this may result in not being able to read the Provenance Events from the {} Journal File. " + "This can be corrected by manually deleting the {} file", tocFile, suggestedMergeFile, tocFile); } } } else { logger.warn("Cannot merge journal files {} because they do not exist on disk", journalFiles); return null; } final long startNanos = System.nanoTime(); final List<RecordReader> readers = new ArrayList<>(); int records = 0; final boolean isCompress = configuration.isCompressOnRollover(); final File writerFile = isCompress ? new File(suggestedMergeFile.getParentFile(), suggestedMergeFile.getName() + ".gz") : suggestedMergeFile; try { for (final File journalFile : availableFiles) { try { readers.add(RecordReaders.newRecordReader(journalFile, null, Integer.MAX_VALUE)); } catch (final EOFException eof) { } catch (final IOException ioe) { logger.warn("Unable to merge {} with other Journal Files due to {}", journalFile, ioe.toString()); if (logger.isDebugEnabled()) { logger.warn("", ioe); } if (eventReporter != null) { eventReporter.reportEvent(Severity.ERROR, EVENT_CATEGORY, "Failed to merge Journal Files due to " + ioe.toString()); } } } final SortedMap<StandardProvenanceEventRecord, RecordReader> recordToReaderMap = new TreeMap<>(new Comparator<StandardProvenanceEventRecord>() { @Override public int compare(final StandardProvenanceEventRecord o1, final StandardProvenanceEventRecord o2) { return Long.compare(o1.getEventId(), o2.getEventId()); } }); long minEventId = 0L; long earliestTimestamp = System.currentTimeMillis(); for (final RecordReader reader : readers) { StandardProvenanceEventRecord record = null; try { record = reader.nextRecord(); } catch (final EOFException eof) { } catch (final Exception e) { logger.warn("Failed to generate Provenance Event Record from Journal due to " + e + "; it's " + "possible that the record wasn't completely written to the file. This journal will be " + "skipped."); if (logger.isDebugEnabled()) { logger.warn("", e); } if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, "Failed to read Provenance Event " + "Record from Journal due to " + e + "; it's possible that the record wasn't " + "completely written to the file. This journal will be skipped."); } } if (record == null) { continue; } if (record.getEventTime() < earliestTimestamp) { earliestTimestamp = record.getEventTime(); } if (record.getEventId() < minEventId) { minEventId = record.getEventId(); } recordToReaderMap.put(record, reader); } final RingBuffer<ProvenanceEventRecord> latestRecords = new RingBuffer<>(1000); try (final RecordWriter writer = RecordWriters.newSchemaRecordWriter(writerFile, idGenerator, configuration.isCompressOnRollover(), true)) { writer.writeHeader(minEventId); final IndexingAction indexingAction = createIndexingAction(); final File indexingDirectory = indexConfig.getWritableIndexDirectory(writerFile, earliestTimestamp); long maxId = 0L; final BlockingQueue<Tuple<StandardProvenanceEventRecord, Integer>> eventQueue = new LinkedBlockingQueue<>(100); final AtomicBoolean finishedAdding = new AtomicBoolean(false); final List<Future<?>> futures = new ArrayList<>(); final EventIndexWriter indexWriter = getIndexManager().borrowIndexWriter(indexingDirectory); try { final ExecutorService exec = Executors.newFixedThreadPool(configuration.getIndexThreadPoolSize(), new ThreadFactory() { @Override public Thread newThread(final Runnable r) { final Thread t = Executors.defaultThreadFactory().newThread(r); t.setName("Index Provenance Events"); return t; } }); final AtomicInteger indexingFailureCount = new AtomicInteger(0); try { for (int i = 0; i < configuration.getIndexThreadPoolSize(); i++) { final Callable<Object> callable = new Callable<Object>() { @Override public Object call() throws IOException { while (!eventQueue.isEmpty() || !finishedAdding.get()) { try { final Tuple<StandardProvenanceEventRecord, Integer> tuple; try { tuple = eventQueue.poll(10, TimeUnit.MILLISECONDS); } catch (final InterruptedException ie) { Thread.currentThread().interrupt(); continue; } if (tuple == null) { continue; } indexingAction.index(tuple.getKey(), indexWriter.getIndexWriter(), tuple.getValue()); } catch (final Throwable t) { logger.error("Failed to index Provenance Event for " + writerFile + " to " + indexingDirectory, t); if (indexingFailureCount.incrementAndGet() >= MAX_INDEXING_FAILURE_COUNT) { return null; } } } return null; } }; final Future<?> future = exec.submit(callable); futures.add(future); } boolean indexEvents = true; while (!recordToReaderMap.isEmpty()) { final Map.Entry<StandardProvenanceEventRecord, RecordReader> entry = recordToReaderMap.entrySet().iterator().next(); final StandardProvenanceEventRecord record = entry.getKey(); final RecordReader reader = entry.getValue(); writer.writeRecord(record); final int blockIndex = writer.getTocWriter().getCurrentBlockIndex(); boolean accepted = false; while (!accepted && indexEvents) { try { accepted = eventQueue.offer(new Tuple<>(record, blockIndex), 10, TimeUnit.MILLISECONDS); } catch (final InterruptedException ie) { Thread.currentThread().interrupt(); } if (!accepted && indexingFailureCount.get() >= MAX_INDEXING_FAILURE_COUNT) { indexEvents = false; eventQueue.clear(); final String warning = String.format("Indexing Provenance Events for %s has failed %s times. This exceeds the maximum threshold of %s failures, " + "so no more Provenance Events will be indexed for this Provenance file.", writerFile, indexingFailureCount.get(), MAX_INDEXING_FAILURE_COUNT); logger.warn(warning); if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, warning); } } } maxId = record.getEventId(); latestRecords.add(truncateAttributes(record)); records++; recordToReaderMap.remove(record); StandardProvenanceEventRecord nextRecord = null; try { nextRecord = reader.nextRecord(); } catch (final EOFException eof) { } catch (final Exception e) { logger.warn("Failed to generate Provenance Event Record from Journal due to " + e + "; it's possible that the record wasn't completely written to the file. " + "The remainder of this journal will be skipped."); if (logger.isDebugEnabled()) { logger.warn("", e); } if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, "Failed to read " + "Provenance Event Record from Journal due to " + e + "; it's possible " + "that the record wasn't completely written to the file. The remainder " + "of this journal will be skipped."); } } if (nextRecord != null) { recordToReaderMap.put(nextRecord, reader); } } } finally { finishedAdding.set(true); exec.shutdown(); } for (final Future<?> future : futures) { try { future.get(); } catch (final ExecutionException ee) { final Throwable t = ee.getCause(); if (t instanceof RuntimeException) { throw (RuntimeException) t; } throw new RuntimeException(t); } catch (final InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException("Thread interrupted"); } } } finally { getIndexManager().returnIndexWriter(indexWriter); } indexConfig.setMaxIdIndexed(maxId); } final RingBuffer<ProvenanceEventRecord> latestRecordBuffer = this.latestRecords; latestRecords.forEach(new ForEachEvaluator<ProvenanceEventRecord>() { @Override public boolean evaluate(final ProvenanceEventRecord event) { latestRecordBuffer.add(event); return true; } }); } finally { for (final RecordReader reader : readers) { try { reader.close(); } catch (final IOException ioe) { } } } for (final File journalFile : availableFiles) { if (!journalFile.delete() && journalFile.exists()) { logger.warn("Failed to remove temporary journal file {}; this file should be cleaned up manually", journalFile.getAbsolutePath()); if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, "Failed to remove temporary journal file " + journalFile.getAbsolutePath() + "; this file should be cleaned up manually"); } } final File tocFile = TocUtil.getTocFile(journalFile); if (!tocFile.delete() && tocFile.exists()) { logger.warn("Failed to remove temporary journal TOC file {}; this file should be cleaned up manually", tocFile.getAbsolutePath()); if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, "Failed to remove temporary journal TOC file " + tocFile.getAbsolutePath() + "; this file should be cleaned up manually"); } } } if (records == 0) { writerFile.delete(); logger.debug("Couldn't merge journals: No Records to merge"); return null; } else { final long nanos = System.nanoTime() - startNanos; final long millis = TimeUnit.MILLISECONDS.convert(nanos, TimeUnit.NANOSECONDS); logger.info("Successfully merged {} journal files ({} records) into single Provenance Log File {} in {} milliseconds", numAvailableFiles, records, suggestedMergeFile, millis); } return writerFile; }
PersistentProvenanceRepository implements ProvenanceRepository { File mergeJournals(final List<File> journalFiles, final File suggestedMergeFile, final EventReporter eventReporter) throws IOException { logger.debug("Merging {} to {}", journalFiles, suggestedMergeFile); if (this.closed.get()) { logger.info("Provenance Repository has been closed; will not merge journal files to {}", suggestedMergeFile); return null; } if (journalFiles.isEmpty()) { logger.debug("Couldn't merge journals: Journal Files is empty; won't merge journals"); return null; } Collections.sort(journalFiles, new Comparator<File>() { @Override public int compare(final File o1, final File o2) { final String suffix1 = LuceneUtil.substringAfterLast(o1.getName(), "."); final String suffix2 = LuceneUtil.substringAfterLast(o2.getName(), "."); try { final int journalIndex1 = Integer.parseInt(suffix1); final int journalIndex2 = Integer.parseInt(suffix2); return Integer.compare(journalIndex1, journalIndex2); } catch (final NumberFormatException nfe) { return o1.getName().compareTo(o2.getName()); } } }); final List<File> availableFiles = filterUnavailableFiles(journalFiles); final int numAvailableFiles = availableFiles.size(); if (numAvailableFiles > 0) { if (suggestedMergeFile.exists()) { logger.warn("Merged Journal File {} already exists; however, all partial journal files also exist " + "so assuming that the merge did not finish. Repeating procedure in order to ensure consistency."); final DeleteIndexAction deleteAction = new DeleteIndexAction(this, indexConfig, getIndexManager()); try { deleteAction.execute(suggestedMergeFile); } catch (final Exception e) { logger.warn("Failed to delete records from Journal File {} from the index; this could potentially result in duplicates. Failure was due to {}", suggestedMergeFile, e.toString()); if (logger.isDebugEnabled()) { logger.warn("", e); } } if (!suggestedMergeFile.delete()) { logger.error("Failed to delete partially written Provenance Journal File {}. This may result in events from this journal " + "file not being able to be displayed. This file should be deleted manually.", suggestedMergeFile); } final File tocFile = TocUtil.getTocFile(suggestedMergeFile); if (tocFile.exists() && !tocFile.delete()) { logger.error("Failed to delete .toc file {}; this may result in not being able to read the Provenance Events from the {} Journal File. " + "This can be corrected by manually deleting the {} file", tocFile, suggestedMergeFile, tocFile); } } } else { logger.warn("Cannot merge journal files {} because they do not exist on disk", journalFiles); return null; } final long startNanos = System.nanoTime(); final List<RecordReader> readers = new ArrayList<>(); int records = 0; final boolean isCompress = configuration.isCompressOnRollover(); final File writerFile = isCompress ? new File(suggestedMergeFile.getParentFile(), suggestedMergeFile.getName() + ".gz") : suggestedMergeFile; try { for (final File journalFile : availableFiles) { try { readers.add(RecordReaders.newRecordReader(journalFile, null, Integer.MAX_VALUE)); } catch (final EOFException eof) { } catch (final IOException ioe) { logger.warn("Unable to merge {} with other Journal Files due to {}", journalFile, ioe.toString()); if (logger.isDebugEnabled()) { logger.warn("", ioe); } if (eventReporter != null) { eventReporter.reportEvent(Severity.ERROR, EVENT_CATEGORY, "Failed to merge Journal Files due to " + ioe.toString()); } } } final SortedMap<StandardProvenanceEventRecord, RecordReader> recordToReaderMap = new TreeMap<>(new Comparator<StandardProvenanceEventRecord>() { @Override public int compare(final StandardProvenanceEventRecord o1, final StandardProvenanceEventRecord o2) { return Long.compare(o1.getEventId(), o2.getEventId()); } }); long minEventId = 0L; long earliestTimestamp = System.currentTimeMillis(); for (final RecordReader reader : readers) { StandardProvenanceEventRecord record = null; try { record = reader.nextRecord(); } catch (final EOFException eof) { } catch (final Exception e) { logger.warn("Failed to generate Provenance Event Record from Journal due to " + e + "; it's " + "possible that the record wasn't completely written to the file. This journal will be " + "skipped."); if (logger.isDebugEnabled()) { logger.warn("", e); } if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, "Failed to read Provenance Event " + "Record from Journal due to " + e + "; it's possible that the record wasn't " + "completely written to the file. This journal will be skipped."); } } if (record == null) { continue; } if (record.getEventTime() < earliestTimestamp) { earliestTimestamp = record.getEventTime(); } if (record.getEventId() < minEventId) { minEventId = record.getEventId(); } recordToReaderMap.put(record, reader); } final RingBuffer<ProvenanceEventRecord> latestRecords = new RingBuffer<>(1000); try (final RecordWriter writer = RecordWriters.newSchemaRecordWriter(writerFile, idGenerator, configuration.isCompressOnRollover(), true)) { writer.writeHeader(minEventId); final IndexingAction indexingAction = createIndexingAction(); final File indexingDirectory = indexConfig.getWritableIndexDirectory(writerFile, earliestTimestamp); long maxId = 0L; final BlockingQueue<Tuple<StandardProvenanceEventRecord, Integer>> eventQueue = new LinkedBlockingQueue<>(100); final AtomicBoolean finishedAdding = new AtomicBoolean(false); final List<Future<?>> futures = new ArrayList<>(); final EventIndexWriter indexWriter = getIndexManager().borrowIndexWriter(indexingDirectory); try { final ExecutorService exec = Executors.newFixedThreadPool(configuration.getIndexThreadPoolSize(), new ThreadFactory() { @Override public Thread newThread(final Runnable r) { final Thread t = Executors.defaultThreadFactory().newThread(r); t.setName("Index Provenance Events"); return t; } }); final AtomicInteger indexingFailureCount = new AtomicInteger(0); try { for (int i = 0; i < configuration.getIndexThreadPoolSize(); i++) { final Callable<Object> callable = new Callable<Object>() { @Override public Object call() throws IOException { while (!eventQueue.isEmpty() || !finishedAdding.get()) { try { final Tuple<StandardProvenanceEventRecord, Integer> tuple; try { tuple = eventQueue.poll(10, TimeUnit.MILLISECONDS); } catch (final InterruptedException ie) { Thread.currentThread().interrupt(); continue; } if (tuple == null) { continue; } indexingAction.index(tuple.getKey(), indexWriter.getIndexWriter(), tuple.getValue()); } catch (final Throwable t) { logger.error("Failed to index Provenance Event for " + writerFile + " to " + indexingDirectory, t); if (indexingFailureCount.incrementAndGet() >= MAX_INDEXING_FAILURE_COUNT) { return null; } } } return null; } }; final Future<?> future = exec.submit(callable); futures.add(future); } boolean indexEvents = true; while (!recordToReaderMap.isEmpty()) { final Map.Entry<StandardProvenanceEventRecord, RecordReader> entry = recordToReaderMap.entrySet().iterator().next(); final StandardProvenanceEventRecord record = entry.getKey(); final RecordReader reader = entry.getValue(); writer.writeRecord(record); final int blockIndex = writer.getTocWriter().getCurrentBlockIndex(); boolean accepted = false; while (!accepted && indexEvents) { try { accepted = eventQueue.offer(new Tuple<>(record, blockIndex), 10, TimeUnit.MILLISECONDS); } catch (final InterruptedException ie) { Thread.currentThread().interrupt(); } if (!accepted && indexingFailureCount.get() >= MAX_INDEXING_FAILURE_COUNT) { indexEvents = false; eventQueue.clear(); final String warning = String.format("Indexing Provenance Events for %s has failed %s times. This exceeds the maximum threshold of %s failures, " + "so no more Provenance Events will be indexed for this Provenance file.", writerFile, indexingFailureCount.get(), MAX_INDEXING_FAILURE_COUNT); logger.warn(warning); if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, warning); } } } maxId = record.getEventId(); latestRecords.add(truncateAttributes(record)); records++; recordToReaderMap.remove(record); StandardProvenanceEventRecord nextRecord = null; try { nextRecord = reader.nextRecord(); } catch (final EOFException eof) { } catch (final Exception e) { logger.warn("Failed to generate Provenance Event Record from Journal due to " + e + "; it's possible that the record wasn't completely written to the file. " + "The remainder of this journal will be skipped."); if (logger.isDebugEnabled()) { logger.warn("", e); } if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, "Failed to read " + "Provenance Event Record from Journal due to " + e + "; it's possible " + "that the record wasn't completely written to the file. The remainder " + "of this journal will be skipped."); } } if (nextRecord != null) { recordToReaderMap.put(nextRecord, reader); } } } finally { finishedAdding.set(true); exec.shutdown(); } for (final Future<?> future : futures) { try { future.get(); } catch (final ExecutionException ee) { final Throwable t = ee.getCause(); if (t instanceof RuntimeException) { throw (RuntimeException) t; } throw new RuntimeException(t); } catch (final InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException("Thread interrupted"); } } } finally { getIndexManager().returnIndexWriter(indexWriter); } indexConfig.setMaxIdIndexed(maxId); } final RingBuffer<ProvenanceEventRecord> latestRecordBuffer = this.latestRecords; latestRecords.forEach(new ForEachEvaluator<ProvenanceEventRecord>() { @Override public boolean evaluate(final ProvenanceEventRecord event) { latestRecordBuffer.add(event); return true; } }); } finally { for (final RecordReader reader : readers) { try { reader.close(); } catch (final IOException ioe) { } } } for (final File journalFile : availableFiles) { if (!journalFile.delete() && journalFile.exists()) { logger.warn("Failed to remove temporary journal file {}; this file should be cleaned up manually", journalFile.getAbsolutePath()); if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, "Failed to remove temporary journal file " + journalFile.getAbsolutePath() + "; this file should be cleaned up manually"); } } final File tocFile = TocUtil.getTocFile(journalFile); if (!tocFile.delete() && tocFile.exists()) { logger.warn("Failed to remove temporary journal TOC file {}; this file should be cleaned up manually", tocFile.getAbsolutePath()); if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, "Failed to remove temporary journal TOC file " + tocFile.getAbsolutePath() + "; this file should be cleaned up manually"); } } } if (records == 0) { writerFile.delete(); logger.debug("Couldn't merge journals: No Records to merge"); return null; } else { final long nanos = System.nanoTime() - startNanos; final long millis = TimeUnit.MILLISECONDS.convert(nanos, TimeUnit.NANOSECONDS); logger.info("Successfully merged {} journal files ({} records) into single Provenance Log File {} in {} milliseconds", numAvailableFiles, records, suggestedMergeFile, millis); } return writerFile; } }
PersistentProvenanceRepository implements ProvenanceRepository { File mergeJournals(final List<File> journalFiles, final File suggestedMergeFile, final EventReporter eventReporter) throws IOException { logger.debug("Merging {} to {}", journalFiles, suggestedMergeFile); if (this.closed.get()) { logger.info("Provenance Repository has been closed; will not merge journal files to {}", suggestedMergeFile); return null; } if (journalFiles.isEmpty()) { logger.debug("Couldn't merge journals: Journal Files is empty; won't merge journals"); return null; } Collections.sort(journalFiles, new Comparator<File>() { @Override public int compare(final File o1, final File o2) { final String suffix1 = LuceneUtil.substringAfterLast(o1.getName(), "."); final String suffix2 = LuceneUtil.substringAfterLast(o2.getName(), "."); try { final int journalIndex1 = Integer.parseInt(suffix1); final int journalIndex2 = Integer.parseInt(suffix2); return Integer.compare(journalIndex1, journalIndex2); } catch (final NumberFormatException nfe) { return o1.getName().compareTo(o2.getName()); } } }); final List<File> availableFiles = filterUnavailableFiles(journalFiles); final int numAvailableFiles = availableFiles.size(); if (numAvailableFiles > 0) { if (suggestedMergeFile.exists()) { logger.warn("Merged Journal File {} already exists; however, all partial journal files also exist " + "so assuming that the merge did not finish. Repeating procedure in order to ensure consistency."); final DeleteIndexAction deleteAction = new DeleteIndexAction(this, indexConfig, getIndexManager()); try { deleteAction.execute(suggestedMergeFile); } catch (final Exception e) { logger.warn("Failed to delete records from Journal File {} from the index; this could potentially result in duplicates. Failure was due to {}", suggestedMergeFile, e.toString()); if (logger.isDebugEnabled()) { logger.warn("", e); } } if (!suggestedMergeFile.delete()) { logger.error("Failed to delete partially written Provenance Journal File {}. This may result in events from this journal " + "file not being able to be displayed. This file should be deleted manually.", suggestedMergeFile); } final File tocFile = TocUtil.getTocFile(suggestedMergeFile); if (tocFile.exists() && !tocFile.delete()) { logger.error("Failed to delete .toc file {}; this may result in not being able to read the Provenance Events from the {} Journal File. " + "This can be corrected by manually deleting the {} file", tocFile, suggestedMergeFile, tocFile); } } } else { logger.warn("Cannot merge journal files {} because they do not exist on disk", journalFiles); return null; } final long startNanos = System.nanoTime(); final List<RecordReader> readers = new ArrayList<>(); int records = 0; final boolean isCompress = configuration.isCompressOnRollover(); final File writerFile = isCompress ? new File(suggestedMergeFile.getParentFile(), suggestedMergeFile.getName() + ".gz") : suggestedMergeFile; try { for (final File journalFile : availableFiles) { try { readers.add(RecordReaders.newRecordReader(journalFile, null, Integer.MAX_VALUE)); } catch (final EOFException eof) { } catch (final IOException ioe) { logger.warn("Unable to merge {} with other Journal Files due to {}", journalFile, ioe.toString()); if (logger.isDebugEnabled()) { logger.warn("", ioe); } if (eventReporter != null) { eventReporter.reportEvent(Severity.ERROR, EVENT_CATEGORY, "Failed to merge Journal Files due to " + ioe.toString()); } } } final SortedMap<StandardProvenanceEventRecord, RecordReader> recordToReaderMap = new TreeMap<>(new Comparator<StandardProvenanceEventRecord>() { @Override public int compare(final StandardProvenanceEventRecord o1, final StandardProvenanceEventRecord o2) { return Long.compare(o1.getEventId(), o2.getEventId()); } }); long minEventId = 0L; long earliestTimestamp = System.currentTimeMillis(); for (final RecordReader reader : readers) { StandardProvenanceEventRecord record = null; try { record = reader.nextRecord(); } catch (final EOFException eof) { } catch (final Exception e) { logger.warn("Failed to generate Provenance Event Record from Journal due to " + e + "; it's " + "possible that the record wasn't completely written to the file. This journal will be " + "skipped."); if (logger.isDebugEnabled()) { logger.warn("", e); } if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, "Failed to read Provenance Event " + "Record from Journal due to " + e + "; it's possible that the record wasn't " + "completely written to the file. This journal will be skipped."); } } if (record == null) { continue; } if (record.getEventTime() < earliestTimestamp) { earliestTimestamp = record.getEventTime(); } if (record.getEventId() < minEventId) { minEventId = record.getEventId(); } recordToReaderMap.put(record, reader); } final RingBuffer<ProvenanceEventRecord> latestRecords = new RingBuffer<>(1000); try (final RecordWriter writer = RecordWriters.newSchemaRecordWriter(writerFile, idGenerator, configuration.isCompressOnRollover(), true)) { writer.writeHeader(minEventId); final IndexingAction indexingAction = createIndexingAction(); final File indexingDirectory = indexConfig.getWritableIndexDirectory(writerFile, earliestTimestamp); long maxId = 0L; final BlockingQueue<Tuple<StandardProvenanceEventRecord, Integer>> eventQueue = new LinkedBlockingQueue<>(100); final AtomicBoolean finishedAdding = new AtomicBoolean(false); final List<Future<?>> futures = new ArrayList<>(); final EventIndexWriter indexWriter = getIndexManager().borrowIndexWriter(indexingDirectory); try { final ExecutorService exec = Executors.newFixedThreadPool(configuration.getIndexThreadPoolSize(), new ThreadFactory() { @Override public Thread newThread(final Runnable r) { final Thread t = Executors.defaultThreadFactory().newThread(r); t.setName("Index Provenance Events"); return t; } }); final AtomicInteger indexingFailureCount = new AtomicInteger(0); try { for (int i = 0; i < configuration.getIndexThreadPoolSize(); i++) { final Callable<Object> callable = new Callable<Object>() { @Override public Object call() throws IOException { while (!eventQueue.isEmpty() || !finishedAdding.get()) { try { final Tuple<StandardProvenanceEventRecord, Integer> tuple; try { tuple = eventQueue.poll(10, TimeUnit.MILLISECONDS); } catch (final InterruptedException ie) { Thread.currentThread().interrupt(); continue; } if (tuple == null) { continue; } indexingAction.index(tuple.getKey(), indexWriter.getIndexWriter(), tuple.getValue()); } catch (final Throwable t) { logger.error("Failed to index Provenance Event for " + writerFile + " to " + indexingDirectory, t); if (indexingFailureCount.incrementAndGet() >= MAX_INDEXING_FAILURE_COUNT) { return null; } } } return null; } }; final Future<?> future = exec.submit(callable); futures.add(future); } boolean indexEvents = true; while (!recordToReaderMap.isEmpty()) { final Map.Entry<StandardProvenanceEventRecord, RecordReader> entry = recordToReaderMap.entrySet().iterator().next(); final StandardProvenanceEventRecord record = entry.getKey(); final RecordReader reader = entry.getValue(); writer.writeRecord(record); final int blockIndex = writer.getTocWriter().getCurrentBlockIndex(); boolean accepted = false; while (!accepted && indexEvents) { try { accepted = eventQueue.offer(new Tuple<>(record, blockIndex), 10, TimeUnit.MILLISECONDS); } catch (final InterruptedException ie) { Thread.currentThread().interrupt(); } if (!accepted && indexingFailureCount.get() >= MAX_INDEXING_FAILURE_COUNT) { indexEvents = false; eventQueue.clear(); final String warning = String.format("Indexing Provenance Events for %s has failed %s times. This exceeds the maximum threshold of %s failures, " + "so no more Provenance Events will be indexed for this Provenance file.", writerFile, indexingFailureCount.get(), MAX_INDEXING_FAILURE_COUNT); logger.warn(warning); if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, warning); } } } maxId = record.getEventId(); latestRecords.add(truncateAttributes(record)); records++; recordToReaderMap.remove(record); StandardProvenanceEventRecord nextRecord = null; try { nextRecord = reader.nextRecord(); } catch (final EOFException eof) { } catch (final Exception e) { logger.warn("Failed to generate Provenance Event Record from Journal due to " + e + "; it's possible that the record wasn't completely written to the file. " + "The remainder of this journal will be skipped."); if (logger.isDebugEnabled()) { logger.warn("", e); } if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, "Failed to read " + "Provenance Event Record from Journal due to " + e + "; it's possible " + "that the record wasn't completely written to the file. The remainder " + "of this journal will be skipped."); } } if (nextRecord != null) { recordToReaderMap.put(nextRecord, reader); } } } finally { finishedAdding.set(true); exec.shutdown(); } for (final Future<?> future : futures) { try { future.get(); } catch (final ExecutionException ee) { final Throwable t = ee.getCause(); if (t instanceof RuntimeException) { throw (RuntimeException) t; } throw new RuntimeException(t); } catch (final InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException("Thread interrupted"); } } } finally { getIndexManager().returnIndexWriter(indexWriter); } indexConfig.setMaxIdIndexed(maxId); } final RingBuffer<ProvenanceEventRecord> latestRecordBuffer = this.latestRecords; latestRecords.forEach(new ForEachEvaluator<ProvenanceEventRecord>() { @Override public boolean evaluate(final ProvenanceEventRecord event) { latestRecordBuffer.add(event); return true; } }); } finally { for (final RecordReader reader : readers) { try { reader.close(); } catch (final IOException ioe) { } } } for (final File journalFile : availableFiles) { if (!journalFile.delete() && journalFile.exists()) { logger.warn("Failed to remove temporary journal file {}; this file should be cleaned up manually", journalFile.getAbsolutePath()); if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, "Failed to remove temporary journal file " + journalFile.getAbsolutePath() + "; this file should be cleaned up manually"); } } final File tocFile = TocUtil.getTocFile(journalFile); if (!tocFile.delete() && tocFile.exists()) { logger.warn("Failed to remove temporary journal TOC file {}; this file should be cleaned up manually", tocFile.getAbsolutePath()); if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, "Failed to remove temporary journal TOC file " + tocFile.getAbsolutePath() + "; this file should be cleaned up manually"); } } } if (records == 0) { writerFile.delete(); logger.debug("Couldn't merge journals: No Records to merge"); return null; } else { final long nanos = System.nanoTime() - startNanos; final long millis = TimeUnit.MILLISECONDS.convert(nanos, TimeUnit.NANOSECONDS); logger.info("Successfully merged {} journal files ({} records) into single Provenance Log File {} in {} milliseconds", numAvailableFiles, records, suggestedMergeFile, millis); } return writerFile; } PersistentProvenanceRepository();  PersistentProvenanceRepository(final NiFiProperties nifiProperties);  PersistentProvenanceRepository(final RepositoryConfiguration configuration, final int rolloverCheckMillis); }
PersistentProvenanceRepository implements ProvenanceRepository { File mergeJournals(final List<File> journalFiles, final File suggestedMergeFile, final EventReporter eventReporter) throws IOException { logger.debug("Merging {} to {}", journalFiles, suggestedMergeFile); if (this.closed.get()) { logger.info("Provenance Repository has been closed; will not merge journal files to {}", suggestedMergeFile); return null; } if (journalFiles.isEmpty()) { logger.debug("Couldn't merge journals: Journal Files is empty; won't merge journals"); return null; } Collections.sort(journalFiles, new Comparator<File>() { @Override public int compare(final File o1, final File o2) { final String suffix1 = LuceneUtil.substringAfterLast(o1.getName(), "."); final String suffix2 = LuceneUtil.substringAfterLast(o2.getName(), "."); try { final int journalIndex1 = Integer.parseInt(suffix1); final int journalIndex2 = Integer.parseInt(suffix2); return Integer.compare(journalIndex1, journalIndex2); } catch (final NumberFormatException nfe) { return o1.getName().compareTo(o2.getName()); } } }); final List<File> availableFiles = filterUnavailableFiles(journalFiles); final int numAvailableFiles = availableFiles.size(); if (numAvailableFiles > 0) { if (suggestedMergeFile.exists()) { logger.warn("Merged Journal File {} already exists; however, all partial journal files also exist " + "so assuming that the merge did not finish. Repeating procedure in order to ensure consistency."); final DeleteIndexAction deleteAction = new DeleteIndexAction(this, indexConfig, getIndexManager()); try { deleteAction.execute(suggestedMergeFile); } catch (final Exception e) { logger.warn("Failed to delete records from Journal File {} from the index; this could potentially result in duplicates. Failure was due to {}", suggestedMergeFile, e.toString()); if (logger.isDebugEnabled()) { logger.warn("", e); } } if (!suggestedMergeFile.delete()) { logger.error("Failed to delete partially written Provenance Journal File {}. This may result in events from this journal " + "file not being able to be displayed. This file should be deleted manually.", suggestedMergeFile); } final File tocFile = TocUtil.getTocFile(suggestedMergeFile); if (tocFile.exists() && !tocFile.delete()) { logger.error("Failed to delete .toc file {}; this may result in not being able to read the Provenance Events from the {} Journal File. " + "This can be corrected by manually deleting the {} file", tocFile, suggestedMergeFile, tocFile); } } } else { logger.warn("Cannot merge journal files {} because they do not exist on disk", journalFiles); return null; } final long startNanos = System.nanoTime(); final List<RecordReader> readers = new ArrayList<>(); int records = 0; final boolean isCompress = configuration.isCompressOnRollover(); final File writerFile = isCompress ? new File(suggestedMergeFile.getParentFile(), suggestedMergeFile.getName() + ".gz") : suggestedMergeFile; try { for (final File journalFile : availableFiles) { try { readers.add(RecordReaders.newRecordReader(journalFile, null, Integer.MAX_VALUE)); } catch (final EOFException eof) { } catch (final IOException ioe) { logger.warn("Unable to merge {} with other Journal Files due to {}", journalFile, ioe.toString()); if (logger.isDebugEnabled()) { logger.warn("", ioe); } if (eventReporter != null) { eventReporter.reportEvent(Severity.ERROR, EVENT_CATEGORY, "Failed to merge Journal Files due to " + ioe.toString()); } } } final SortedMap<StandardProvenanceEventRecord, RecordReader> recordToReaderMap = new TreeMap<>(new Comparator<StandardProvenanceEventRecord>() { @Override public int compare(final StandardProvenanceEventRecord o1, final StandardProvenanceEventRecord o2) { return Long.compare(o1.getEventId(), o2.getEventId()); } }); long minEventId = 0L; long earliestTimestamp = System.currentTimeMillis(); for (final RecordReader reader : readers) { StandardProvenanceEventRecord record = null; try { record = reader.nextRecord(); } catch (final EOFException eof) { } catch (final Exception e) { logger.warn("Failed to generate Provenance Event Record from Journal due to " + e + "; it's " + "possible that the record wasn't completely written to the file. This journal will be " + "skipped."); if (logger.isDebugEnabled()) { logger.warn("", e); } if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, "Failed to read Provenance Event " + "Record from Journal due to " + e + "; it's possible that the record wasn't " + "completely written to the file. This journal will be skipped."); } } if (record == null) { continue; } if (record.getEventTime() < earliestTimestamp) { earliestTimestamp = record.getEventTime(); } if (record.getEventId() < minEventId) { minEventId = record.getEventId(); } recordToReaderMap.put(record, reader); } final RingBuffer<ProvenanceEventRecord> latestRecords = new RingBuffer<>(1000); try (final RecordWriter writer = RecordWriters.newSchemaRecordWriter(writerFile, idGenerator, configuration.isCompressOnRollover(), true)) { writer.writeHeader(minEventId); final IndexingAction indexingAction = createIndexingAction(); final File indexingDirectory = indexConfig.getWritableIndexDirectory(writerFile, earliestTimestamp); long maxId = 0L; final BlockingQueue<Tuple<StandardProvenanceEventRecord, Integer>> eventQueue = new LinkedBlockingQueue<>(100); final AtomicBoolean finishedAdding = new AtomicBoolean(false); final List<Future<?>> futures = new ArrayList<>(); final EventIndexWriter indexWriter = getIndexManager().borrowIndexWriter(indexingDirectory); try { final ExecutorService exec = Executors.newFixedThreadPool(configuration.getIndexThreadPoolSize(), new ThreadFactory() { @Override public Thread newThread(final Runnable r) { final Thread t = Executors.defaultThreadFactory().newThread(r); t.setName("Index Provenance Events"); return t; } }); final AtomicInteger indexingFailureCount = new AtomicInteger(0); try { for (int i = 0; i < configuration.getIndexThreadPoolSize(); i++) { final Callable<Object> callable = new Callable<Object>() { @Override public Object call() throws IOException { while (!eventQueue.isEmpty() || !finishedAdding.get()) { try { final Tuple<StandardProvenanceEventRecord, Integer> tuple; try { tuple = eventQueue.poll(10, TimeUnit.MILLISECONDS); } catch (final InterruptedException ie) { Thread.currentThread().interrupt(); continue; } if (tuple == null) { continue; } indexingAction.index(tuple.getKey(), indexWriter.getIndexWriter(), tuple.getValue()); } catch (final Throwable t) { logger.error("Failed to index Provenance Event for " + writerFile + " to " + indexingDirectory, t); if (indexingFailureCount.incrementAndGet() >= MAX_INDEXING_FAILURE_COUNT) { return null; } } } return null; } }; final Future<?> future = exec.submit(callable); futures.add(future); } boolean indexEvents = true; while (!recordToReaderMap.isEmpty()) { final Map.Entry<StandardProvenanceEventRecord, RecordReader> entry = recordToReaderMap.entrySet().iterator().next(); final StandardProvenanceEventRecord record = entry.getKey(); final RecordReader reader = entry.getValue(); writer.writeRecord(record); final int blockIndex = writer.getTocWriter().getCurrentBlockIndex(); boolean accepted = false; while (!accepted && indexEvents) { try { accepted = eventQueue.offer(new Tuple<>(record, blockIndex), 10, TimeUnit.MILLISECONDS); } catch (final InterruptedException ie) { Thread.currentThread().interrupt(); } if (!accepted && indexingFailureCount.get() >= MAX_INDEXING_FAILURE_COUNT) { indexEvents = false; eventQueue.clear(); final String warning = String.format("Indexing Provenance Events for %s has failed %s times. This exceeds the maximum threshold of %s failures, " + "so no more Provenance Events will be indexed for this Provenance file.", writerFile, indexingFailureCount.get(), MAX_INDEXING_FAILURE_COUNT); logger.warn(warning); if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, warning); } } } maxId = record.getEventId(); latestRecords.add(truncateAttributes(record)); records++; recordToReaderMap.remove(record); StandardProvenanceEventRecord nextRecord = null; try { nextRecord = reader.nextRecord(); } catch (final EOFException eof) { } catch (final Exception e) { logger.warn("Failed to generate Provenance Event Record from Journal due to " + e + "; it's possible that the record wasn't completely written to the file. " + "The remainder of this journal will be skipped."); if (logger.isDebugEnabled()) { logger.warn("", e); } if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, "Failed to read " + "Provenance Event Record from Journal due to " + e + "; it's possible " + "that the record wasn't completely written to the file. The remainder " + "of this journal will be skipped."); } } if (nextRecord != null) { recordToReaderMap.put(nextRecord, reader); } } } finally { finishedAdding.set(true); exec.shutdown(); } for (final Future<?> future : futures) { try { future.get(); } catch (final ExecutionException ee) { final Throwable t = ee.getCause(); if (t instanceof RuntimeException) { throw (RuntimeException) t; } throw new RuntimeException(t); } catch (final InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException("Thread interrupted"); } } } finally { getIndexManager().returnIndexWriter(indexWriter); } indexConfig.setMaxIdIndexed(maxId); } final RingBuffer<ProvenanceEventRecord> latestRecordBuffer = this.latestRecords; latestRecords.forEach(new ForEachEvaluator<ProvenanceEventRecord>() { @Override public boolean evaluate(final ProvenanceEventRecord event) { latestRecordBuffer.add(event); return true; } }); } finally { for (final RecordReader reader : readers) { try { reader.close(); } catch (final IOException ioe) { } } } for (final File journalFile : availableFiles) { if (!journalFile.delete() && journalFile.exists()) { logger.warn("Failed to remove temporary journal file {}; this file should be cleaned up manually", journalFile.getAbsolutePath()); if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, "Failed to remove temporary journal file " + journalFile.getAbsolutePath() + "; this file should be cleaned up manually"); } } final File tocFile = TocUtil.getTocFile(journalFile); if (!tocFile.delete() && tocFile.exists()) { logger.warn("Failed to remove temporary journal TOC file {}; this file should be cleaned up manually", tocFile.getAbsolutePath()); if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, "Failed to remove temporary journal TOC file " + tocFile.getAbsolutePath() + "; this file should be cleaned up manually"); } } } if (records == 0) { writerFile.delete(); logger.debug("Couldn't merge journals: No Records to merge"); return null; } else { final long nanos = System.nanoTime() - startNanos; final long millis = TimeUnit.MILLISECONDS.convert(nanos, TimeUnit.NANOSECONDS); logger.info("Successfully merged {} journal files ({} records) into single Provenance Log File {} in {} milliseconds", numAvailableFiles, records, suggestedMergeFile, millis); } return writerFile; } PersistentProvenanceRepository();  PersistentProvenanceRepository(final NiFiProperties nifiProperties);  PersistentProvenanceRepository(final RepositoryConfiguration configuration, final int rolloverCheckMillis); @Override void initialize(final EventReporter eventReporter, final Authorizer authorizer, final ProvenanceAuthorizableFactory resourceFactory,
        final IdentifierLookup idLookup); int getMaxAttributeCharacters(); @Override StandardProvenanceEventRecord.Builder eventBuilder(); @Override void registerEvent(final ProvenanceEventRecord event); @Override void registerEvents(final Iterable<ProvenanceEventRecord> events); boolean isAuthorized(final ProvenanceEventRecord event, final NiFiUser user); void authorize(final ProvenanceEventRecord event, final NiFiUser user); List<ProvenanceEventRecord> filterUnauthorizedEvents(final List<ProvenanceEventRecord> events, final NiFiUser user); Set<ProvenanceEventRecord> replaceUnauthorizedWithPlaceholders(final Set<ProvenanceEventRecord> events, final NiFiUser user); @Override List<ProvenanceEventRecord> getEvents(final long firstRecordId, final int maxRecords); @Override List<ProvenanceEventRecord> getEvents(final long firstRecordId, final int maxRecords, final NiFiUser user); RepositoryConfiguration getConfiguration(); @Override synchronized void close(); boolean isShutdownComplete(); long getSize(final List<File> logFiles, final long timeCutoff); void waitForRollover(); @Override List<SearchableField> getSearchableFields(); @Override List<SearchableField> getSearchableAttributes(); @Override QuerySubmission submitQuery(final Query query, final NiFiUser user); @Override ComputeLineageSubmission submitLineageComputation(final long eventId, final NiFiUser user); @Override AsyncLineageSubmission submitLineageComputation(final String flowFileUuid, final NiFiUser user); @Override AsyncLineageSubmission submitExpandChildren(final long eventId, final NiFiUser user); @Override AsyncLineageSubmission submitExpandParents(final long eventId, final NiFiUser user); @Override AsyncLineageSubmission retrieveLineageSubmission(final String lineageIdentifier, final NiFiUser user); @Override QuerySubmission retrieveQuerySubmission(final String queryIdentifier, final NiFiUser user); @Override ProvenanceEventRecord getEvent(final long id); @Override ProvenanceEventRecord getEvent(final long id, final NiFiUser user); @Override ProvenanceEventRepository getProvenanceEventRepository(); Collection<Path> getAllLogFiles(); @Override Long getMaxEventId(); }
PersistentProvenanceRepository implements ProvenanceRepository { File mergeJournals(final List<File> journalFiles, final File suggestedMergeFile, final EventReporter eventReporter) throws IOException { logger.debug("Merging {} to {}", journalFiles, suggestedMergeFile); if (this.closed.get()) { logger.info("Provenance Repository has been closed; will not merge journal files to {}", suggestedMergeFile); return null; } if (journalFiles.isEmpty()) { logger.debug("Couldn't merge journals: Journal Files is empty; won't merge journals"); return null; } Collections.sort(journalFiles, new Comparator<File>() { @Override public int compare(final File o1, final File o2) { final String suffix1 = LuceneUtil.substringAfterLast(o1.getName(), "."); final String suffix2 = LuceneUtil.substringAfterLast(o2.getName(), "."); try { final int journalIndex1 = Integer.parseInt(suffix1); final int journalIndex2 = Integer.parseInt(suffix2); return Integer.compare(journalIndex1, journalIndex2); } catch (final NumberFormatException nfe) { return o1.getName().compareTo(o2.getName()); } } }); final List<File> availableFiles = filterUnavailableFiles(journalFiles); final int numAvailableFiles = availableFiles.size(); if (numAvailableFiles > 0) { if (suggestedMergeFile.exists()) { logger.warn("Merged Journal File {} already exists; however, all partial journal files also exist " + "so assuming that the merge did not finish. Repeating procedure in order to ensure consistency."); final DeleteIndexAction deleteAction = new DeleteIndexAction(this, indexConfig, getIndexManager()); try { deleteAction.execute(suggestedMergeFile); } catch (final Exception e) { logger.warn("Failed to delete records from Journal File {} from the index; this could potentially result in duplicates. Failure was due to {}", suggestedMergeFile, e.toString()); if (logger.isDebugEnabled()) { logger.warn("", e); } } if (!suggestedMergeFile.delete()) { logger.error("Failed to delete partially written Provenance Journal File {}. This may result in events from this journal " + "file not being able to be displayed. This file should be deleted manually.", suggestedMergeFile); } final File tocFile = TocUtil.getTocFile(suggestedMergeFile); if (tocFile.exists() && !tocFile.delete()) { logger.error("Failed to delete .toc file {}; this may result in not being able to read the Provenance Events from the {} Journal File. " + "This can be corrected by manually deleting the {} file", tocFile, suggestedMergeFile, tocFile); } } } else { logger.warn("Cannot merge journal files {} because they do not exist on disk", journalFiles); return null; } final long startNanos = System.nanoTime(); final List<RecordReader> readers = new ArrayList<>(); int records = 0; final boolean isCompress = configuration.isCompressOnRollover(); final File writerFile = isCompress ? new File(suggestedMergeFile.getParentFile(), suggestedMergeFile.getName() + ".gz") : suggestedMergeFile; try { for (final File journalFile : availableFiles) { try { readers.add(RecordReaders.newRecordReader(journalFile, null, Integer.MAX_VALUE)); } catch (final EOFException eof) { } catch (final IOException ioe) { logger.warn("Unable to merge {} with other Journal Files due to {}", journalFile, ioe.toString()); if (logger.isDebugEnabled()) { logger.warn("", ioe); } if (eventReporter != null) { eventReporter.reportEvent(Severity.ERROR, EVENT_CATEGORY, "Failed to merge Journal Files due to " + ioe.toString()); } } } final SortedMap<StandardProvenanceEventRecord, RecordReader> recordToReaderMap = new TreeMap<>(new Comparator<StandardProvenanceEventRecord>() { @Override public int compare(final StandardProvenanceEventRecord o1, final StandardProvenanceEventRecord o2) { return Long.compare(o1.getEventId(), o2.getEventId()); } }); long minEventId = 0L; long earliestTimestamp = System.currentTimeMillis(); for (final RecordReader reader : readers) { StandardProvenanceEventRecord record = null; try { record = reader.nextRecord(); } catch (final EOFException eof) { } catch (final Exception e) { logger.warn("Failed to generate Provenance Event Record from Journal due to " + e + "; it's " + "possible that the record wasn't completely written to the file. This journal will be " + "skipped."); if (logger.isDebugEnabled()) { logger.warn("", e); } if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, "Failed to read Provenance Event " + "Record from Journal due to " + e + "; it's possible that the record wasn't " + "completely written to the file. This journal will be skipped."); } } if (record == null) { continue; } if (record.getEventTime() < earliestTimestamp) { earliestTimestamp = record.getEventTime(); } if (record.getEventId() < minEventId) { minEventId = record.getEventId(); } recordToReaderMap.put(record, reader); } final RingBuffer<ProvenanceEventRecord> latestRecords = new RingBuffer<>(1000); try (final RecordWriter writer = RecordWriters.newSchemaRecordWriter(writerFile, idGenerator, configuration.isCompressOnRollover(), true)) { writer.writeHeader(minEventId); final IndexingAction indexingAction = createIndexingAction(); final File indexingDirectory = indexConfig.getWritableIndexDirectory(writerFile, earliestTimestamp); long maxId = 0L; final BlockingQueue<Tuple<StandardProvenanceEventRecord, Integer>> eventQueue = new LinkedBlockingQueue<>(100); final AtomicBoolean finishedAdding = new AtomicBoolean(false); final List<Future<?>> futures = new ArrayList<>(); final EventIndexWriter indexWriter = getIndexManager().borrowIndexWriter(indexingDirectory); try { final ExecutorService exec = Executors.newFixedThreadPool(configuration.getIndexThreadPoolSize(), new ThreadFactory() { @Override public Thread newThread(final Runnable r) { final Thread t = Executors.defaultThreadFactory().newThread(r); t.setName("Index Provenance Events"); return t; } }); final AtomicInteger indexingFailureCount = new AtomicInteger(0); try { for (int i = 0; i < configuration.getIndexThreadPoolSize(); i++) { final Callable<Object> callable = new Callable<Object>() { @Override public Object call() throws IOException { while (!eventQueue.isEmpty() || !finishedAdding.get()) { try { final Tuple<StandardProvenanceEventRecord, Integer> tuple; try { tuple = eventQueue.poll(10, TimeUnit.MILLISECONDS); } catch (final InterruptedException ie) { Thread.currentThread().interrupt(); continue; } if (tuple == null) { continue; } indexingAction.index(tuple.getKey(), indexWriter.getIndexWriter(), tuple.getValue()); } catch (final Throwable t) { logger.error("Failed to index Provenance Event for " + writerFile + " to " + indexingDirectory, t); if (indexingFailureCount.incrementAndGet() >= MAX_INDEXING_FAILURE_COUNT) { return null; } } } return null; } }; final Future<?> future = exec.submit(callable); futures.add(future); } boolean indexEvents = true; while (!recordToReaderMap.isEmpty()) { final Map.Entry<StandardProvenanceEventRecord, RecordReader> entry = recordToReaderMap.entrySet().iterator().next(); final StandardProvenanceEventRecord record = entry.getKey(); final RecordReader reader = entry.getValue(); writer.writeRecord(record); final int blockIndex = writer.getTocWriter().getCurrentBlockIndex(); boolean accepted = false; while (!accepted && indexEvents) { try { accepted = eventQueue.offer(new Tuple<>(record, blockIndex), 10, TimeUnit.MILLISECONDS); } catch (final InterruptedException ie) { Thread.currentThread().interrupt(); } if (!accepted && indexingFailureCount.get() >= MAX_INDEXING_FAILURE_COUNT) { indexEvents = false; eventQueue.clear(); final String warning = String.format("Indexing Provenance Events for %s has failed %s times. This exceeds the maximum threshold of %s failures, " + "so no more Provenance Events will be indexed for this Provenance file.", writerFile, indexingFailureCount.get(), MAX_INDEXING_FAILURE_COUNT); logger.warn(warning); if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, warning); } } } maxId = record.getEventId(); latestRecords.add(truncateAttributes(record)); records++; recordToReaderMap.remove(record); StandardProvenanceEventRecord nextRecord = null; try { nextRecord = reader.nextRecord(); } catch (final EOFException eof) { } catch (final Exception e) { logger.warn("Failed to generate Provenance Event Record from Journal due to " + e + "; it's possible that the record wasn't completely written to the file. " + "The remainder of this journal will be skipped."); if (logger.isDebugEnabled()) { logger.warn("", e); } if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, "Failed to read " + "Provenance Event Record from Journal due to " + e + "; it's possible " + "that the record wasn't completely written to the file. The remainder " + "of this journal will be skipped."); } } if (nextRecord != null) { recordToReaderMap.put(nextRecord, reader); } } } finally { finishedAdding.set(true); exec.shutdown(); } for (final Future<?> future : futures) { try { future.get(); } catch (final ExecutionException ee) { final Throwable t = ee.getCause(); if (t instanceof RuntimeException) { throw (RuntimeException) t; } throw new RuntimeException(t); } catch (final InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException("Thread interrupted"); } } } finally { getIndexManager().returnIndexWriter(indexWriter); } indexConfig.setMaxIdIndexed(maxId); } final RingBuffer<ProvenanceEventRecord> latestRecordBuffer = this.latestRecords; latestRecords.forEach(new ForEachEvaluator<ProvenanceEventRecord>() { @Override public boolean evaluate(final ProvenanceEventRecord event) { latestRecordBuffer.add(event); return true; } }); } finally { for (final RecordReader reader : readers) { try { reader.close(); } catch (final IOException ioe) { } } } for (final File journalFile : availableFiles) { if (!journalFile.delete() && journalFile.exists()) { logger.warn("Failed to remove temporary journal file {}; this file should be cleaned up manually", journalFile.getAbsolutePath()); if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, "Failed to remove temporary journal file " + journalFile.getAbsolutePath() + "; this file should be cleaned up manually"); } } final File tocFile = TocUtil.getTocFile(journalFile); if (!tocFile.delete() && tocFile.exists()) { logger.warn("Failed to remove temporary journal TOC file {}; this file should be cleaned up manually", tocFile.getAbsolutePath()); if (eventReporter != null) { eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, "Failed to remove temporary journal TOC file " + tocFile.getAbsolutePath() + "; this file should be cleaned up manually"); } } } if (records == 0) { writerFile.delete(); logger.debug("Couldn't merge journals: No Records to merge"); return null; } else { final long nanos = System.nanoTime() - startNanos; final long millis = TimeUnit.MILLISECONDS.convert(nanos, TimeUnit.NANOSECONDS); logger.info("Successfully merged {} journal files ({} records) into single Provenance Log File {} in {} milliseconds", numAvailableFiles, records, suggestedMergeFile, millis); } return writerFile; } PersistentProvenanceRepository();  PersistentProvenanceRepository(final NiFiProperties nifiProperties);  PersistentProvenanceRepository(final RepositoryConfiguration configuration, final int rolloverCheckMillis); @Override void initialize(final EventReporter eventReporter, final Authorizer authorizer, final ProvenanceAuthorizableFactory resourceFactory,
        final IdentifierLookup idLookup); int getMaxAttributeCharacters(); @Override StandardProvenanceEventRecord.Builder eventBuilder(); @Override void registerEvent(final ProvenanceEventRecord event); @Override void registerEvents(final Iterable<ProvenanceEventRecord> events); boolean isAuthorized(final ProvenanceEventRecord event, final NiFiUser user); void authorize(final ProvenanceEventRecord event, final NiFiUser user); List<ProvenanceEventRecord> filterUnauthorizedEvents(final List<ProvenanceEventRecord> events, final NiFiUser user); Set<ProvenanceEventRecord> replaceUnauthorizedWithPlaceholders(final Set<ProvenanceEventRecord> events, final NiFiUser user); @Override List<ProvenanceEventRecord> getEvents(final long firstRecordId, final int maxRecords); @Override List<ProvenanceEventRecord> getEvents(final long firstRecordId, final int maxRecords, final NiFiUser user); RepositoryConfiguration getConfiguration(); @Override synchronized void close(); boolean isShutdownComplete(); long getSize(final List<File> logFiles, final long timeCutoff); void waitForRollover(); @Override List<SearchableField> getSearchableFields(); @Override List<SearchableField> getSearchableAttributes(); @Override QuerySubmission submitQuery(final Query query, final NiFiUser user); @Override ComputeLineageSubmission submitLineageComputation(final long eventId, final NiFiUser user); @Override AsyncLineageSubmission submitLineageComputation(final String flowFileUuid, final NiFiUser user); @Override AsyncLineageSubmission submitExpandChildren(final long eventId, final NiFiUser user); @Override AsyncLineageSubmission submitExpandParents(final long eventId, final NiFiUser user); @Override AsyncLineageSubmission retrieveLineageSubmission(final String lineageIdentifier, final NiFiUser user); @Override QuerySubmission retrieveQuerySubmission(final String queryIdentifier, final NiFiUser user); @Override ProvenanceEventRecord getEvent(final long id); @Override ProvenanceEventRecord getEvent(final long id, final NiFiUser user); @Override ProvenanceEventRepository getProvenanceEventRepository(); Collection<Path> getAllLogFiles(); @Override Long getMaxEventId(); static final String EVENT_CATEGORY; static final Pattern NUMBER_PATTERN; static final Pattern INDEX_PATTERN; static final Pattern LOG_FILENAME_PATTERN; static final int MAX_UNDELETED_QUERY_RESULTS; static final int MAX_INDEXING_FAILURE_COUNT; static final int MAX_JOURNAL_ROLLOVER_RETRIES; }