You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by oz...@apache.org on 2017/02/22 17:42:37 UTC
[07/11] nifi git commit: NIFI-3356: Initial implementation of
writeahead provenance repository - The idea behind NIFI-3356 was to improve
the efficiency and throughput of the Provenance Repository,
as it is often the bottleneck. While testing the newly d
http://git-wip-us.apache.org/repos/asf/nifi/blob/96ed405d/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/index/lucene/LuceneEventIndex.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/index/lucene/LuceneEventIndex.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/index/lucene/LuceneEventIndex.java
new file mode 100644
index 0000000..a583403
--- /dev/null
+++ b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/index/lucene/LuceneEventIndex.java
@@ -0,0 +1,737 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.provenance.index.lucene;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Date;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.lucene.document.Document;
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.Term;
+import org.apache.lucene.search.BooleanClause.Occur;
+import org.apache.lucene.search.BooleanQuery;
+import org.apache.lucene.search.TermQuery;
+import org.apache.nifi.authorization.AccessDeniedException;
+import org.apache.nifi.authorization.user.NiFiUser;
+import org.apache.nifi.events.EventReporter;
+import org.apache.nifi.provenance.AsyncLineageSubmission;
+import org.apache.nifi.provenance.AsyncQuerySubmission;
+import org.apache.nifi.provenance.ProvenanceEventRecord;
+import org.apache.nifi.provenance.RepositoryConfiguration;
+import org.apache.nifi.provenance.SearchableFields;
+import org.apache.nifi.provenance.StandardLineageResult;
+import org.apache.nifi.provenance.StandardQueryResult;
+import org.apache.nifi.provenance.authorization.EventAuthorizer;
+import org.apache.nifi.provenance.authorization.EventTransformer;
+import org.apache.nifi.provenance.index.EventIndex;
+import org.apache.nifi.provenance.index.EventIndexSearcher;
+import org.apache.nifi.provenance.index.EventIndexWriter;
+import org.apache.nifi.provenance.lineage.ComputeLineageSubmission;
+import org.apache.nifi.provenance.lineage.LineageComputationType;
+import org.apache.nifi.provenance.lucene.IndexManager;
+import org.apache.nifi.provenance.lucene.LuceneUtil;
+import org.apache.nifi.provenance.search.Query;
+import org.apache.nifi.provenance.search.QuerySubmission;
+import org.apache.nifi.provenance.serialization.StorageSummary;
+import org.apache.nifi.provenance.store.EventStore;
+import org.apache.nifi.provenance.util.DirectoryUtils;
+import org.apache.nifi.provenance.util.NamedThreadFactory;
+import org.apache.nifi.reporting.Severity;
+import org.apache.nifi.util.file.FileUtils;
+import org.apache.nifi.util.timebuffer.LongEntityAccess;
+import org.apache.nifi.util.timebuffer.TimedBuffer;
+import org.apache.nifi.util.timebuffer.TimestampedLong;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class LuceneEventIndex implements EventIndex {
+ private static final Logger logger = LoggerFactory.getLogger(LuceneEventIndex.class);
+ private static final String EVENT_CATEGORY = "Provenance Repository";
+
+ public static final int MAX_UNDELETED_QUERY_RESULTS = 10;
+ public static final int MAX_DELETE_INDEX_WAIT_SECONDS = 30;
+ public static final int MAX_LINEAGE_NODES = 1000;
+ public static final int MAX_INDEX_THREADS = 100;
+
+ private final ConcurrentMap<String, AsyncQuerySubmission> querySubmissionMap = new ConcurrentHashMap<>();
+ private final ConcurrentMap<String, AsyncLineageSubmission> lineageSubmissionMap = new ConcurrentHashMap<>();
+ private final BlockingQueue<StoredDocument> documentQueue = new LinkedBlockingQueue<>(1000);
+ private final List<EventIndexTask> indexTasks = Collections.synchronizedList(new ArrayList<>());
+ private final ExecutorService queryExecutor;
+ private final ExecutorService indexExecutor;
+ private final RepositoryConfiguration config;
+ private final IndexManager indexManager;
+ private final ConvertEventToLuceneDocument eventConverter;
+ private final IndexDirectoryManager directoryManager;
+ private volatile boolean closed = false;
+
+ private final TimedBuffer<TimestampedLong> queuePauseNanos = new TimedBuffer<>(TimeUnit.SECONDS, 300, new LongEntityAccess());
+ private final TimedBuffer<TimestampedLong> eventsIndexed = new TimedBuffer<>(TimeUnit.SECONDS, 300, new LongEntityAccess());
+ private final AtomicLong eventCount = new AtomicLong(0L);
+ private final EventReporter eventReporter;
+
+ private final List<CachedQuery> cachedQueries = new ArrayList<>();
+
+ private ScheduledExecutorService maintenanceExecutor; // effectively final
+ private ScheduledExecutorService cacheWarmerExecutor;
+ private EventStore eventStore;
+
+ public LuceneEventIndex(final RepositoryConfiguration config, final IndexManager indexManager, final EventReporter eventReporter) {
+ this(config, indexManager, EventIndexTask.DEFAULT_MAX_EVENTS_PER_COMMIT, eventReporter);
+ }
+
+ public LuceneEventIndex(final RepositoryConfiguration config, final IndexManager indexManager, final int maxEventsPerCommit, final EventReporter eventReporter) {
+ this.eventReporter = eventReporter;
+ queryExecutor = Executors.newFixedThreadPool(config.getQueryThreadPoolSize(), new NamedThreadFactory("Provenance Query"));
+ indexExecutor = Executors.newFixedThreadPool(config.getIndexThreadPoolSize(), new NamedThreadFactory("Index Provenance Events"));
+ cacheWarmerExecutor = Executors.newScheduledThreadPool(config.getStorageDirectories().size(), new NamedThreadFactory("Warm Lucene Index", true));
+ directoryManager = new IndexDirectoryManager(config);
+
+ // Limit number of indexing threads to 100. When we restore the repository on restart,
+ // we have to re-index up to MAX_THREADS * MAX_DOCUMENTS_PER_THREADS events prior to
+ // the last event that the index holds. This is done because we could have that many
+ // events 'in flight', waiting to be indexed when the last index writer was committed,
+ // so even though the index says the largest event ID is 1,000,000 for instance, Event
+ // with ID 999,999 may still not have been indexed because another thread was in the
+ // process of writing the event to the index.
+ final int configuredIndexPoolSize = config.getIndexThreadPoolSize();
+ final int numIndexThreads;
+ if (configuredIndexPoolSize > MAX_INDEX_THREADS) {
+ logger.warn("The Provenance Repository is configured to perform indexing of events using {} threads. This number exceeds the maximum allowable number of threads, which is {}. "
+ + "Will proceed using {} threads. This value is limited because the performance of indexing will decrease and startup times will increase when setting this value too high.",
+ configuredIndexPoolSize, MAX_INDEX_THREADS, MAX_INDEX_THREADS);
+ numIndexThreads = MAX_INDEX_THREADS;
+ } else {
+ numIndexThreads = configuredIndexPoolSize;
+ }
+
+ for (int i = 0; i < numIndexThreads; i++) {
+ final EventIndexTask task = new EventIndexTask(documentQueue, config, indexManager, directoryManager, maxEventsPerCommit, eventReporter);
+ indexTasks.add(task);
+ indexExecutor.submit(task);
+ }
+ this.config = config;
+ this.indexManager = indexManager;
+ this.eventConverter = new ConvertEventToLuceneDocument(config.getSearchableFields(), config.getSearchableAttributes());
+ }
+
+ @Override
+ public void initialize(final EventStore eventStore) {
+ this.eventStore = eventStore;
+ directoryManager.initialize();
+
+ maintenanceExecutor = Executors.newScheduledThreadPool(1, new NamedThreadFactory("Provenance Repository Maintenance"));
+ maintenanceExecutor.scheduleWithFixedDelay(() -> performMaintenance(), 1, 1, TimeUnit.MINUTES);
+ maintenanceExecutor.scheduleWithFixedDelay(new RemoveExpiredQueryResults(), 30, 30, TimeUnit.SECONDS);
+
+ cachedQueries.add(new LatestEventsQuery());
+ cachedQueries.add(new LatestEventsPerProcessorQuery());
+
+ final Optional<Integer> warmCacheMinutesOption = config.getWarmCacheFrequencyMinutes();
+ if (warmCacheMinutesOption.isPresent() && warmCacheMinutesOption.get() > 0) {
+ for (final File storageDir : config.getStorageDirectories().values()) {
+ final int minutes = warmCacheMinutesOption.get();
+ cacheWarmerExecutor.scheduleWithFixedDelay(new LuceneCacheWarmer(storageDir, indexManager), 1, minutes, TimeUnit.MINUTES);
+ }
+ }
+ }
+
+ @Override
+ public long getMinimumEventIdToReindex(final String partitionName) {
+ return Math.max(0, getMaxEventId(partitionName) - EventIndexTask.MAX_DOCUMENTS_PER_THREAD * LuceneEventIndex.MAX_INDEX_THREADS);
+ }
+
+ protected IndexDirectoryManager getDirectoryManager() {
+ return directoryManager;
+ }
+
+ @Override
+ public void close() throws IOException {
+ closed = true;
+ queryExecutor.shutdownNow();
+ indexExecutor.shutdown();
+ cacheWarmerExecutor.shutdown();
+
+ if (maintenanceExecutor != null) {
+ maintenanceExecutor.shutdown();
+ }
+
+ for (final EventIndexTask task : indexTasks) {
+ task.shutdown();
+ }
+ }
+
+ long getMaxEventId(final String partitionName) {
+ final List<File> allDirectories = getDirectoryManager().getDirectories(0L, Long.MAX_VALUE, partitionName);
+ if (allDirectories.isEmpty()) {
+ return -1L;
+ }
+
+ Collections.sort(allDirectories, DirectoryUtils.NEWEST_INDEX_FIRST);
+
+ for (final File directory : allDirectories) {
+ final EventIndexSearcher searcher;
+ try {
+ searcher = indexManager.borrowIndexSearcher(directory);
+ } catch (final IOException ioe) {
+ logger.warn("Unable to read from Index Directory {}. Will assume that the index is incomplete and not consider this index when determining max event ID", directory);
+ continue;
+ }
+
+ try {
+ final IndexReader reader = searcher.getIndexSearcher().getIndexReader();
+ final int maxDocId = reader.maxDoc() - 1;
+ final Document document = reader.document(maxDocId);
+ final long eventId = document.getField(SearchableFields.Identifier.getSearchableFieldName()).numericValue().longValue();
+ logger.info("Determined that Max Event ID indexed for Partition {} is approximately {} based on index {}", partitionName, eventId, directory);
+ return eventId;
+ } catch (final IOException ioe) {
+ logger.warn("Unable to search Index Directory {}. Will assume that the index is incomplete and not consider this index when determining max event ID", directory, ioe);
+ } finally {
+ indexManager.returnIndexSearcher(searcher);
+ }
+ }
+
+ return -1L;
+ }
+
+ @Override
+ public void reindexEvents(final Map<ProvenanceEventRecord, StorageSummary> events) {
+ final EventIndexTask indexTask = new EventIndexTask(documentQueue, config, indexManager, directoryManager, EventIndexTask.DEFAULT_MAX_EVENTS_PER_COMMIT, eventReporter);
+
+ File lastIndexDir = null;
+ long lastEventTime = -2L;
+
+ final List<IndexableDocument> indexableDocs = new ArrayList<>(events.size());
+ for (final Map.Entry<ProvenanceEventRecord, StorageSummary> entry : events.entrySet()) {
+ final ProvenanceEventRecord event = entry.getKey();
+ final StorageSummary summary = entry.getValue();
+
+ for (final CachedQuery cachedQuery : cachedQueries) {
+ cachedQuery.update(event, summary);
+ }
+
+ final Document document = eventConverter.convert(event, summary);
+ if (document == null) {
+ logger.debug("Received Provenance Event {} to index but it contained no information that should be indexed, so skipping it", event);
+ } else {
+ final File indexDir;
+ if (event.getEventTime() == lastEventTime) {
+ indexDir = lastIndexDir;
+ } else {
+ final List<File> files = getDirectoryManager().getDirectories(event.getEventTime(), null);
+ indexDir = files.isEmpty() ? null : files.get(0);
+ lastIndexDir = indexDir;
+ }
+
+ final IndexableDocument doc = new IndexableDocument(document, summary, indexDir);
+ indexableDocs.add(doc);
+ }
+ }
+
+ try {
+ indexTask.reIndex(indexableDocs, CommitPreference.PREVENT_COMMIT);
+ } catch (final IOException ioe) {
+ logger.error("Failed to reindex some Provenance Events", ioe);
+ eventReporter.reportEvent(Severity.ERROR, EVENT_CATEGORY, "Failed to re-index some Provenance Events. "
+ + "Some Provenance Events may not be available for querying. See logs for more information.");
+ }
+ }
+
+ @Override
+ public void commitChanges(final String partitionName) throws IOException {
+ final Optional<File> indexDir = directoryManager.getActiveIndexDirectory(partitionName);
+ if (indexDir.isPresent()) {
+ final EventIndexWriter eventIndexWriter = indexManager.borrowIndexWriter(indexDir.get());
+ try {
+ eventIndexWriter.commit();
+ } finally {
+ indexManager.returnIndexWriter(eventIndexWriter, false, false);
+ }
+ }
+ }
+
+ protected void addEvent(final ProvenanceEventRecord event, final StorageSummary location) {
+ for (final CachedQuery cachedQuery : cachedQueries) {
+ cachedQuery.update(event, location);
+ }
+
+ final Document document = eventConverter.convert(event, location);
+ if (document == null) {
+ logger.debug("Received Provenance Event {} to index but it contained no information that should be indexed, so skipping it", event);
+ } else {
+ final StoredDocument doc = new StoredDocument(document, location);
+ boolean added = false;
+ while (!added && !closed) {
+
+ added = documentQueue.offer(doc);
+ if (!added) {
+ final long start = System.nanoTime();
+ try {
+ added = documentQueue.offer(doc, 1, TimeUnit.SECONDS);
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ logger.warn("Interrupted while attempting to enqueue Provenance Event for indexing; this event will not be indexed");
+ return;
+ }
+ final long nanos = System.nanoTime() - start;
+ queuePauseNanos.add(new TimestampedLong(nanos));
+ }
+
+ if (added) {
+ final long totalEventCount = eventCount.incrementAndGet();
+ if (totalEventCount % 1_000_000 == 0 && logger.isDebugEnabled()) {
+ incrementAndReportStats();
+ }
+ }
+ }
+ }
+ }
+
+ private void incrementAndReportStats() {
+ final long fiveMinutesAgo = System.currentTimeMillis() - TimeUnit.MINUTES.toMillis(5);
+ final TimestampedLong nanosLastFive = queuePauseNanos.getAggregateValue(fiveMinutesAgo);
+ if (nanosLastFive == null) {
+ return;
+ }
+
+ final TimestampedLong eventsLast5 = eventsIndexed.getAggregateValue(fiveMinutesAgo);
+ if (eventsLast5 == null) {
+ return;
+ }
+
+ final long numEventsLast5 = eventsLast5.getValue();
+
+ final long millis = TimeUnit.NANOSECONDS.toMillis(nanosLastFive.getValue());
+ logger.debug("In the last 5 minutes, have spent {} CPU-millis waiting to enqueue events for indexing and have indexed {} events ({} since NiFi started)",
+ millis, numEventsLast5, eventCount.get());
+ }
+
+ @Override
+ public void addEvents(final Map<ProvenanceEventRecord, StorageSummary> events) {
+ eventsIndexed.add(new TimestampedLong((long) events.size()));
+
+ for (final Map.Entry<ProvenanceEventRecord, StorageSummary> entry : events.entrySet()) {
+ addEvent(entry.getKey(), entry.getValue());
+ }
+ }
+
+
+ @Override
+ public ComputeLineageSubmission submitLineageComputation(final long eventId, final NiFiUser user, final EventAuthorizer eventAuthorizer) {
+ final Optional<ProvenanceEventRecord> eventOption;
+ try {
+ eventOption = eventStore.getEvent(eventId);
+ } catch (final Exception e) {
+ logger.error("Failed to retrieve Provenance Event with ID " + eventId + " to calculate data lineage due to: " + e, e);
+ final AsyncLineageSubmission result = new AsyncLineageSubmission(LineageComputationType.FLOWFILE_LINEAGE, eventId, Collections.<String> emptySet(), 1, user.getIdentity());
+ result.getResult().setError("Failed to retrieve Provenance Event with ID " + eventId + ". See logs for more information.");
+ return result;
+ }
+
+ if (!eventOption.isPresent()) {
+ final AsyncLineageSubmission result = new AsyncLineageSubmission(LineageComputationType.FLOWFILE_LINEAGE, eventId, Collections.<String> emptySet(), 1, user.getIdentity());
+ result.getResult().setError("Could not find Provenance Event with ID " + eventId);
+ lineageSubmissionMap.put(result.getLineageIdentifier(), result);
+ return result;
+ }
+
+ final ProvenanceEventRecord event = eventOption.get();
+ return submitLineageComputation(Collections.singleton(event.getFlowFileUuid()), user, eventAuthorizer, LineageComputationType.FLOWFILE_LINEAGE,
+ eventId, event.getLineageStartDate(), Long.MAX_VALUE);
+ }
+
+
+ private ComputeLineageSubmission submitLineageComputation(final Collection<String> flowFileUuids, final NiFiUser user, final EventAuthorizer eventAuthorizer,
+ final LineageComputationType computationType, final Long eventId, final long startTimestamp, final long endTimestamp) {
+
+ final List<File> indexDirs = directoryManager.getDirectories(startTimestamp, endTimestamp);
+ final AsyncLineageSubmission submission = new AsyncLineageSubmission(computationType, eventId, flowFileUuids, indexDirs.size(), user.getIdentity());
+ lineageSubmissionMap.put(submission.getLineageIdentifier(), submission);
+
+ final BooleanQuery lineageQuery = buildLineageQuery(flowFileUuids);
+ final List<File> indexDirectories = directoryManager.getDirectories(startTimestamp, endTimestamp);
+ if (indexDirectories.isEmpty()) {
+ submission.getResult().update(Collections.emptyList(), 0L);
+ } else {
+ Collections.sort(indexDirectories, DirectoryUtils.OLDEST_INDEX_FIRST);
+
+ for (final File indexDir : indexDirectories) {
+ queryExecutor.submit(new QueryTask(lineageQuery, submission.getResult(), MAX_LINEAGE_NODES, indexManager, indexDir,
+ eventStore, eventAuthorizer, EventTransformer.PLACEHOLDER_TRANSFORMER));
+ }
+ }
+
+ // Some computations will complete very quickly. In this case, we don't want to wait
+ // for the client to submit a second query to obtain the result. Instead, we want to just
+ // wait some short period of time for the computation to complete before returning the submission.
+ try {
+ submission.getResult().awaitCompletion(500, TimeUnit.MILLISECONDS);
+ } catch (final InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ }
+
+ return submission;
+ }
+
+ private BooleanQuery buildLineageQuery(final Collection<String> flowFileUuids) {
+ // Create a query for all Events related to the FlowFiles of interest. We do this by adding all ID's as
+ // "SHOULD" clauses and then setting the minimum required to 1.
+ final BooleanQuery lineageQuery;
+ if (flowFileUuids == null || flowFileUuids.isEmpty()) {
+ lineageQuery = null;
+ } else {
+ lineageQuery = new BooleanQuery();
+ for (final String flowFileUuid : flowFileUuids) {
+ lineageQuery.add(new TermQuery(new Term(SearchableFields.FlowFileUUID.getSearchableFieldName(), flowFileUuid)), Occur.SHOULD);
+ }
+ lineageQuery.setMinimumNumberShouldMatch(1);
+ }
+
+ return lineageQuery;
+ }
+
+ @Override
+ public QuerySubmission submitQuery(final Query query, final EventAuthorizer authorizer, final String userId) {
+ validate(query);
+
+ // Check if we have any cached queries first that can give us the answer
+ for (final CachedQuery cachedQuery : cachedQueries) {
+ final Optional<List<Long>> eventIdListOption = cachedQuery.evaluate(query);
+ if (eventIdListOption.isPresent()) {
+ final AsyncQuerySubmission submission = new AsyncQuerySubmission(query, 1, userId);
+ querySubmissionMap.put(query.getIdentifier(), submission);
+
+ final List<Long> eventIds = eventIdListOption.get();
+
+ queryExecutor.submit(() -> {
+ List<ProvenanceEventRecord> events;
+ try {
+ events = eventStore.getEvents(eventIds, authorizer, EventTransformer.EMPTY_TRANSFORMER);
+ submission.getResult().update(events, eventIds.size());
+ } catch (final Exception e) {
+ submission.getResult().setError("Failed to retrieve Provenance Events from store; see logs for more details");
+ logger.error("Failed to retrieve Provenance Events from store", e);
+ }
+ });
+
+ // There are some queries that are optimized and will complete very quickly. As a result,
+ // we don't want to wait for the client to issue a second request, so we will give the query
+ // up to 500 milliseconds to complete before running.
+ try {
+ submission.getResult().awaitCompletion(500, TimeUnit.MILLISECONDS);
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+
+ return submission;
+ }
+ }
+
+ final List<File> indexDirectories = directoryManager.getDirectories(
+ query.getStartDate() == null ? null : query.getStartDate().getTime(),
+ query.getEndDate() == null ? null : query.getEndDate().getTime());
+
+ final AsyncQuerySubmission submission = new AsyncQuerySubmission(query, indexDirectories.size(), userId);
+ querySubmissionMap.put(query.getIdentifier(), submission);
+
+ final org.apache.lucene.search.Query luceneQuery = LuceneUtil.convertQuery(query);
+ logger.debug("Submitting query {} with identifier {} against index directories {}", luceneQuery, query.getIdentifier(), indexDirectories);
+
+ if (indexDirectories.isEmpty()) {
+ submission.getResult().update(Collections.emptyList(), 0L);
+ } else {
+ Collections.sort(indexDirectories, DirectoryUtils.NEWEST_INDEX_FIRST);
+
+ for (final File indexDir : indexDirectories) {
+ queryExecutor.submit(new QueryTask(luceneQuery, submission.getResult(), query.getMaxResults(), indexManager, indexDir,
+ eventStore, authorizer, EventTransformer.EMPTY_TRANSFORMER));
+ }
+ }
+
+ // There are some queries that are optimized and will complete very quickly. As a result,
+ // we don't want to wait for the client to issue a second request, so we will give the query
+ // up to 500 milliseconds to complete before running.
+ try {
+ submission.getResult().awaitCompletion(500, TimeUnit.MILLISECONDS);
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+
+ return submission;
+ }
+
+
+ @Override
+ public ComputeLineageSubmission submitLineageComputation(final String flowFileUuid, final NiFiUser user, final EventAuthorizer eventAuthorizer) {
+ return submitLineageComputation(Collections.singleton(flowFileUuid), user, eventAuthorizer, LineageComputationType.FLOWFILE_LINEAGE, null, 0L, Long.MAX_VALUE);
+ }
+
+ @Override
+ public ComputeLineageSubmission submitExpandChildren(final long eventId, final NiFiUser user, final EventAuthorizer authorizer) {
+ final String userId = user.getIdentity();
+
+ try {
+ final Optional<ProvenanceEventRecord> eventOption = eventStore.getEvent(eventId);
+ if (!eventOption.isPresent()) {
+ final AsyncLineageSubmission submission = new AsyncLineageSubmission(LineageComputationType.EXPAND_CHILDREN, eventId, Collections.emptyList(), 1, userId);
+ lineageSubmissionMap.put(submission.getLineageIdentifier(), submission);
+ submission.getResult().update(Collections.emptyList(), 0L);
+ return submission;
+ }
+
+ final ProvenanceEventRecord event = eventOption.get();
+ switch (event.getEventType()) {
+ case CLONE:
+ case FORK:
+ case JOIN:
+ case REPLAY: {
+ return submitLineageComputation(event.getChildUuids(), user, authorizer, LineageComputationType.EXPAND_CHILDREN,
+ eventId, event.getEventTime(), Long.MAX_VALUE);
+ }
+ default: {
+ final AsyncLineageSubmission submission = new AsyncLineageSubmission(LineageComputationType.EXPAND_CHILDREN,
+ eventId, Collections.<String> emptyList(), 1, userId);
+
+ lineageSubmissionMap.put(submission.getLineageIdentifier(), submission);
+ submission.getResult().setError("Event ID " + eventId + " indicates an event of type " + event.getEventType() + " so its children cannot be expanded");
+ return submission;
+ }
+ }
+ } catch (final Exception e) {
+ final AsyncLineageSubmission submission = new AsyncLineageSubmission(LineageComputationType.EXPAND_CHILDREN,
+ eventId, Collections.<String> emptyList(), 1, userId);
+ lineageSubmissionMap.put(submission.getLineageIdentifier(), submission);
+ submission.getResult().setError("Failed to expand children for lineage of event with ID " + eventId + " due to: " + e);
+ return submission;
+ }
+ }
+
+ @Override
+ public ComputeLineageSubmission submitExpandParents(final long eventId, final NiFiUser user, final EventAuthorizer authorizer) {
+ final String userId = user.getIdentity();
+
+ try {
+ final Optional<ProvenanceEventRecord> eventOption = eventStore.getEvent(eventId);
+ if (!eventOption.isPresent()) {
+ final AsyncLineageSubmission submission = new AsyncLineageSubmission(LineageComputationType.EXPAND_PARENTS, eventId, Collections.emptyList(), 1, userId);
+ lineageSubmissionMap.put(submission.getLineageIdentifier(), submission);
+ submission.getResult().update(Collections.emptyList(), 0L);
+ return submission;
+ }
+
+ final ProvenanceEventRecord event = eventOption.get();
+ switch (event.getEventType()) {
+ case JOIN:
+ case FORK:
+ case CLONE:
+ case REPLAY: {
+ return submitLineageComputation(event.getParentUuids(), user, authorizer, LineageComputationType.EXPAND_PARENTS,
+ eventId, event.getLineageStartDate(), event.getEventTime());
+ }
+ default: {
+ final AsyncLineageSubmission submission = new AsyncLineageSubmission(LineageComputationType.EXPAND_PARENTS,
+ eventId, Collections.<String> emptyList(), 1, userId);
+
+ lineageSubmissionMap.put(submission.getLineageIdentifier(), submission);
+ submission.getResult().setError("Event ID " + eventId + " indicates an event of type " + event.getEventType() + " so its parents cannot be expanded");
+ return submission;
+ }
+ }
+ } catch (final Exception e) {
+ final AsyncLineageSubmission submission = new AsyncLineageSubmission(LineageComputationType.EXPAND_PARENTS,
+ eventId, Collections.<String> emptyList(), 1, userId);
+ lineageSubmissionMap.put(submission.getLineageIdentifier(), submission);
+
+ submission.getResult().setError("Failed to expand parents for lineage of event with ID " + eventId + " due to: " + e);
+ return submission;
+ }
+ }
+
+ @Override
+ public AsyncLineageSubmission retrieveLineageSubmission(final String lineageIdentifier, final NiFiUser user) {
+ final AsyncLineageSubmission submission = lineageSubmissionMap.get(lineageIdentifier);
+ final String userId = submission.getSubmitterIdentity();
+
+ if (user == null && userId == null) {
+ return submission;
+ }
+
+ if (user == null) {
+ throw new AccessDeniedException("Cannot retrieve Provenance Lineage Submission because no user id was provided");
+ }
+
+ if (userId == null || userId.equals(user.getIdentity())) {
+ return submission;
+ }
+
+ throw new AccessDeniedException("Cannot retrieve Provenance Lineage Submission because " + user.getIdentity() + " is not the user who submitted the request");
+ }
+
+ @Override
+ public QuerySubmission retrieveQuerySubmission(final String queryIdentifier, final NiFiUser user) {
+ final QuerySubmission submission = querySubmissionMap.get(queryIdentifier);
+
+ final String userId = submission.getSubmitterIdentity();
+
+ if (user == null && userId == null) {
+ return submission;
+ }
+
+ if (user == null) {
+ throw new AccessDeniedException("Cannot retrieve Provenance Query Submission because no user id was provided");
+ }
+
+ if (userId == null || userId.equals(user.getIdentity())) {
+ return submission;
+ }
+
+ throw new AccessDeniedException("Cannot retrieve Provenance Query Submission because " + user.getIdentity() + " is not the user who submitted the request");
+ }
+
+ @Override
+ public long getSize() {
+ long total = 0;
+ for (final File file : directoryManager.getDirectories(null, null)) {
+ total += DirectoryUtils.getSize(file);
+ }
+ return total;
+ }
+
+ private void validate(final Query query) {
+ final int numQueries = querySubmissionMap.size();
+ if (numQueries > MAX_UNDELETED_QUERY_RESULTS) {
+ throw new IllegalStateException("Cannot process query because there are currently " + numQueries + " queries whose results have not "
+ + "been deleted due to poorly behaving clients not issuing DELETE requests. Please try again later.");
+ }
+
+ if (query.getEndDate() != null && query.getStartDate() != null && query.getStartDate().getTime() > query.getEndDate().getTime()) {
+ throw new IllegalArgumentException("Query End Time cannot be before Query Start Time");
+ }
+ }
+
+ void performMaintenance() {
+ try {
+ final List<ProvenanceEventRecord> firstEvents = eventStore.getEvents(0, 1);
+ if (firstEvents.isEmpty()) {
+ return;
+ }
+
+ final ProvenanceEventRecord firstEvent = firstEvents.get(0);
+ final long earliestEventTime = firstEvent.getEventTime();
+ logger.debug("First Event Time is {} ({}) with Event ID {}; will delete any Lucene Index that is older than this",
+ earliestEventTime, new Date(earliestEventTime), firstEvent.getEventId());
+ final List<File> indicesBeforeEarliestEvent = directoryManager.getDirectoriesBefore(earliestEventTime);
+
+ for (final File index : indicesBeforeEarliestEvent) {
+ logger.debug("Index directory {} is now expired. Attempting to remove index", index);
+ tryDeleteIndex(index);
+ }
+ } catch (final Exception e) {
+ logger.error("Failed to perform background maintenance procedures", e);
+ eventReporter.reportEvent(Severity.ERROR, EVENT_CATEGORY, "Failed to perform maintenance of Provenance Repository. See logs for more information.");
+ }
+ }
+
+ protected boolean tryDeleteIndex(final File indexDirectory) {
+ final long startNanos = System.nanoTime();
+ boolean removed = false;
+ while (!removed && System.nanoTime() - startNanos < TimeUnit.SECONDS.toNanos(MAX_DELETE_INDEX_WAIT_SECONDS)) {
+ removed = indexManager.removeIndex(indexDirectory);
+
+ if (!removed) {
+ try {
+ Thread.sleep(5000L);
+ } catch (final InterruptedException ie) {
+ logger.debug("Interrupted when trying to remove index {} from IndexManager; will not remove index", indexDirectory);
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ }
+ }
+
+ if (removed) {
+ try {
+ FileUtils.deleteFile(indexDirectory, true);
+ logger.debug("Successfully deleted directory {}", indexDirectory);
+ } catch (final IOException e) {
+ logger.warn("The Lucene Index located at " + indexDirectory + " has expired and contains no Provenance Events that still exist in the respository. "
+ + "However, the directory could not be deleted.", e);
+ }
+
+ directoryManager.deleteDirectory(indexDirectory);
+ logger.info("Successfully removed expired Lucene Index {}", indexDirectory);
+ } else {
+ logger.warn("The Lucene Index located at {} has expired and contains no Provenance Events that still exist in the respository. "
+ + "However, the directory could not be deleted because it is still actively being used. Will continue to try to delete "
+ + "in a subsequent maintenance cycle.", indexDirectory);
+ }
+
+ return removed;
+ }
+
+ private class RemoveExpiredQueryResults implements Runnable {
+ @Override
+ public void run() {
+ try {
+ final Date now = new Date();
+
+ final Iterator<Map.Entry<String, AsyncQuerySubmission>> queryIterator = querySubmissionMap.entrySet().iterator();
+ while (queryIterator.hasNext()) {
+ final Map.Entry<String, AsyncQuerySubmission> entry = queryIterator.next();
+
+ final StandardQueryResult result = entry.getValue().getResult();
+ if (entry.getValue().isCanceled() || result.isFinished() && result.getExpiration().before(now)) {
+ queryIterator.remove();
+ }
+ }
+
+ final Iterator<Map.Entry<String, AsyncLineageSubmission>> lineageIterator = lineageSubmissionMap.entrySet().iterator();
+ while (lineageIterator.hasNext()) {
+ final Map.Entry<String, AsyncLineageSubmission> entry = lineageIterator.next();
+
+ final StandardLineageResult result = entry.getValue().getResult();
+ if (entry.getValue().isCanceled() || result.isFinished() && result.getExpiration().before(now)) {
+ lineageIterator.remove();
+ }
+ }
+ } catch (final Exception e) {
+ logger.error("Failed to expire Provenance Query Results due to {}", e.toString());
+ logger.error("", e);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/96ed405d/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/index/lucene/QueryTask.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/index/lucene/QueryTask.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/index/lucene/QueryTask.java
new file mode 100644
index 0000000..38d3f61
--- /dev/null
+++ b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/index/lucene/QueryTask.java
@@ -0,0 +1,208 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.provenance.index.lucene;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.search.Query;
+import org.apache.lucene.search.TopDocs;
+import org.apache.nifi.provenance.ProgressiveResult;
+import org.apache.nifi.provenance.ProvenanceEventRecord;
+import org.apache.nifi.provenance.SearchableFields;
+import org.apache.nifi.provenance.authorization.EventAuthorizer;
+import org.apache.nifi.provenance.authorization.EventTransformer;
+import org.apache.nifi.provenance.index.EventIndexSearcher;
+import org.apache.nifi.provenance.index.SearchFailedException;
+import org.apache.nifi.provenance.lucene.IndexManager;
+import org.apache.nifi.provenance.store.EventStore;
+import org.apache.nifi.util.Tuple;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class QueryTask implements Runnable {
+ private static final Logger logger = LoggerFactory.getLogger(QueryTask.class);
+ private static final Set<String> LUCENE_FIELDS_TO_LOAD = Collections.singleton(SearchableFields.Identifier.getSearchableFieldName());
+
+ private final Query query;
+ private final ProgressiveResult queryResult;
+ private final int maxResults;
+ private final IndexManager indexManager;
+ private final File indexDir;
+ private final EventStore eventStore;
+ private final EventAuthorizer authorizer;
+ private final EventTransformer transformer;
+
+ public QueryTask(final Query query, final ProgressiveResult result, final int maxResults, final IndexManager indexManager,
+ final File indexDir, final EventStore eventStore, final EventAuthorizer authorizer,
+ final EventTransformer unauthorizedTransformer) {
+ this.query = query;
+ this.queryResult = result;
+ this.maxResults = maxResults;
+ this.indexManager = indexManager;
+ this.indexDir = indexDir;
+ this.eventStore = eventStore;
+ this.authorizer = authorizer;
+ this.transformer = unauthorizedTransformer;
+ }
+
+ @Override
+ public void run() {
+ if (queryResult.getTotalHitCount() >= maxResults) {
+ logger.debug("Will not query lucene index {} because maximum results have already been obtained", indexDir);
+ queryResult.update(Collections.emptyList(), 0L);
+ return;
+ }
+
+ if (queryResult.isFinished()) {
+ logger.debug("Will not query lucene index {} because the query is already finished", indexDir);
+ return;
+ }
+
+
+ final long borrowStart = System.nanoTime();
+ final EventIndexSearcher searcher;
+ try {
+ searcher = indexManager.borrowIndexSearcher(indexDir);
+ } catch (final FileNotFoundException fnfe) {
+ // We do not consider this an error because it may well just be the case that the event index has aged off and
+ // been deleted or that we've just created the index and haven't yet committed the writer. So instead, we just
+ // update the result ot indicate that this index search is complete with no results.
+ queryResult.update(Collections.emptyList(), 0);
+
+ // nothing has been indexed yet, or the data has already aged off
+ logger.info("Attempted to search Provenance Index {} but could not find the directory or the directory did not contain a valid Lucene index. "
+ + "This usually indicates that either the index was just created and hasn't fully been initialized, or that the index was recently aged off.", indexDir);
+ return;
+ } catch (final IOException ioe) {
+ queryResult.setError("Failed to query index " + indexDir + "; see logs for more details");
+ logger.error("Failed to query index " + indexDir, ioe);
+ return;
+ }
+
+ try {
+ final long borrowMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - borrowStart);
+ logger.debug("Borrowing index searcher for {} took {} ms", indexDir, borrowMillis);
+ final long startNanos = System.nanoTime();
+
+ // If max number of results are retrieved, do not bother querying lucene
+ if (queryResult.getTotalHitCount() >= maxResults) {
+ logger.debug("Will not query lucene index {} because maximum results have already been obtained", indexDir);
+ queryResult.update(Collections.emptyList(), 0L);
+ return;
+ }
+
+ if (queryResult.isFinished()) {
+ logger.debug("Will not query lucene index {} because the query is already finished", indexDir);
+ return;
+ }
+
+ // Query lucene
+ final IndexReader indexReader = searcher.getIndexSearcher().getIndexReader();
+ final TopDocs topDocs;
+ try {
+ topDocs = searcher.getIndexSearcher().search(query, maxResults);
+ } catch (final Exception e) {
+ logger.error("Failed to query Lucene for index " + indexDir, e);
+ queryResult.setError("Failed to query Lucene for index " + indexDir + " due to " + e);
+ return;
+ } finally {
+ final long ms = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ logger.debug("Querying Lucene for index {} took {} ms", indexDir, ms);
+ }
+
+ // If max number of results are retrieved, do not bother reading docs
+ if (queryResult.getTotalHitCount() >= maxResults) {
+ logger.debug("Will not read events from store for {} because maximum results have already been obtained", indexDir);
+ queryResult.update(Collections.emptyList(), 0L);
+ return;
+ }
+
+ if (queryResult.isFinished()) {
+ logger.debug("Will not read events from store for {} because the query has already finished", indexDir);
+ return;
+ }
+
+ final Tuple<List<ProvenanceEventRecord>, Integer> eventsAndTotalHits = readDocuments(topDocs, indexReader);
+
+ if (eventsAndTotalHits == null) {
+ queryResult.update(Collections.emptyList(), 0L);
+ logger.info("Will not update query results for queried index {} for query {} because the maximum number of results have been reached already",
+ indexDir, query);
+ } else {
+ queryResult.update(eventsAndTotalHits.getKey(), eventsAndTotalHits.getValue());
+
+ final long searchNanos = System.nanoTime() - startNanos;
+ final long millis = TimeUnit.NANOSECONDS.toMillis(searchNanos);
+ logger.info("Successfully queried index {} for query {}; retrieved {} events with a total of {} hits in {} millis",
+ indexDir, query, eventsAndTotalHits.getKey().size(), eventsAndTotalHits.getValue(), millis);
+ }
+ } catch (final Exception e) {
+ logger.error("Failed to query events against index " + indexDir, e);
+ queryResult.setError("Failed to complete query due to " + e);
+ } finally {
+ indexManager.returnIndexSearcher(searcher);
+ }
+ }
+
+ private Tuple<List<ProvenanceEventRecord>, Integer> readDocuments(final TopDocs topDocs, final IndexReader indexReader) {
+ // If no topDocs is supplied, just provide a Tuple that has no records and a hit count of 0.
+ if (topDocs == null || topDocs.totalHits == 0) {
+ return new Tuple<>(Collections.<ProvenanceEventRecord> emptyList(), 0);
+ }
+
+ final long start = System.nanoTime();
+ final List<Long> eventIds = Arrays.stream(topDocs.scoreDocs)
+ .mapToInt(scoreDoc -> scoreDoc.doc)
+ .mapToObj(docId -> {
+ try {
+ return indexReader.document(docId, LUCENE_FIELDS_TO_LOAD);
+ } catch (final Exception e) {
+ throw new SearchFailedException("Failed to read Provenance Events from Event File", e);
+ }
+ })
+ .map(doc -> doc.getField(SearchableFields.Identifier.getSearchableFieldName()).numericValue().longValue())
+ .collect(Collectors.toList());
+
+ final long endConvert = System.nanoTime();
+ final long ms = TimeUnit.NANOSECONDS.toMillis(endConvert - start);
+ logger.debug("Converting documents took {} ms", ms);
+
+ List<ProvenanceEventRecord> events;
+ try {
+ events = eventStore.getEvents(eventIds, authorizer, transformer);
+ } catch (IOException e) {
+ throw new SearchFailedException("Unable to retrieve events from the Provenance Store", e);
+ }
+
+ final long fetchEventNanos = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - endConvert);
+ logger.debug("Fetching {} events from Event Store took {} ms ({} events actually fetched)", eventIds.size(), fetchEventNanos, events.size());
+
+ final int totalHits = topDocs.totalHits;
+ return new Tuple<>(events, totalHits);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/96ed405d/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/index/lucene/StoredDocument.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/index/lucene/StoredDocument.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/index/lucene/StoredDocument.java
new file mode 100644
index 0000000..207ba9f
--- /dev/null
+++ b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/index/lucene/StoredDocument.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.provenance.index.lucene;
+
+import org.apache.lucene.document.Document;
+import org.apache.nifi.provenance.serialization.StorageSummary;
+
+public class StoredDocument {
+ private final Document document;
+ private final StorageSummary storageSummary;
+
+ public StoredDocument(final Document document, final StorageSummary summary) {
+ this.document = document;
+ this.storageSummary = summary;
+ }
+
+ public Document getDocument() {
+ return document;
+ }
+
+ public StorageSummary getStorageSummary() {
+ return storageSummary;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/96ed405d/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/CachingIndexManager.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/CachingIndexManager.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/CachingIndexManager.java
index ddfa0db..eefcecb 100644
--- a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/CachingIndexManager.java
+++ b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/CachingIndexManager.java
@@ -36,6 +36,8 @@ import org.apache.lucene.index.IndexWriterConfig;
import org.apache.lucene.search.IndexSearcher;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.FSDirectory;
+import org.apache.nifi.provenance.index.EventIndexSearcher;
+import org.apache.nifi.provenance.index.EventIndexWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -47,53 +49,61 @@ public class CachingIndexManager implements Closeable, IndexManager {
private final Map<File, List<ActiveIndexSearcher>> activeSearchers = new HashMap<>();
- public void removeIndex(final File indexDirectory) {
+ @Override
+ public boolean removeIndex(final File indexDirectory) {
final File absoluteFile = indexDirectory.getAbsoluteFile();
logger.info("Removing index {}", indexDirectory);
lock.lock();
try {
final IndexWriterCount count = writerCounts.remove(absoluteFile);
- if ( count != null ) {
+ if (count != null) {
try {
count.close();
} catch (final IOException ioe) {
logger.warn("Failed to close Index Writer {} for {}", count.getWriter(), absoluteFile);
- if ( logger.isDebugEnabled() ) {
+ if (logger.isDebugEnabled()) {
logger.warn("", ioe);
}
+
+ return false;
}
}
final List<ActiveIndexSearcher> searcherList = activeSearchers.remove(absoluteFile);
if (searcherList != null) {
- for ( final ActiveIndexSearcher searcher : searcherList ) {
+ for (final ActiveIndexSearcher searcher : searcherList) {
try {
searcher.close();
} catch (final IOException ioe) {
logger.warn("Failed to close Index Searcher {} for {} due to {}",
- searcher.getSearcher(), absoluteFile, ioe);
- if ( logger.isDebugEnabled() ) {
+ searcher.getSearcher(), absoluteFile, ioe);
+ if (logger.isDebugEnabled()) {
logger.warn("", ioe);
}
+
+ return false;
}
}
}
} finally {
lock.unlock();
}
+
+ return true;
}
- public IndexWriter borrowIndexWriter(final File indexingDirectory) throws IOException {
- final File absoluteFile = indexingDirectory.getAbsoluteFile();
- logger.trace("Borrowing index writer for {}", indexingDirectory);
+ @Override
+ public EventIndexWriter borrowIndexWriter(final File indexDirectory) throws IOException {
+ final File absoluteFile = indexDirectory.getAbsoluteFile();
+ logger.trace("Borrowing index writer for {}", indexDirectory);
lock.lock();
try {
IndexWriterCount writerCount = writerCounts.remove(absoluteFile);
- if ( writerCount == null ) {
+ if (writerCount == null) {
final List<Closeable> closeables = new ArrayList<>();
- final Directory directory = FSDirectory.open(indexingDirectory);
+ final Directory directory = FSDirectory.open(indexDirectory);
closeables.add(directory);
try {
@@ -104,10 +114,11 @@ public class CachingIndexManager implements Closeable, IndexManager {
config.setWriteLockTimeout(300000L);
final IndexWriter indexWriter = new IndexWriter(directory, config);
- writerCount = new IndexWriterCount(indexWriter, analyzer, directory, 1);
- logger.debug("Providing new index writer for {}", indexingDirectory);
+ final EventIndexWriter eventIndexWriter = new LuceneEventIndexWriter(indexWriter, indexDirectory);
+ writerCount = new IndexWriterCount(eventIndexWriter, analyzer, directory, 1);
+ logger.debug("Providing new index writer for {}", indexDirectory);
} catch (final IOException ioe) {
- for ( final Closeable closeable : closeables ) {
+ for (final Closeable closeable : closeables) {
try {
closeable.close();
} catch (final IOException ioe2) {
@@ -122,16 +133,16 @@ public class CachingIndexManager implements Closeable, IndexManager {
// Mark any active searchers as poisoned because we are updating the index
final List<ActiveIndexSearcher> searchers = activeSearchers.get(absoluteFile);
- if ( searchers != null ) {
+ if (searchers != null) {
for (final ActiveIndexSearcher activeSearcher : searchers) {
- logger.debug("Poisoning {} because it is searching {}, which is getting updated", activeSearcher, indexingDirectory);
+ logger.debug("Poisoning {} because it is searching {}, which is getting updated", activeSearcher, indexDirectory);
activeSearcher.poison();
}
}
} else {
- logger.debug("Providing existing index writer for {} and incrementing count to {}", indexingDirectory, writerCount.getCount() + 1);
+ logger.debug("Providing existing index writer for {} and incrementing count to {}", indexDirectory, writerCount.getCount() + 1);
writerCounts.put(absoluteFile, new IndexWriterCount(writerCount.getWriter(),
- writerCount.getAnalyzer(), writerCount.getDirectory(), writerCount.getCount() + 1));
+ writerCount.getAnalyzer(), writerCount.getDirectory(), writerCount.getCount() + 1));
}
return writerCount.getWriter();
@@ -140,31 +151,53 @@ public class CachingIndexManager implements Closeable, IndexManager {
}
}
- public void returnIndexWriter(final File indexingDirectory, final IndexWriter writer) {
- final File absoluteFile = indexingDirectory.getAbsoluteFile();
- logger.trace("Returning Index Writer for {} to IndexManager", indexingDirectory);
+
+ @Override
+ public void returnIndexWriter(final EventIndexWriter writer) {
+ returnIndexWriter(writer, true, true);
+ }
+
+ @Override
+ public void returnIndexWriter(final EventIndexWriter writer, final boolean commit, final boolean isCloseable) {
+ final File indexDirectory = writer.getDirectory();
+ final File absoluteFile = indexDirectory.getAbsoluteFile();
+ logger.trace("Returning Index Writer for {} to IndexManager", indexDirectory);
lock.lock();
try {
- final IndexWriterCount count = writerCounts.remove(absoluteFile);
+ final IndexWriterCount count = writerCounts.get(absoluteFile);
try {
- if ( count == null ) {
+ if (count == null) {
logger.warn("Index Writer {} was returned to IndexManager for {}, but this writer is not known. "
- + "This could potentially lead to a resource leak", writer, indexingDirectory);
+ + "This could potentially lead to a resource leak", writer, indexDirectory);
writer.close();
- } else if ( count.getCount() <= 1 ) {
+ } else if (count.getCount() <= 1) {
// we are finished with this writer.
- logger.debug("Decrementing count for Index Writer for {} to {}; Closing writer", indexingDirectory, count.getCount() - 1);
- count.close();
+ logger.info("Decrementing count for Index Writer for {} to {}. Now finished writing to this Index Directory",
+ indexDirectory, count.getCount() - 1);
+
+ try {
+ if (commit) {
+ writer.commit();
+ }
+ } finally {
+ if (isCloseable) {
+ try {
+ count.close();
+ } finally {
+ writerCounts.remove(absoluteFile);
+ }
+ }
+ }
} else {
// decrement the count.
- logger.debug("Decrementing count for Index Writer for {} to {}", indexingDirectory, count.getCount() - 1);
+ logger.debug("Decrementing count for Index Writer for {} to {}", indexDirectory, count.getCount() - 1);
writerCounts.put(absoluteFile, new IndexWriterCount(count.getWriter(), count.getAnalyzer(), count.getDirectory(), count.getCount() - 1));
}
} catch (final IOException ioe) {
logger.warn("Failed to close Index Writer {} due to {}", writer, ioe);
- if ( logger.isDebugEnabled() ) {
+ if (logger.isDebugEnabled()) {
logger.warn("", ioe);
}
}
@@ -174,7 +207,8 @@ public class CachingIndexManager implements Closeable, IndexManager {
}
- public IndexSearcher borrowIndexSearcher(final File indexDir) throws IOException {
+ @Override
+ public EventIndexSearcher borrowIndexSearcher(final File indexDir) throws IOException {
final File absoluteFile = indexDir.getAbsoluteFile();
logger.trace("Borrowing index searcher for {}", indexDir);
@@ -182,7 +216,7 @@ public class CachingIndexManager implements Closeable, IndexManager {
try {
// check if we already have a reader cached.
List<ActiveIndexSearcher> currentlyCached = activeSearchers.get(absoluteFile);
- if ( currentlyCached == null ) {
+ if (currentlyCached == null) {
currentlyCached = new ArrayList<>();
activeSearchers.put(absoluteFile, currentlyCached);
} else {
@@ -197,7 +231,7 @@ public class CachingIndexManager implements Closeable, IndexManager {
// if there are no references to the reader, it will have been closed. Since there is no
// isClosed() method, this is how we determine whether it's been closed or not.
- final int refCount = searcher.getSearcher().getIndexReader().getRefCount();
+ final int refCount = searcher.getSearcher().getIndexSearcher().getIndexReader().getRefCount();
if (refCount <= 0) {
// if refCount == 0, then the reader has been closed, so we cannot use the searcher
logger.debug("Reference count for cached Index Searcher for {} is currently {}; "
@@ -216,16 +250,17 @@ public class CachingIndexManager implements Closeable, IndexManager {
// if we have an Index Writer, and if so create a Reader based on the Index Writer.
// This will provide us a 'near real time' index reader.
final IndexWriterCount writerCount = writerCounts.remove(absoluteFile);
- if ( writerCount == null ) {
+ if (writerCount == null) {
final Directory directory = FSDirectory.open(absoluteFile);
logger.debug("No Index Writer currently exists for {}; creating a cachable reader", indexDir);
try {
final DirectoryReader directoryReader = DirectoryReader.open(directory);
final IndexSearcher searcher = new IndexSearcher(directoryReader);
+ final EventIndexSearcher eventIndexSearcher = new LuceneEventIndexSearcher(searcher, indexDir, directory, directoryReader);
// we want to cache the searcher that we create, since it's just a reader.
- final ActiveIndexSearcher cached = new ActiveIndexSearcher(searcher, absoluteFile, directoryReader, directory, true);
+ final ActiveIndexSearcher cached = new ActiveIndexSearcher(eventIndexSearcher, absoluteFile, directoryReader, directory, true);
currentlyCached.add(cached);
return cached.getSearcher();
@@ -243,22 +278,23 @@ public class CachingIndexManager implements Closeable, IndexManager {
}
} else {
logger.debug("Index Writer currently exists for {}; creating a non-cachable reader and incrementing "
- + "counter to {}", indexDir, writerCount.getCount() + 1);
+ + "counter to {}", indexDir, writerCount.getCount() + 1);
// increment the writer count to ensure that it's kept open.
writerCounts.put(absoluteFile, new IndexWriterCount(writerCount.getWriter(),
- writerCount.getAnalyzer(), writerCount.getDirectory(), writerCount.getCount() + 1));
+ writerCount.getAnalyzer(), writerCount.getDirectory(), writerCount.getCount() + 1));
// create a new Index Searcher from the writer so that we don't have an issue with trying
// to read from a directory that's locked. If we get the "no segments* file found" with
// Lucene, this indicates that an IndexWriter already has the directory open.
- final IndexWriter writer = writerCount.getWriter();
- final DirectoryReader directoryReader = DirectoryReader.open(writer, false);
+ final EventIndexWriter writer = writerCount.getWriter();
+ final DirectoryReader directoryReader = DirectoryReader.open(writer.getIndexWriter(), false);
final IndexSearcher searcher = new IndexSearcher(directoryReader);
+ final EventIndexSearcher eventIndexSearcher = new LuceneEventIndexSearcher(searcher, indexDir, null, directoryReader);
// we don't want to cache this searcher because it's based on a writer, so we want to get
// new values the next time that we search.
- final ActiveIndexSearcher activeSearcher = new ActiveIndexSearcher(searcher, absoluteFile, directoryReader, null, false);
+ final ActiveIndexSearcher activeSearcher = new ActiveIndexSearcher(eventIndexSearcher, absoluteFile, directoryReader, null, false);
currentlyCached.add(activeSearcher);
return activeSearcher.getSearcher();
@@ -269,7 +305,9 @@ public class CachingIndexManager implements Closeable, IndexManager {
}
- public void returnIndexSearcher(final File indexDirectory, final IndexSearcher searcher) {
+ @Override
+ public void returnIndexSearcher(final EventIndexSearcher searcher) {
+ final File indexDirectory = searcher.getIndexDirectory();
final File absoluteFile = indexDirectory.getAbsoluteFile();
logger.trace("Returning index searcher for {} to IndexManager", indexDirectory);
@@ -277,9 +315,9 @@ public class CachingIndexManager implements Closeable, IndexManager {
try {
// check if we already have a reader cached.
final List<ActiveIndexSearcher> currentlyCached = activeSearchers.get(absoluteFile);
- if ( currentlyCached == null ) {
+ if (currentlyCached == null) {
logger.warn("Received Index Searcher for {} but no searcher was provided for that directory; this could "
- + "result in a resource leak", indexDirectory);
+ + "result in a resource leak", indexDirectory);
return;
}
@@ -289,20 +327,20 @@ public class CachingIndexManager implements Closeable, IndexManager {
boolean activeSearcherFound = false;
while (itr.hasNext()) {
final ActiveIndexSearcher activeSearcher = itr.next();
- if ( activeSearcher.getSearcher().equals(searcher) ) {
+ if (activeSearcher.getSearcher().equals(searcher)) {
activeSearcherFound = true;
- if ( activeSearcher.isCache() ) {
+ if (activeSearcher.isCache()) {
// if the searcher is poisoned, close it and remove from "pool". Otherwise,
// just decrement the count. Note here that when we call close() it won't actually close
// the underlying directory reader unless there are no more references to it
- if ( activeSearcher.isPoisoned() ) {
+ if (activeSearcher.isPoisoned()) {
itr.remove();
try {
activeSearcher.close();
} catch (final IOException ioe) {
logger.warn("Failed to close Index Searcher for {} due to {}", absoluteFile, ioe);
- if ( logger.isDebugEnabled() ) {
+ if (logger.isDebugEnabled()) {
logger.warn("", ioe);
}
}
@@ -322,26 +360,26 @@ public class CachingIndexManager implements Closeable, IndexManager {
// decrement the writer count because we incremented it when creating the searcher
final IndexWriterCount writerCount = writerCounts.remove(absoluteFile);
- if ( writerCount != null ) {
- if ( writerCount.getCount() <= 1 ) {
+ if (writerCount != null) {
+ if (writerCount.getCount() <= 1) {
try {
logger.debug("Index searcher for {} is not cached. Writer count is "
- + "decremented to {}; closing writer", indexDirectory, writerCount.getCount() - 1);
+ + "decremented to {}; closing writer", indexDirectory, writerCount.getCount() - 1);
writerCount.close();
} catch (final IOException ioe) {
logger.warn("Failed to close Index Writer for {} due to {}", absoluteFile, ioe);
- if ( logger.isDebugEnabled() ) {
+ if (logger.isDebugEnabled()) {
logger.warn("", ioe);
}
}
} else {
logger.debug("Index searcher for {} is not cached. Writer count is decremented "
- + "to {}; leaving writer open", indexDirectory, writerCount.getCount() - 1);
+ + "to {}; leaving writer open", indexDirectory, writerCount.getCount() - 1);
writerCounts.put(absoluteFile, new IndexWriterCount(writerCount.getWriter(),
- writerCount.getAnalyzer(), writerCount.getDirectory(),
- writerCount.getCount() - 1));
+ writerCount.getAnalyzer(), writerCount.getDirectory(),
+ writerCount.getCount() - 1));
}
}
@@ -353,7 +391,7 @@ public class CachingIndexManager implements Closeable, IndexManager {
}
} catch (final IOException ioe) {
logger.warn("Failed to close Index Searcher for {} due to {}", absoluteFile, ioe);
- if ( logger.isDebugEnabled() ) {
+ if (logger.isDebugEnabled()) {
logger.warn("", ioe);
}
}
@@ -378,11 +416,11 @@ public class CachingIndexManager implements Closeable, IndexManager {
try {
IOException ioe = null;
- for ( final IndexWriterCount count : writerCounts.values() ) {
+ for (final IndexWriterCount count : writerCounts.values()) {
try {
count.close();
} catch (final IOException e) {
- if ( ioe == null ) {
+ if (ioe == null) {
ioe = e;
} else {
ioe.addSuppressed(e);
@@ -395,7 +433,7 @@ public class CachingIndexManager implements Closeable, IndexManager {
try {
searcher.close();
} catch (final IOException e) {
- if ( ioe == null ) {
+ if (ioe == null) {
ioe = e;
} else {
ioe.addSuppressed(e);
@@ -404,7 +442,7 @@ public class CachingIndexManager implements Closeable, IndexManager {
}
}
- if ( ioe != null ) {
+ if (ioe != null) {
throw ioe;
}
} finally {
@@ -415,15 +453,15 @@ public class CachingIndexManager implements Closeable, IndexManager {
private static void close(final Closeable... closeables) throws IOException {
IOException ioe = null;
- for ( final Closeable closeable : closeables ) {
- if ( closeable == null ) {
+ for (final Closeable closeable : closeables) {
+ if (closeable == null) {
continue;
}
try {
closeable.close();
} catch (final IOException e) {
- if ( ioe == null ) {
+ if (ioe == null) {
ioe = e;
} else {
ioe.addSuppressed(e);
@@ -431,14 +469,14 @@ public class CachingIndexManager implements Closeable, IndexManager {
}
}
- if ( ioe != null ) {
+ if (ioe != null) {
throw ioe;
}
}
private static class ActiveIndexSearcher {
- private final IndexSearcher searcher;
+ private final EventIndexSearcher searcher;
private final DirectoryReader directoryReader;
private final File indexDirectory;
private final Directory directory;
@@ -446,8 +484,8 @@ public class CachingIndexManager implements Closeable, IndexManager {
private final AtomicInteger referenceCount = new AtomicInteger(1);
private volatile boolean poisoned = false;
- public ActiveIndexSearcher(final IndexSearcher searcher, final File indexDirectory, final DirectoryReader directoryReader,
- final Directory directory, final boolean cache) {
+ public ActiveIndexSearcher(final EventIndexSearcher searcher, final File indexDirectory, final DirectoryReader directoryReader,
+ final Directory directory, final boolean cache) {
this.searcher = searcher;
this.directoryReader = directoryReader;
this.indexDirectory = indexDirectory;
@@ -459,7 +497,7 @@ public class CachingIndexManager implements Closeable, IndexManager {
return cache;
}
- public IndexSearcher getSearcher() {
+ public EventIndexSearcher getSearcher() {
return searcher;
}
@@ -499,12 +537,12 @@ public class CachingIndexManager implements Closeable, IndexManager {
private static class IndexWriterCount implements Closeable {
- private final IndexWriter writer;
+ private final EventIndexWriter writer;
private final Analyzer analyzer;
private final Directory directory;
private final int count;
- public IndexWriterCount(final IndexWriter writer, final Analyzer analyzer, final Directory directory, final int count) {
+ public IndexWriterCount(final EventIndexWriter writer, final Analyzer analyzer, final Directory directory, final int count) {
this.writer = writer;
this.analyzer = analyzer;
this.directory = directory;
@@ -519,7 +557,7 @@ public class CachingIndexManager implements Closeable, IndexManager {
return directory;
}
- public IndexWriter getWriter() {
+ public EventIndexWriter getWriter() {
return writer;
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/96ed405d/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/DeleteIndexAction.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/DeleteIndexAction.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/DeleteIndexAction.java
index 7707352..f372a2d 100644
--- a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/DeleteIndexAction.java
+++ b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/DeleteIndexAction.java
@@ -25,6 +25,7 @@ import org.apache.lucene.index.Term;
import org.apache.nifi.provenance.IndexConfiguration;
import org.apache.nifi.provenance.PersistentProvenanceRepository;
import org.apache.nifi.provenance.expiration.ExpirationAction;
+import org.apache.nifi.provenance.index.EventIndexWriter;
import org.apache.nifi.provenance.serialization.RecordReader;
import org.apache.nifi.provenance.serialization.RecordReaders;
import org.slf4j.Logger;
@@ -60,15 +61,16 @@ public class DeleteIndexAction implements ExpirationAction {
final Term term = new Term(FieldNames.STORAGE_FILENAME, LuceneUtil.substringBefore(expiredFile.getName(), "."));
boolean deleteDir = false;
- final IndexWriter writer = indexManager.borrowIndexWriter(indexingDirectory);
+ final EventIndexWriter writer = indexManager.borrowIndexWriter(indexingDirectory);
try {
- writer.deleteDocuments(term);
- writer.commit();
- final int docsLeft = writer.numDocs();
+ final IndexWriter indexWriter = writer.getIndexWriter();
+ indexWriter.deleteDocuments(term);
+ indexWriter.commit();
+ final int docsLeft = indexWriter.numDocs();
deleteDir = docsLeft <= 0;
logger.debug("After expiring {}, there are {} docs left for index {}", expiredFile, docsLeft, indexingDirectory);
} finally {
- indexManager.returnIndexWriter(indexingDirectory, writer);
+ indexManager.returnIndexWriter(writer);
}
// we've confirmed that all documents have been removed. Delete the index directory.
http://git-wip-us.apache.org/repos/asf/nifi/blob/96ed405d/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/DocsReader.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/DocsReader.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/DocsReader.java
index ce62152..0e96b62 100644
--- a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/DocsReader.java
+++ b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/DocsReader.java
@@ -30,25 +30,25 @@ import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.IndexableField;
+import org.apache.lucene.search.ScoreDoc;
+import org.apache.lucene.search.TopDocs;
import org.apache.nifi.provenance.ProvenanceEventRecord;
import org.apache.nifi.provenance.SearchableFields;
import org.apache.nifi.provenance.StandardProvenanceEventRecord;
-import org.apache.nifi.provenance.authorization.AuthorizationCheck;
+import org.apache.nifi.provenance.authorization.EventAuthorizer;
import org.apache.nifi.provenance.serialization.RecordReader;
import org.apache.nifi.provenance.serialization.RecordReaders;
import org.apache.nifi.provenance.toc.TocReader;
-import org.apache.lucene.document.Document;
-import org.apache.lucene.index.IndexReader;
-import org.apache.lucene.index.IndexableField;
-import org.apache.lucene.search.ScoreDoc;
-import org.apache.lucene.search.TopDocs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-class DocsReader {
+public class DocsReader {
private final Logger logger = LoggerFactory.getLogger(DocsReader.class);
- public Set<ProvenanceEventRecord> read(final TopDocs topDocs, final AuthorizationCheck authCheck, final IndexReader indexReader, final Collection<Path> allProvenanceLogFiles,
+ public Set<ProvenanceEventRecord> read(final TopDocs topDocs, final EventAuthorizer authorizer, final IndexReader indexReader, final Collection<Path> allProvenanceLogFiles,
final AtomicInteger retrievalCount, final int maxResults, final int maxAttributeChars) throws IOException {
if (retrievalCount.get() >= maxResults) {
return Collections.emptySet();
@@ -67,7 +67,7 @@ class DocsReader {
final long readDocuments = System.nanoTime() - start;
logger.debug("Reading {} Lucene Documents took {} millis", docs.size(), TimeUnit.NANOSECONDS.toMillis(readDocuments));
- return read(docs, authCheck, allProvenanceLogFiles, retrievalCount, maxResults, maxAttributeChars);
+ return read(docs, authorizer, allProvenanceLogFiles, retrievalCount, maxResults, maxAttributeChars);
}
@@ -106,7 +106,7 @@ class DocsReader {
return record;
}
- public Set<ProvenanceEventRecord> read(final List<Document> docs, final AuthorizationCheck authCheck, final Collection<Path> allProvenanceLogFiles,
+ public Set<ProvenanceEventRecord> read(final List<Document> docs, final EventAuthorizer authorizer, final Collection<Path> allProvenanceLogFiles,
final AtomicInteger retrievalCount, final int maxResults, final int maxAttributeChars) throws IOException {
if (retrievalCount.get() >= maxResults) {
@@ -114,38 +114,33 @@ class DocsReader {
}
final long start = System.nanoTime();
-
- Set<ProvenanceEventRecord> matchingRecords = new LinkedHashSet<>();
-
- Map<String, List<Document>> byStorageNameDocGroups = LuceneUtil.groupDocsByStorageFileName(docs);
+ final Set<ProvenanceEventRecord> matchingRecords = new LinkedHashSet<>();
+ final Map<String, List<Document>> byStorageNameDocGroups = LuceneUtil.groupDocsByStorageFileName(docs);
int eventsReadThisFile = 0;
int logFileCount = 0;
for (String storageFileName : byStorageNameDocGroups.keySet()) {
- File provenanceEventFile = LuceneUtil.getProvenanceLogFile(storageFileName, allProvenanceLogFiles);
- if (provenanceEventFile != null) {
- try (RecordReader reader = RecordReaders.newRecordReader(provenanceEventFile, allProvenanceLogFiles,
- maxAttributeChars)) {
-
- Iterator<Document> docIter = byStorageNameDocGroups.get(storageFileName).iterator();
- while (docIter.hasNext() && retrievalCount.getAndIncrement() < maxResults) {
- ProvenanceEventRecord event = this.getRecord(docIter.next(), reader);
- if (event != null && authCheck.isAuthorized(event)) {
- matchingRecords.add(event);
- eventsReadThisFile++;
- }
- }
+ final File provenanceEventFile = LuceneUtil.getProvenanceLogFile(storageFileName, allProvenanceLogFiles);
+ if (provenanceEventFile == null) {
+ logger.warn("Could not find Provenance Log File with "
+ + "basename {} in the Provenance Repository; assuming "
+ + "file has expired and continuing without it", storageFileName);
+ continue;
+ }
- } catch (Exception e) {
- logger.warn("Failed while trying to read Provenance Events. The event file '"
- + provenanceEventFile.getAbsolutePath() +
- "' may be missing or corrupted.", e);
+ try (final RecordReader reader = RecordReaders.newRecordReader(provenanceEventFile, allProvenanceLogFiles, maxAttributeChars)) {
+ final Iterator<Document> docIter = byStorageNameDocGroups.get(storageFileName).iterator();
+ while (docIter.hasNext() && retrievalCount.getAndIncrement() < maxResults) {
+ final ProvenanceEventRecord event = getRecord(docIter.next(), reader);
+ if (event != null && authorizer.isAuthorized(event)) {
+ matchingRecords.add(event);
+ eventsReadThisFile++;
+ }
}
- } else {
- logger.warn("Could not find Provenance Log File with "
- + "basename {} in the Provenance Repository; assuming "
- + "file has expired and continuing without it", storageFileName);
+ } catch (final Exception e) {
+ logger.warn("Failed to read Provenance Events. The event file '"
+ + provenanceEventFile.getAbsolutePath() + "' may be missing or corrupt.", e);
}
}