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:36 UTC
[06/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/lucene/DocumentToEventConverter.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/DocumentToEventConverter.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/DocumentToEventConverter.java
new file mode 100644
index 0000000..18d3860
--- /dev/null
+++ b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/DocumentToEventConverter.java
@@ -0,0 +1,30 @@
+/*
+ * 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.lucene;
+
+import java.io.IOException;
+import java.util.Set;
+
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.search.TopDocs;
+import org.apache.nifi.provenance.ProvenanceEventRecord;
+
+public interface DocumentToEventConverter {
+
+ Set<ProvenanceEventRecord> convert(TopDocs topDocs, IndexReader indexReader) throws IOException;
+}
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/IndexManager.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/IndexManager.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/IndexManager.java
index f84021f..331d141 100644
--- a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/IndexManager.java
+++ b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/IndexManager.java
@@ -21,17 +21,19 @@ import java.io.Closeable;
import java.io.File;
import java.io.IOException;
-import org.apache.lucene.index.IndexWriter;
-import org.apache.lucene.search.IndexSearcher;
+import org.apache.nifi.provenance.index.EventIndexSearcher;
+import org.apache.nifi.provenance.index.EventIndexWriter;
public interface IndexManager extends Closeable {
- IndexSearcher borrowIndexSearcher(File indexDir) throws IOException;
+ EventIndexSearcher borrowIndexSearcher(File indexDir) throws IOException;
- IndexWriter borrowIndexWriter(File indexingDirectory) throws IOException;
+ EventIndexWriter borrowIndexWriter(File indexDirectory) throws IOException;
- void removeIndex(final File indexDirectory);
+ boolean removeIndex(final File indexDirectory);
- void returnIndexSearcher(File indexDirectory, IndexSearcher searcher);
+ void returnIndexSearcher(EventIndexSearcher searcher);
- void returnIndexWriter(File indexingDirectory, IndexWriter writer);
+ void returnIndexWriter(EventIndexWriter writer, boolean commit, boolean isCloseable);
+
+ void returnIndexWriter(EventIndexWriter writer);
}
\ No newline at end of file
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/IndexSearch.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/IndexSearch.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/IndexSearch.java
index 8d7df8b..514af38 100644
--- a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/IndexSearch.java
+++ b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/IndexSearch.java
@@ -21,18 +21,20 @@ import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Collections;
import java.util.Date;
+import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
-import org.apache.lucene.search.IndexSearcher;
import org.apache.lucene.search.Query;
import org.apache.lucene.search.TopDocs;
+import org.apache.nifi.authorization.AccessDeniedException;
import org.apache.nifi.authorization.user.NiFiUser;
import org.apache.nifi.provenance.PersistentProvenanceRepository;
import org.apache.nifi.provenance.ProvenanceEventRecord;
import org.apache.nifi.provenance.StandardQueryResult;
-import org.apache.nifi.provenance.authorization.AuthorizationCheck;
+import org.apache.nifi.provenance.authorization.EventAuthorizer;
+import org.apache.nifi.provenance.index.EventIndexSearcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -86,14 +88,14 @@ public class IndexSearch {
final Query luceneQuery = LuceneUtil.convertQuery(provenanceQuery);
final long start = System.nanoTime();
- IndexSearcher searcher = null;
+ EventIndexSearcher searcher = null;
try {
searcher = indexManager.borrowIndexSearcher(indexDirectory);
final long searchStartNanos = System.nanoTime();
final long openSearcherNanos = searchStartNanos - start;
logger.debug("Searching {} for {}", this, provenanceQuery);
- final TopDocs topDocs = searcher.search(luceneQuery, provenanceQuery.getMaxResults());
+ final TopDocs topDocs = searcher.getIndexSearcher().search(luceneQuery, provenanceQuery.getMaxResults());
final long finishSearch = System.nanoTime();
final long searchNanos = finishSearch - searchStartNanos;
@@ -107,9 +109,29 @@ public class IndexSearch {
final DocsReader docsReader = new DocsReader();
- final AuthorizationCheck authCheck = event -> repository.isAuthorized(event, user);
-
- matchingRecords = docsReader.read(topDocs, authCheck, searcher.getIndexReader(), repository.getAllLogFiles(), retrievedCount,
+ final EventAuthorizer authorizer = new EventAuthorizer() {
+ @Override
+ public boolean isAuthorized(ProvenanceEventRecord event) {
+ return repository.isAuthorized(event, user);
+ }
+
+ @Override
+ public void authorize(ProvenanceEventRecord event) throws AccessDeniedException {
+ repository.authorize(event, user);
+ }
+
+ @Override
+ public List<ProvenanceEventRecord> filterUnauthorizedEvents(List<ProvenanceEventRecord> events) {
+ return repository.filterUnauthorizedEvents(events, user);
+ }
+
+ @Override
+ public Set<ProvenanceEventRecord> replaceUnauthorizedWithPlaceholders(Set<ProvenanceEventRecord> events) {
+ return repository.replaceUnauthorizedWithPlaceholders(events, user);
+ }
+ };
+
+ matchingRecords = docsReader.read(topDocs, authorizer, searcher.getIndexSearcher().getIndexReader(), repository.getAllLogFiles(), retrievedCount,
provenanceQuery.getMaxResults(), maxAttributeChars);
final long readRecordsNanos = System.nanoTime() - finishSearch;
@@ -133,7 +155,7 @@ public class IndexSearch {
return sqr;
} finally {
if ( searcher != null ) {
- indexManager.returnIndexSearcher(indexDirectory, searcher);
+ indexManager.returnIndexSearcher(searcher);
}
}
}
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/IndexingAction.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/IndexingAction.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/IndexingAction.java
index f725208..a0be319 100644
--- a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/IndexingAction.java
+++ b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/IndexingAction.java
@@ -19,6 +19,7 @@ package org.apache.nifi.provenance.lucene;
import java.io.IOException;
import java.util.Collections;
import java.util.HashSet;
+import java.util.List;
import java.util.Set;
import org.apache.lucene.document.Document;
@@ -28,23 +29,22 @@ import org.apache.lucene.document.LongField;
import org.apache.lucene.document.StringField;
import org.apache.lucene.index.IndexWriter;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
-import org.apache.nifi.provenance.PersistentProvenanceRepository;
import org.apache.nifi.provenance.ProvenanceEventType;
import org.apache.nifi.provenance.SearchableFields;
import org.apache.nifi.provenance.StandardProvenanceEventRecord;
import org.apache.nifi.provenance.search.SearchableField;
public class IndexingAction {
- private final Set<SearchableField> nonAttributeSearchableFields;
- private final Set<SearchableField> attributeSearchableFields;
+ private final Set<SearchableField> searchableEventFields;
+ private final Set<SearchableField> searchableAttributeFields;
- public IndexingAction(final PersistentProvenanceRepository repo) {
- attributeSearchableFields = Collections.unmodifiableSet(new HashSet<>(repo.getConfiguration().getSearchableAttributes()));
- nonAttributeSearchableFields = Collections.unmodifiableSet(new HashSet<>(repo.getConfiguration().getSearchableFields()));
+ public IndexingAction(final List<SearchableField> searchableEventFields, final List<SearchableField> searchableAttributes) {
+ this.searchableEventFields = Collections.unmodifiableSet(new HashSet<>(searchableEventFields));
+ this.searchableAttributeFields = Collections.unmodifiableSet(new HashSet<>(searchableAttributes));
}
private void addField(final Document doc, final SearchableField field, final String value, final Store store) {
- if (value == null || (!field.isAttribute() && !nonAttributeSearchableFields.contains(field))) {
+ if (value == null || (!field.isAttribute() && !searchableEventFields.contains(field))) {
return;
}
@@ -67,7 +67,7 @@ public class IndexingAction {
addField(doc, SearchableFields.SourceQueueIdentifier, record.getSourceQueueIdentifier(), Store.NO);
addField(doc, SearchableFields.TransitURI, record.getTransitUri(), Store.NO);
- for (final SearchableField searchableField : attributeSearchableFields) {
+ for (final SearchableField searchableField : searchableAttributeFields) {
addField(doc, searchableField, LuceneUtil.truncateIndexField(record.getAttribute(searchableField.getSearchableFieldName())), Store.NO);
}
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/LineageQuery.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/LineageQuery.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/LineageQuery.java
index 1b13504..2388483 100644
--- a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/LineageQuery.java
+++ b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/LineageQuery.java
@@ -25,18 +25,15 @@ import java.util.Collection;
import java.util.Collections;
import java.util.Set;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
import org.apache.lucene.index.Term;
import org.apache.lucene.search.BooleanClause.Occur;
import org.apache.lucene.search.BooleanQuery;
-import org.apache.lucene.search.IndexSearcher;
import org.apache.lucene.search.TermQuery;
import org.apache.lucene.search.TopDocs;
-import org.apache.nifi.provenance.PersistentProvenanceRepository;
import org.apache.nifi.provenance.ProvenanceEventRecord;
import org.apache.nifi.provenance.SearchableFields;
-import org.apache.nifi.provenance.authorization.AuthorizationCheck;
+import org.apache.nifi.provenance.index.EventIndexSearcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -46,8 +43,8 @@ public class LineageQuery {
public static final int MAX_LINEAGE_UUIDS = 100;
private static final Logger logger = LoggerFactory.getLogger(LineageQuery.class);
- public static Set<ProvenanceEventRecord> computeLineageForFlowFiles(final PersistentProvenanceRepository repo, final IndexManager indexManager, final File indexDirectory,
- final String lineageIdentifier, final Collection<String> flowFileUuids, final int maxAttributeChars) throws IOException {
+ public static Set<ProvenanceEventRecord> computeLineageForFlowFiles(final IndexManager indexManager, final File indexDirectory,
+ final String lineageIdentifier, final Collection<String> flowFileUuids, final DocumentToEventConverter docsToEventConverter) throws IOException {
if (requireNonNull(flowFileUuids).size() > MAX_LINEAGE_UUIDS) {
throw new IllegalArgumentException(String.format("Cannot compute lineage for more than %s FlowFiles. This lineage contains %s.", MAX_LINEAGE_UUIDS, flowFileUuids.size()));
}
@@ -56,7 +53,7 @@ public class LineageQuery {
throw new IllegalArgumentException("Must specify either Lineage Identifier or FlowFile UUIDs to compute lineage");
}
- final IndexSearcher searcher;
+ final EventIndexSearcher searcher;
try {
searcher = indexManager.borrowIndexSearcher(indexDirectory);
try {
@@ -75,16 +72,10 @@ public class LineageQuery {
final long searchStart = System.nanoTime();
logger.debug("Searching {} for {}", indexDirectory, flowFileIdQuery);
- final TopDocs uuidQueryTopDocs = searcher.search(flowFileIdQuery, MAX_QUERY_RESULTS);
+ final TopDocs uuidQueryTopDocs = searcher.getIndexSearcher().search(flowFileIdQuery, MAX_QUERY_RESULTS);
final long searchEnd = System.nanoTime();
- // Always authorized. We do this because we need to pull back the event, regardless of whether or not
- // the user is truly authorized, because instead of ignoring unauthorized events, we want to replace them.
- final AuthorizationCheck authCheck = event -> true;
-
- final DocsReader docsReader = new DocsReader();
- final Set<ProvenanceEventRecord> recs = docsReader.read(uuidQueryTopDocs, authCheck, searcher.getIndexReader(), repo.getAllLogFiles(),
- new AtomicInteger(0), Integer.MAX_VALUE, maxAttributeChars);
+ final Set<ProvenanceEventRecord> recs = docsToEventConverter.convert(uuidQueryTopDocs, searcher.getIndexSearcher().getIndexReader());
final long readDocsEnd = System.nanoTime();
logger.debug("Finished Lineage Query against {}; Lucene search took {} millis, reading records took {} millis",
@@ -92,7 +83,7 @@ public class LineageQuery {
return recs;
} finally {
- indexManager.returnIndexSearcher(indexDirectory, searcher);
+ indexManager.returnIndexSearcher(searcher);
}
} catch (final FileNotFoundException fnfe) {
// nothing has been indexed yet, or the data has already aged off
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/LuceneEventIndexSearcher.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/LuceneEventIndexSearcher.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/LuceneEventIndexSearcher.java
new file mode 100644
index 0000000..07b9167
--- /dev/null
+++ b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/LuceneEventIndexSearcher.java
@@ -0,0 +1,92 @@
+/*
+ * 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.lucene;
+
+import java.io.Closeable;
+import java.io.File;
+
+import org.apache.lucene.index.DirectoryReader;
+import org.apache.lucene.search.IndexSearcher;
+import org.apache.lucene.store.Directory;
+import org.apache.nifi.provenance.index.EventIndexSearcher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class LuceneEventIndexSearcher implements EventIndexSearcher {
+ private static final Logger logger = LoggerFactory.getLogger(LuceneEventIndexSearcher.class);
+
+ private final IndexSearcher indexSearcher;
+ private final File indexDirectory;
+ private final Directory directory;
+ private final DirectoryReader directoryReader;
+
+ // guarded by synchronizing on 'this'
+ private int usageCounter = 0;
+ private boolean closed = false;
+
+ public LuceneEventIndexSearcher(final IndexSearcher indexSearcher, final File indexDirectory, final Directory directory, final DirectoryReader directoryReader) {
+ this.indexSearcher = indexSearcher;
+ this.indexDirectory = indexDirectory;
+ this.directory = directory;
+ this.directoryReader = directoryReader;
+ }
+
+ @Override
+ public IndexSearcher getIndexSearcher() {
+ return indexSearcher;
+ }
+
+ @Override
+ public File getIndexDirectory() {
+ return indexDirectory;
+ }
+
+ @Override
+ public synchronized void close() {
+ closed = true;
+ if (usageCounter == 0) {
+ closeQuietly(directoryReader);
+ closeQuietly(directory);
+ }
+ }
+
+ public synchronized void incrementUsageCounter() {
+ usageCounter++;
+ }
+
+ public synchronized void decrementUsageCounter() {
+ usageCounter--;
+ if (usageCounter == 0 && closed) {
+ closeQuietly(directoryReader);
+ closeQuietly(directory);
+ }
+ }
+
+ private void closeQuietly(final Closeable closeable) {
+ if (closeable == null) {
+ return;
+ }
+
+ try {
+ closeable.close();
+ } catch (final Exception e) {
+ logger.warn("Failed to close {} due to {}", closeable, 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/lucene/LuceneEventIndexWriter.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/LuceneEventIndexWriter.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/LuceneEventIndexWriter.java
new file mode 100644
index 0000000..db8c528
--- /dev/null
+++ b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/LuceneEventIndexWriter.java
@@ -0,0 +1,144 @@
+/*
+ * 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.lucene;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.lucene.document.Document;
+import org.apache.lucene.index.IndexWriter;
+import org.apache.nifi.provenance.index.EventIndexWriter;
+
+public class LuceneEventIndexWriter implements EventIndexWriter {
+ private final IndexWriter indexWriter;
+ private final File directory;
+ private final long maxCommitNanos;
+
+ private final AtomicReference<CommitStats> commitStats = new AtomicReference<>();
+ private final AtomicLong totalIndexed = new AtomicLong(0L);
+ private final AtomicLong lastCommitTotalIndexed = new AtomicLong(0L);
+
+ public LuceneEventIndexWriter(final IndexWriter indexWriter, final File directory) {
+ this(indexWriter, directory, TimeUnit.SECONDS.toNanos(30L));
+ }
+
+ public LuceneEventIndexWriter(final IndexWriter indexWriter, final File directory, final long maxCommitNanos) {
+ this.indexWriter = indexWriter;
+ this.directory = directory;
+ this.maxCommitNanos = maxCommitNanos;
+
+ commitStats.set(new CommitStats(0, System.nanoTime() + maxCommitNanos));
+ }
+
+ @Override
+ public void close() throws IOException {
+ indexWriter.close();
+ }
+
+ @Override
+ public boolean index(final Document document, final int commitThreshold) throws IOException {
+ return index(Collections.singletonList(document), commitThreshold);
+ }
+
+ @Override
+ public boolean index(List<Document> documents, final int commitThreshold) throws IOException {
+ if (documents.isEmpty()) {
+ return false;
+ }
+
+ final int numDocs = documents.size();
+ indexWriter.addDocuments(documents);
+ totalIndexed.addAndGet(numDocs);
+
+ boolean updated = false;
+ while (!updated) {
+ final CommitStats stats = commitStats.get();
+ CommitStats updatedStats = new CommitStats(stats.getIndexedSinceCommit() + numDocs, stats.getNextCommitTimestamp());
+
+ if (updatedStats.getIndexedSinceCommit() >= commitThreshold || System.nanoTime() >= updatedStats.getNextCommitTimestamp()) {
+ updatedStats = new CommitStats(0, System.nanoTime() + maxCommitNanos);
+ updated = commitStats.compareAndSet(stats, updatedStats);
+ if (updated) {
+ return true;
+ }
+ } else {
+ updated = commitStats.compareAndSet(stats, updatedStats);
+ }
+ }
+
+ return false;
+ }
+
+ @Override
+ public File getDirectory() {
+ return directory;
+ }
+
+ @Override
+ public long commit() throws IOException {
+ final long lastCommitCount = lastCommitTotalIndexed.get();
+ final long currentCommitCount = totalIndexed.get();
+ indexWriter.commit();
+ commitStats.set(new CommitStats(0, System.nanoTime() + maxCommitNanos));
+ lastCommitTotalIndexed.set(currentCommitCount);
+ return currentCommitCount - lastCommitCount;
+ }
+
+ @Override
+ public int getEventsIndexedSinceCommit() {
+ return commitStats.get().getIndexedSinceCommit();
+ }
+
+ @Override
+ public long getEventsIndexed() {
+ return totalIndexed.get();
+ }
+
+ @Override
+ public IndexWriter getIndexWriter() {
+ return indexWriter;
+ }
+
+ @Override
+ public String toString() {
+ return "LuceneEventIndexWriter[dir=" + directory + "]";
+ }
+
+ private static class CommitStats {
+ private final long nextCommitTimestamp;
+ private final int indexedSinceCommit;
+
+ public CommitStats(final int indexedCount, final long nextCommitTime) {
+ this.nextCommitTimestamp = nextCommitTime;
+ this.indexedSinceCommit = indexedCount;
+ }
+
+ public long getNextCommitTimestamp() {
+ return nextCommitTimestamp;
+ }
+
+ public int getIndexedSinceCommit() {
+ return indexedSinceCommit;
+ }
+ }
+}
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/SimpleIndexManager.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/SimpleIndexManager.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/SimpleIndexManager.java
index 81816d2..b0b01e5 100644
--- a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/SimpleIndexManager.java
+++ b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/SimpleIndexManager.java
@@ -24,155 +24,306 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.analysis.standard.StandardAnalyzer;
+import org.apache.lucene.index.ConcurrentMergeScheduler;
import org.apache.lucene.index.DirectoryReader;
import org.apache.lucene.index.IndexWriter;
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.RepositoryConfiguration;
+import org.apache.nifi.provenance.index.EventIndexSearcher;
+import org.apache.nifi.provenance.index.EventIndexWriter;
+import org.apache.nifi.provenance.util.NamedThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class SimpleIndexManager implements IndexManager {
private static final Logger logger = LoggerFactory.getLogger(SimpleIndexManager.class);
- private final ConcurrentMap<Object, List<Closeable>> closeables = new ConcurrentHashMap<>();
- private final Map<File, IndexWriterCount> writerCounts = new HashMap<>();
-
- private final ExecutorService searchExecutor = Executors.newCachedThreadPool();
+ private final Map<File, IndexWriterCount> writerCounts = new HashMap<>(); // guarded by synchronizing on map itself
+ private final ExecutorService searchExecutor;
+ private final RepositoryConfiguration repoConfig;
+ public SimpleIndexManager(final RepositoryConfiguration repoConfig) {
+ this.repoConfig = repoConfig;
+ this.searchExecutor = Executors.newFixedThreadPool(repoConfig.getQueryThreadPoolSize(), new NamedThreadFactory("Search Lucene Index"));
+ }
@Override
public void close() throws IOException {
logger.debug("Shutting down SimpleIndexManager search executor");
- this.searchExecutor.shutdown();
+
+ searchExecutor.shutdown();
try {
- if (!this.searchExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
- this.searchExecutor.shutdownNow();
+ if (!searchExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
+ searchExecutor.shutdownNow();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- this.searchExecutor.shutdownNow();
+ searchExecutor.shutdownNow();
}
}
@Override
- public IndexSearcher borrowIndexSearcher(final File indexDir) throws IOException {
- logger.debug("Creating index searcher for {}", indexDir);
- final Directory directory = FSDirectory.open(indexDir);
- final DirectoryReader directoryReader = DirectoryReader.open(directory);
- final IndexSearcher searcher = new IndexSearcher(directoryReader, this.searchExecutor);
+ public EventIndexSearcher borrowIndexSearcher(final File indexDir) throws IOException {
+ final File absoluteFile = indexDir.getAbsoluteFile();
+
+ final IndexWriterCount writerCount;
+ synchronized (writerCounts) {
+ writerCount = writerCounts.remove(absoluteFile);
- final List<Closeable> closeableList = new ArrayList<>(2);
- closeableList.add(directoryReader);
- closeableList.add(directory);
- closeables.put(searcher, closeableList);
- logger.debug("Created index searcher {} for {}", searcher, indexDir);
+ if (writerCount != null) {
+ // Increment writer count and create an Index Searcher based on the writer
+ writerCounts.put(absoluteFile, new IndexWriterCount(writerCount.getWriter(), writerCount.getAnalyzer(),
+ writerCount.getDirectory(), writerCount.getCount() + 1, writerCount.isCloseableWhenUnused()));
+ }
+ }
+
+ final DirectoryReader directoryReader;
+ if (writerCount == null) {
+ logger.trace("Creating index searcher for {}", indexDir);
+ final Directory directory = FSDirectory.open(indexDir);
+ directoryReader = DirectoryReader.open(directory);
+ } else {
+ final EventIndexWriter eventIndexWriter = writerCount.getWriter();
+ directoryReader = DirectoryReader.open(eventIndexWriter.getIndexWriter(), false);
+ }
- return searcher;
+ final IndexSearcher searcher = new IndexSearcher(directoryReader, this.searchExecutor);
+
+ logger.trace("Created index searcher {} for {}", searcher, indexDir);
+ return new LuceneEventIndexSearcher(searcher, indexDir, null, directoryReader);
}
@Override
- public void returnIndexSearcher(final File indexDirectory, final IndexSearcher searcher) {
+ public void returnIndexSearcher(final EventIndexSearcher searcher) {
+ final File indexDirectory = searcher.getIndexDirectory();
logger.debug("Closing index searcher {} for {}", searcher, indexDirectory);
+ closeQuietly(searcher);
+ logger.debug("Closed index searcher {}", searcher);
+
+ final IndexWriterCount count;
+ boolean closeWriter = false;
+ synchronized (writerCounts) {
+ final File absoluteFile = searcher.getIndexDirectory().getAbsoluteFile();
+ count = writerCounts.get(absoluteFile);
+ if (count == null) {
+ logger.debug("Returning EventIndexSearcher for {}; there is no active writer for this searcher so will not decrement writerCounts", absoluteFile);
+ return;
+ }
- final List<Closeable> closeableList = closeables.get(searcher);
- if (closeableList != null) {
- for (final Closeable closeable : closeableList) {
- closeQuietly(closeable);
+ if (count.getCount() <= 1) {
+ // we are finished with this writer.
+ final boolean close = count.isCloseableWhenUnused();
+ logger.debug("Decrementing count for Index Writer for {} to {}{}", indexDirectory, count.getCount() - 1, close ? "; closing writer" : "");
+
+ if (close) {
+ writerCounts.remove(absoluteFile);
+ closeWriter = true;
+ } else {
+ writerCounts.put(absoluteFile, new IndexWriterCount(count.getWriter(), count.getAnalyzer(), count.getDirectory(),
+ count.getCount() - 1, count.isCloseableWhenUnused()));
+ }
+ } else {
+ writerCounts.put(absoluteFile, new IndexWriterCount(count.getWriter(), count.getAnalyzer(), count.getDirectory(),
+ count.getCount() - 1, count.isCloseableWhenUnused()));
}
}
- logger.debug("Closed index searcher {}", searcher);
+ if (closeWriter) {
+ try {
+ close(count);
+ } catch (final Exception e) {
+ logger.warn("Failed to close Index Writer {} due to {}", count.getWriter(), e.toString(), e);
+ }
+ }
}
@Override
- public void removeIndex(final File indexDirectory) {
+ public boolean removeIndex(final File indexDirectory) {
+ final File absoluteFile = indexDirectory.getAbsoluteFile();
+ logger.debug("Attempting to remove index {} from SimpleIndexManager", absoluteFile);
+
+ IndexWriterCount writerCount;
+ synchronized (writerCounts) {
+ writerCount = writerCounts.remove(absoluteFile);
+ if (writerCount == null) {
+ logger.debug("Allowing removal of index {} because there is no IndexWriterCount for this directory", absoluteFile);
+ return true; // return true since directory has no writers
+ }
+
+ if (writerCount.getCount() > 0) {
+ logger.debug("Not allowing removal of index {} because the active writer count for this directory is {}", absoluteFile, writerCount.getCount());
+ writerCounts.put(absoluteFile, writerCount);
+ return false;
+ }
+ }
+
+ try {
+ logger.debug("Removing index {} from SimpleIndexManager and closing the writer", absoluteFile);
+
+ close(writerCount);
+ } catch (final Exception e) {
+ logger.error("Failed to close Index Writer for {} while removing Index from the repository;"
+ + "this directory may need to be cleaned up manually.", e);
+ }
+
+ return true;
}
- @Override
- public synchronized IndexWriter borrowIndexWriter(final File indexingDirectory) throws IOException {
- final File absoluteFile = indexingDirectory.getAbsoluteFile();
- logger.trace("Borrowing index writer for {}", indexingDirectory);
+ private IndexWriterCount createWriter(final File indexDirectory) throws IOException {
+ final List<Closeable> closeables = new ArrayList<>();
+ final Directory directory = FSDirectory.open(indexDirectory);
+ closeables.add(directory);
- IndexWriterCount writerCount = writerCounts.remove(absoluteFile);
- if (writerCount == null) {
- final List<Closeable> closeables = new ArrayList<>();
- final Directory directory = FSDirectory.open(indexingDirectory);
- closeables.add(directory);
+ try {
+ final Analyzer analyzer = new StandardAnalyzer();
+ closeables.add(analyzer);
- try {
- final Analyzer analyzer = new StandardAnalyzer();
- closeables.add(analyzer);
-
- final IndexWriterConfig config = new IndexWriterConfig(LuceneUtil.LUCENE_VERSION, analyzer);
- 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);
- } catch (final IOException ioe) {
- for (final Closeable closeable : closeables) {
- try {
- closeable.close();
- } catch (final IOException ioe2) {
- ioe.addSuppressed(ioe2);
- }
+ final IndexWriterConfig config = new IndexWriterConfig(LuceneUtil.LUCENE_VERSION, analyzer);
+
+ final ConcurrentMergeScheduler mergeScheduler = new ConcurrentMergeScheduler();
+ final int mergeThreads = repoConfig.getConcurrentMergeThreads();
+ mergeScheduler.setMaxMergesAndThreads(mergeThreads, mergeThreads);
+ config.setMergeScheduler(mergeScheduler);
+
+ final IndexWriter indexWriter = new IndexWriter(directory, config);
+ final EventIndexWriter eventIndexWriter = new LuceneEventIndexWriter(indexWriter, indexDirectory);
+
+ final IndexWriterCount writerCount = new IndexWriterCount(eventIndexWriter, analyzer, directory, 1, false);
+ logger.debug("Providing new index writer for {}", indexDirectory);
+ return writerCount;
+ } catch (final IOException ioe) {
+ for (final Closeable closeable : closeables) {
+ try {
+ closeable.close();
+ } catch (final IOException ioe2) {
+ ioe.addSuppressed(ioe2);
}
+ }
+
+ throw ioe;
+ }
+ }
- throw ioe;
+ @Override
+ public EventIndexWriter borrowIndexWriter(final File indexDirectory) throws IOException {
+ final File absoluteFile = indexDirectory.getAbsoluteFile();
+ logger.trace("Borrowing index writer for {}", indexDirectory);
+
+ IndexWriterCount writerCount = null;
+ synchronized (writerCounts) {
+ writerCount = writerCounts.get(absoluteFile);
+
+ if (writerCount == null) {
+ writerCount = createWriter(indexDirectory);
+ writerCounts.put(absoluteFile, writerCount);
+ } else {
+ logger.trace("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.isCloseableWhenUnused()));
}
- writerCounts.put(absoluteFile, writerCount);
- } else {
- logger.debug("Providing existing index writer for {} and incrementing count to {}", indexingDirectory, writerCount.getCount() + 1);
- writerCounts.put(absoluteFile, new IndexWriterCount(writerCount.getWriter(),
- writerCount.getAnalyzer(), writerCount.getDirectory(), writerCount.getCount() + 1));
+ if (writerCounts.size() > repoConfig.getStorageDirectories().size() * 2) {
+ logger.debug("Index Writer returned; writer count map now has size {}; writerCount = {}; full writerCounts map = {}",
+ writerCounts.size(), writerCount, writerCounts);
+ }
}
return writerCount.getWriter();
}
+ @Override
+ public void returnIndexWriter(final EventIndexWriter writer) {
+ returnIndexWriter(writer, true, true);
+ }
@Override
- public synchronized void returnIndexWriter(final File indexingDirectory, final IndexWriter writer) {
- final File absoluteFile = indexingDirectory.getAbsoluteFile();
- logger.trace("Returning Index Writer for {} to IndexManager", indexingDirectory);
+ 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);
+
+ boolean unused = false;
+ IndexWriterCount count = null;
+ boolean close = isCloseable;
+ try {
+ synchronized (writerCounts) {
+ count = writerCounts.get(absoluteFile);
+ if (count != null && count.isCloseableWhenUnused()) {
+ close = true;
+ }
- final IndexWriterCount count = writerCounts.remove(absoluteFile);
+ 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, indexDirectory);
+ writer.close();
+ } else if (count.getCount() <= 1) {
+ // we are finished with this writer.
+ unused = true;
+ if (close) {
+ logger.debug("Decrementing count for Index Writer for {} to {}; closing writer", indexDirectory, count.getCount() - 1);
+ writerCounts.remove(absoluteFile);
+ } else {
+ logger.trace("Decrementing count for Index Writer for {} to {}", indexDirectory, count.getCount() - 1);
+
+ // If writer is not closeable, then we need to decrement its count.
+ writerCounts.put(absoluteFile, new IndexWriterCount(count.getWriter(), count.getAnalyzer(), count.getDirectory(),
+ count.getCount() - 1, close));
+ }
+ } else {
+ // decrement the count.
+ if (close) {
+ logger.debug("Decrementing count for Index Writer for {} to {} and marking as closeable when no longer in use", indexDirectory, count.getCount() - 1);
+ } else {
+ logger.trace("Decrementing count for Index Writer for {} to {}", indexDirectory, count.getCount() - 1);
+ }
- try {
- 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);
- writer.close();
- } 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);
+ writerCounts.put(absoluteFile, new IndexWriterCount(count.getWriter(), count.getAnalyzer(),
+ count.getDirectory(), count.getCount() - 1, close));
+ }
+
+ if (writerCounts.size() > repoConfig.getStorageDirectories().size() * 2) {
+ logger.debug("Index Writer returned; writer count map now has size {}; writer = {}, commit = {}, isCloseable = {}, writerCount = {}; full writerCounts Map = {}",
+ writerCounts.size(), writer, commit, isCloseable, count, writerCounts);
+ }
+ }
+
+ // Committing and closing are very expensive, so we want to do those outside of the synchronized block.
+ // So we use an 'unused' variable to tell us whether or not we should actually do so.
+ if (unused) {
try {
- writer.commit();
+ if (commit) {
+ writer.commit();
+ }
} finally {
- count.close();
+ if (close) {
+ logger.info("Index Writer for {} has been returned to Index Manager and is no longer in use. Closing Index Writer", indexDirectory);
+ close(count);
+ }
}
- } else {
- // decrement the count.
- logger.debug("Decrementing count for Index Writer for {} to {}", indexingDirectory, 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()) {
- logger.warn("", ioe);
}
+ } catch (final Exception e) {
+ logger.warn("Failed to close Index Writer {} due to {}", writer, e.toString(), e);
+ }
+ }
+
+ // This method exists solely for unit testing purposes.
+ protected void close(final IndexWriterCount count) throws IOException {
+ count.close();
+ }
+
+ protected int getWriterCount() {
+ synchronized (writerCounts) {
+ return writerCounts.size();
}
}
@@ -191,17 +342,23 @@ public class SimpleIndexManager implements IndexManager {
}
- private static class IndexWriterCount implements Closeable {
- private final IndexWriter writer;
+ protected static class IndexWriterCount implements Closeable {
+ private final EventIndexWriter writer;
private final Analyzer analyzer;
private final Directory directory;
private final int count;
+ private final boolean closeableWhenUnused;
- 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, final boolean closeableWhenUnused) {
this.writer = writer;
this.analyzer = analyzer;
this.directory = directory;
this.count = count;
+ this.closeableWhenUnused = closeableWhenUnused;
+ }
+
+ public boolean isCloseableWhenUnused() {
+ return closeableWhenUnused;
}
public Analyzer getAnalyzer() {
@@ -212,7 +369,7 @@ public class SimpleIndexManager implements IndexManager {
return directory;
}
- public IndexWriter getWriter() {
+ public EventIndexWriter getWriter() {
return writer;
}
@@ -224,5 +381,10 @@ public class SimpleIndexManager implements IndexManager {
public void close() throws IOException {
closeQuietly(writer, analyzer, directory);
}
+
+ @Override
+ public String toString() {
+ return "IndexWriterCount[count=" + count + ", writer=" + writer + ", closeableWhenUnused=" + closeableWhenUnused + "]";
+ }
}
}
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/schema/EventFieldNames.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/schema/EventFieldNames.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/schema/EventFieldNames.java
new file mode 100644
index 0000000..d6f50dd
--- /dev/null
+++ b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/schema/EventFieldNames.java
@@ -0,0 +1,59 @@
+/*
+ * 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.schema;
+
+public class EventFieldNames {
+ public static final String EVENT_IDENTIFIER = "Event ID";
+ public static final String EVENT_TYPE = "Event Type";
+ public static final String EVENT_TIME = "Event Time";
+ public static final String FLOWFILE_ENTRY_DATE = "FlowFile Entry Date";
+ public static final String EVENT_DURATION = "Event Duration";
+ public static final String LINEAGE_START_DATE = "Lineage Start Date";
+ public static final String COMPONENT_ID = "Component ID";
+ public static final String COMPONENT_TYPE = "Component Type";
+ public static final String FLOWFILE_UUID = "FlowFile UUID";
+ public static final String EVENT_DETAILS = "Event Details";
+ public static final String SOURCE_QUEUE_IDENTIFIER = "Source Queue Identifier";
+ public static final String CONTENT_CLAIM = "Content Claim";
+ public static final String PREVIOUS_CONTENT_CLAIM = "Previous Content Claim";
+ public static final String EXPLICIT_CURRENT_CONTENT_CLAIM = "Full Current Content Claim";
+ public static final String PARENT_UUIDS = "Parent UUIDs";
+ public static final String CHILD_UUIDS = "Child UUIDs";
+
+ public static final String ATTRIBUTE_NAME = "Attribute Name";
+ public static final String ATTRIBUTE_VALUE = "Attribute Value";
+ public static final String PREVIOUS_ATTRIBUTES = "Previous Attributes";
+ public static final String UPDATED_ATTRIBUTES = "Updated Attributes";
+
+ public static final String CONTENT_CLAIM_CONTAINER = "Content Claim Container";
+ public static final String CONTENT_CLAIM_SECTION = "Content Claim Section";
+ public static final String CONTENT_CLAIM_IDENTIFIER = "Content Claim Identifier";
+ public static final String CONTENT_CLAIM_OFFSET = "Content Claim Offset";
+ public static final String CONTENT_CLAIM_SIZE = "Content Claim Size";
+
+ public static final String TRANSIT_URI = "Transit URI";
+ public static final String SOURCE_SYSTEM_FLOWFILE_IDENTIFIER = "Source System FlowFile Identifier";
+ public static final String ALTERNATE_IDENTIFIER = "Alternate Identifier";
+ public static final String RELATIONSHIP = "Relationship";
+
+ // For Lookup Tables
+ public static final String NO_VALUE = "No Value";
+ public static final String EXPLICIT_VALUE = "Explicit Value";
+ public static final String LOOKUP_VALUE = "Lookup Value";
+ public static final String UNCHANGED_VALUE = "Unchanged";
+}
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/schema/EventIdFirstHeaderSchema.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/schema/EventIdFirstHeaderSchema.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/schema/EventIdFirstHeaderSchema.java
new file mode 100644
index 0000000..1c35c5a
--- /dev/null
+++ b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/schema/EventIdFirstHeaderSchema.java
@@ -0,0 +1,52 @@
+/*
+ * 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.schema;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.nifi.repository.schema.FieldType;
+import org.apache.nifi.repository.schema.RecordField;
+import org.apache.nifi.repository.schema.RecordSchema;
+import org.apache.nifi.repository.schema.Repetition;
+import org.apache.nifi.repository.schema.SimpleRecordField;
+
+public class EventIdFirstHeaderSchema {
+
+ public static RecordSchema SCHEMA = buildSchema();
+
+ public static final class FieldNames {
+ public static final String FIRST_EVENT_ID = "First Event ID";
+ public static final String TIMESTAMP_OFFSET = "Timestamp Offset";
+ public static final String COMPONENT_IDS = "Component Identifiers";
+ public static final String COMPONENT_TYPES = "Component Types";
+ public static final String QUEUE_IDS = "Queue Identifiers";
+ public static final String EVENT_TYPES = "Event Types";
+ }
+
+ private static RecordSchema buildSchema() {
+ final List<RecordField> fields = new ArrayList<>();
+ fields.add(new SimpleRecordField(FieldNames.FIRST_EVENT_ID, FieldType.LONG, Repetition.EXACTLY_ONE));
+ fields.add(new SimpleRecordField(FieldNames.TIMESTAMP_OFFSET, FieldType.LONG, Repetition.EXACTLY_ONE));
+ fields.add(new SimpleRecordField(FieldNames.COMPONENT_IDS, FieldType.STRING, Repetition.ZERO_OR_MORE));
+ fields.add(new SimpleRecordField(FieldNames.COMPONENT_TYPES, FieldType.STRING, Repetition.ZERO_OR_MORE));
+ fields.add(new SimpleRecordField(FieldNames.QUEUE_IDS, FieldType.STRING, Repetition.ZERO_OR_MORE));
+ fields.add(new SimpleRecordField(FieldNames.EVENT_TYPES, FieldType.STRING, Repetition.ZERO_OR_MORE));
+ return new RecordSchema(fields);
+ }
+}
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/schema/EventRecord.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/schema/EventRecord.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/schema/EventRecord.java
index c9e7dc8..8c82b11 100644
--- a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/schema/EventRecord.java
+++ b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/schema/EventRecord.java
@@ -69,47 +69,47 @@ public class EventRecord implements Record {
@Override
public Object getFieldValue(final String fieldName) {
switch (fieldName) {
- case EventRecordFields.Names.EVENT_IDENTIFIER:
+ case EventFieldNames.EVENT_IDENTIFIER:
return eventId;
- case EventRecordFields.Names.ALTERNATE_IDENTIFIER:
+ case EventFieldNames.ALTERNATE_IDENTIFIER:
return event.getAlternateIdentifierUri();
- case EventRecordFields.Names.CHILD_UUIDS:
+ case EventFieldNames.CHILD_UUIDS:
return event.getChildUuids();
- case EventRecordFields.Names.COMPONENT_ID:
+ case EventFieldNames.COMPONENT_ID:
return event.getComponentId();
- case EventRecordFields.Names.COMPONENT_TYPE:
+ case EventFieldNames.COMPONENT_TYPE:
return event.getComponentType();
- case EventRecordFields.Names.CONTENT_CLAIM:
+ case EventFieldNames.CONTENT_CLAIM:
return contentClaimRecord;
- case EventRecordFields.Names.EVENT_DETAILS:
+ case EventFieldNames.EVENT_DETAILS:
return event.getDetails();
- case EventRecordFields.Names.EVENT_DURATION:
+ case EventFieldNames.EVENT_DURATION:
return event.getEventDuration();
- case EventRecordFields.Names.EVENT_TIME:
+ case EventFieldNames.EVENT_TIME:
return event.getEventTime();
- case EventRecordFields.Names.EVENT_TYPE:
+ case EventFieldNames.EVENT_TYPE:
return event.getEventType().name();
- case EventRecordFields.Names.FLOWFILE_ENTRY_DATE:
+ case EventFieldNames.FLOWFILE_ENTRY_DATE:
return event.getFlowFileEntryDate();
- case EventRecordFields.Names.FLOWFILE_UUID:
+ case EventFieldNames.FLOWFILE_UUID:
return event.getFlowFileUuid();
- case EventRecordFields.Names.LINEAGE_START_DATE:
+ case EventFieldNames.LINEAGE_START_DATE:
return event.getLineageStartDate();
- case EventRecordFields.Names.PARENT_UUIDS:
+ case EventFieldNames.PARENT_UUIDS:
return event.getParentUuids();
- case EventRecordFields.Names.PREVIOUS_ATTRIBUTES:
+ case EventFieldNames.PREVIOUS_ATTRIBUTES:
return event.getPreviousAttributes();
- case EventRecordFields.Names.PREVIOUS_CONTENT_CLAIM:
+ case EventFieldNames.PREVIOUS_CONTENT_CLAIM:
return previousClaimRecord;
- case EventRecordFields.Names.RELATIONSHIP:
+ case EventFieldNames.RELATIONSHIP:
return event.getRelationship();
- case EventRecordFields.Names.SOURCE_QUEUE_IDENTIFIER:
+ case EventFieldNames.SOURCE_QUEUE_IDENTIFIER:
return event.getSourceQueueIdentifier();
- case EventRecordFields.Names.SOURCE_SYSTEM_FLOWFILE_IDENTIFIER:
+ case EventFieldNames.SOURCE_SYSTEM_FLOWFILE_IDENTIFIER:
return event.getSourceSystemFlowFileIdentifier();
- case EventRecordFields.Names.TRANSIT_URI:
+ case EventFieldNames.TRANSIT_URI:
return event.getTransitUri();
- case EventRecordFields.Names.UPDATED_ATTRIBUTES:
+ case EventFieldNames.UPDATED_ATTRIBUTES:
return event.getUpdatedAttributes();
}
@@ -119,48 +119,52 @@ public class EventRecord implements Record {
@SuppressWarnings("unchecked")
public static StandardProvenanceEventRecord getEvent(final Record record, final String storageFilename, final long storageByteOffset, final int maxAttributeLength) {
final StandardProvenanceEventRecord.Builder builder = new StandardProvenanceEventRecord.Builder();
- builder.setAlternateIdentifierUri((String) record.getFieldValue(EventRecordFields.Names.ALTERNATE_IDENTIFIER));
- builder.setChildUuids((List<String>) record.getFieldValue(EventRecordFields.Names.CHILD_UUIDS));
- builder.setComponentId((String) record.getFieldValue(EventRecordFields.Names.COMPONENT_ID));
- builder.setComponentType((String) record.getFieldValue(EventRecordFields.Names.COMPONENT_TYPE));
- builder.setDetails((String) record.getFieldValue(EventRecordFields.Names.EVENT_DETAILS));
- builder.setEventDuration((Long) record.getFieldValue(EventRecordFields.Names.EVENT_DURATION));
- builder.setEventTime((Long) record.getFieldValue(EventRecordFields.Names.EVENT_TIME));
- builder.setEventType(ProvenanceEventType.valueOf((String) record.getFieldValue(EventRecordFields.Names.EVENT_TYPE)));
- builder.setFlowFileEntryDate((Long) record.getFieldValue(EventRecordFields.Names.FLOWFILE_ENTRY_DATE));
- builder.setFlowFileUUID((String) record.getFieldValue(EventRecordFields.Names.FLOWFILE_UUID));
- builder.setLineageStartDate((Long) record.getFieldValue(EventRecordFields.Names.LINEAGE_START_DATE));
- builder.setParentUuids((List<String>) record.getFieldValue(EventRecordFields.Names.PARENT_UUIDS));
- builder.setPreviousAttributes(truncateAttributes((Map<String, String>) record.getFieldValue(EventRecordFields.Names.PREVIOUS_ATTRIBUTES), maxAttributeLength));
- builder.setEventId((Long) record.getFieldValue(EventRecordFields.Names.EVENT_IDENTIFIER));
- builder.setRelationship((String) record.getFieldValue(EventRecordFields.Names.RELATIONSHIP));
- builder.setSourceQueueIdentifier((String) record.getFieldValue(EventRecordFields.Names.SOURCE_QUEUE_IDENTIFIER));
- builder.setSourceSystemFlowFileIdentifier((String) record.getFieldValue(EventRecordFields.Names.SOURCE_SYSTEM_FLOWFILE_IDENTIFIER));
- builder.setTransitUri((String) record.getFieldValue(EventRecordFields.Names.TRANSIT_URI));
- builder.setUpdatedAttributes(truncateAttributes((Map<String, String>) record.getFieldValue(EventRecordFields.Names.UPDATED_ATTRIBUTES), maxAttributeLength));
+ builder.setAlternateIdentifierUri((String) record.getFieldValue(EventFieldNames.ALTERNATE_IDENTIFIER));
+ builder.setChildUuids((List<String>) record.getFieldValue(EventFieldNames.CHILD_UUIDS));
+ builder.setComponentId((String) record.getFieldValue(EventFieldNames.COMPONENT_ID));
+ builder.setComponentType((String) record.getFieldValue(EventFieldNames.COMPONENT_TYPE));
+ builder.setDetails((String) record.getFieldValue(EventFieldNames.EVENT_DETAILS));
+ builder.setEventDuration((Long) record.getFieldValue(EventFieldNames.EVENT_DURATION));
+ builder.setEventTime((Long) record.getFieldValue(EventFieldNames.EVENT_TIME));
+ builder.setEventType(ProvenanceEventType.valueOf((String) record.getFieldValue(EventFieldNames.EVENT_TYPE)));
+ builder.setFlowFileEntryDate((Long) record.getFieldValue(EventFieldNames.FLOWFILE_ENTRY_DATE));
+ builder.setFlowFileUUID((String) record.getFieldValue(EventFieldNames.FLOWFILE_UUID));
+ builder.setLineageStartDate((Long) record.getFieldValue(EventFieldNames.LINEAGE_START_DATE));
+ builder.setParentUuids((List<String>) record.getFieldValue(EventFieldNames.PARENT_UUIDS));
+ builder.setPreviousAttributes(truncateAttributes((Map<String, String>) record.getFieldValue(EventFieldNames.PREVIOUS_ATTRIBUTES), maxAttributeLength));
+ builder.setRelationship((String) record.getFieldValue(EventFieldNames.RELATIONSHIP));
+ builder.setSourceQueueIdentifier((String) record.getFieldValue(EventFieldNames.SOURCE_QUEUE_IDENTIFIER));
+ builder.setSourceSystemFlowFileIdentifier((String) record.getFieldValue(EventFieldNames.SOURCE_SYSTEM_FLOWFILE_IDENTIFIER));
+ builder.setTransitUri((String) record.getFieldValue(EventFieldNames.TRANSIT_URI));
+ builder.setUpdatedAttributes(truncateAttributes((Map<String, String>) record.getFieldValue(EventFieldNames.UPDATED_ATTRIBUTES), maxAttributeLength));
+
+ final Long eventId = (Long) record.getFieldValue(EventFieldNames.EVENT_IDENTIFIER);
+ if (eventId != null) {
+ builder.setEventId(eventId);
+ }
builder.setStorageLocation(storageFilename, storageByteOffset);
- final Record currentClaimRecord = (Record) record.getFieldValue(EventRecordFields.Names.CONTENT_CLAIM);
+ final Record currentClaimRecord = (Record) record.getFieldValue(EventFieldNames.CONTENT_CLAIM);
if (currentClaimRecord == null) {
builder.setCurrentContentClaim(null, null, null, null, 0L);
} else {
builder.setCurrentContentClaim(
- (String) currentClaimRecord.getFieldValue(EventRecordFields.Names.CONTENT_CLAIM_CONTAINER),
- (String) currentClaimRecord.getFieldValue(EventRecordFields.Names.CONTENT_CLAIM_SECTION),
- (String) currentClaimRecord.getFieldValue(EventRecordFields.Names.CONTENT_CLAIM_IDENTIFIER),
- (Long) currentClaimRecord.getFieldValue(EventRecordFields.Names.CONTENT_CLAIM_OFFSET),
- (Long) currentClaimRecord.getFieldValue(EventRecordFields.Names.CONTENT_CLAIM_SIZE));
+ (String) currentClaimRecord.getFieldValue(EventFieldNames.CONTENT_CLAIM_CONTAINER),
+ (String) currentClaimRecord.getFieldValue(EventFieldNames.CONTENT_CLAIM_SECTION),
+ (String) currentClaimRecord.getFieldValue(EventFieldNames.CONTENT_CLAIM_IDENTIFIER),
+ (Long) currentClaimRecord.getFieldValue(EventFieldNames.CONTENT_CLAIM_OFFSET),
+ (Long) currentClaimRecord.getFieldValue(EventFieldNames.CONTENT_CLAIM_SIZE));
}
- final Record previousClaimRecord = (Record) record.getFieldValue(EventRecordFields.Names.PREVIOUS_CONTENT_CLAIM);
+ final Record previousClaimRecord = (Record) record.getFieldValue(EventFieldNames.PREVIOUS_CONTENT_CLAIM);
if (previousClaimRecord != null) {
builder.setPreviousContentClaim(
- (String) previousClaimRecord.getFieldValue(EventRecordFields.Names.CONTENT_CLAIM_CONTAINER),
- (String) previousClaimRecord.getFieldValue(EventRecordFields.Names.CONTENT_CLAIM_SECTION),
- (String) previousClaimRecord.getFieldValue(EventRecordFields.Names.CONTENT_CLAIM_IDENTIFIER),
- (Long) previousClaimRecord.getFieldValue(EventRecordFields.Names.CONTENT_CLAIM_OFFSET),
- (Long) previousClaimRecord.getFieldValue(EventRecordFields.Names.CONTENT_CLAIM_SIZE));
+ (String) previousClaimRecord.getFieldValue(EventFieldNames.CONTENT_CLAIM_CONTAINER),
+ (String) previousClaimRecord.getFieldValue(EventFieldNames.CONTENT_CLAIM_SECTION),
+ (String) previousClaimRecord.getFieldValue(EventFieldNames.CONTENT_CLAIM_IDENTIFIER),
+ (Long) previousClaimRecord.getFieldValue(EventFieldNames.CONTENT_CLAIM_OFFSET),
+ (Long) previousClaimRecord.getFieldValue(EventFieldNames.CONTENT_CLAIM_SIZE));
}
return builder.build();
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/schema/EventRecordFields.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/schema/EventRecordFields.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/schema/EventRecordFields.java
index 0582dd8..3d79ab4 100644
--- a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/schema/EventRecordFields.java
+++ b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/schema/EventRecordFields.java
@@ -29,82 +29,48 @@ import org.apache.nifi.repository.schema.SimpleRecordField;
public class EventRecordFields {
- public static class Names {
- public static final String EVENT_IDENTIFIER = "Event ID";
- public static final String EVENT_TYPE = "Event Type";
- public static final String EVENT_TIME = "Event Time";
- public static final String FLOWFILE_ENTRY_DATE = "FlowFile Entry Date";
- public static final String EVENT_DURATION = "Event Duration";
- public static final String LINEAGE_START_DATE = "Lineage Start Date";
- public static final String COMPONENT_ID = "Component ID";
- public static final String COMPONENT_TYPE = "Component Type";
- public static final String FLOWFILE_UUID = "FlowFile UUID";
- public static final String EVENT_DETAILS = "Event Details";
- public static final String SOURCE_QUEUE_IDENTIFIER = "Source Queue Identifier";
- public static final String CONTENT_CLAIM = "Content Claim";
- public static final String PREVIOUS_CONTENT_CLAIM = "Previous Content Claim";
- public static final String PARENT_UUIDS = "Parent UUIDs";
- public static final String CHILD_UUIDS = "Child UUIDs";
-
- public static final String ATTRIBUTE_NAME = "Attribute Name";
- public static final String ATTRIBUTE_VALUE = "Attribute Value";
- public static final String PREVIOUS_ATTRIBUTES = "Previous Attributes";
- public static final String UPDATED_ATTRIBUTES = "Updated Attributes";
-
- public static final String CONTENT_CLAIM_CONTAINER = "Content Claim Container";
- public static final String CONTENT_CLAIM_SECTION = "Content Claim Section";
- public static final String CONTENT_CLAIM_IDENTIFIER = "Content Claim Identifier";
- public static final String CONTENT_CLAIM_OFFSET = "Content Claim Offset";
- public static final String CONTENT_CLAIM_SIZE = "Content Claim Size";
-
- public static final String TRANSIT_URI = "Transit URI";
- public static final String SOURCE_SYSTEM_FLOWFILE_IDENTIFIER = "Source System FlowFile Identifier";
- public static final String ALTERNATE_IDENTIFIER = "Alternate Identifier";
- public static final String RELATIONSHIP = "Relationship";
- }
-
// General Event fields.
- public static final RecordField RECORD_IDENTIFIER = new SimpleRecordField(Names.EVENT_IDENTIFIER, FieldType.LONG, EXACTLY_ONE);
- public static final RecordField EVENT_TYPE = new SimpleRecordField(Names.EVENT_TYPE, FieldType.STRING, EXACTLY_ONE);
- public static final RecordField EVENT_TIME = new SimpleRecordField(Names.EVENT_TIME, FieldType.LONG, EXACTLY_ONE);
- public static final RecordField FLOWFILE_ENTRY_DATE = new SimpleRecordField(Names.FLOWFILE_ENTRY_DATE, FieldType.LONG, EXACTLY_ONE);
- public static final RecordField EVENT_DURATION = new SimpleRecordField(Names.EVENT_DURATION, FieldType.LONG, EXACTLY_ONE);
- public static final RecordField LINEAGE_START_DATE = new SimpleRecordField(Names.LINEAGE_START_DATE, FieldType.LONG, EXACTLY_ONE);
- public static final RecordField COMPONENT_ID = new SimpleRecordField(Names.COMPONENT_ID, FieldType.STRING, ZERO_OR_ONE);
- public static final RecordField COMPONENT_TYPE = new SimpleRecordField(Names.COMPONENT_TYPE, FieldType.STRING, ZERO_OR_ONE);
- public static final RecordField FLOWFILE_UUID = new SimpleRecordField(Names.FLOWFILE_UUID, FieldType.STRING, EXACTLY_ONE);
- public static final RecordField EVENT_DETAILS = new SimpleRecordField(Names.EVENT_DETAILS, FieldType.STRING, ZERO_OR_ONE);
- public static final RecordField SOURCE_QUEUE_IDENTIFIER = new SimpleRecordField(Names.SOURCE_QUEUE_IDENTIFIER, FieldType.STRING, ZERO_OR_ONE);
+ public static final RecordField RECORD_IDENTIFIER = new SimpleRecordField(EventFieldNames.EVENT_IDENTIFIER, FieldType.LONG, EXACTLY_ONE);
+ public static final RecordField EVENT_TYPE = new SimpleRecordField(EventFieldNames.EVENT_TYPE, FieldType.STRING, EXACTLY_ONE);
+ public static final RecordField EVENT_TIME = new SimpleRecordField(EventFieldNames.EVENT_TIME, FieldType.LONG, EXACTLY_ONE);
+ public static final RecordField FLOWFILE_ENTRY_DATE = new SimpleRecordField(EventFieldNames.FLOWFILE_ENTRY_DATE, FieldType.LONG, EXACTLY_ONE);
+ public static final RecordField EVENT_DURATION = new SimpleRecordField(EventFieldNames.EVENT_DURATION, FieldType.LONG, EXACTLY_ONE);
+ public static final RecordField LINEAGE_START_DATE = new SimpleRecordField(EventFieldNames.LINEAGE_START_DATE, FieldType.LONG, EXACTLY_ONE);
+ public static final RecordField COMPONENT_ID = new SimpleRecordField(EventFieldNames.COMPONENT_ID, FieldType.STRING, ZERO_OR_ONE);
+ public static final RecordField COMPONENT_TYPE = new SimpleRecordField(EventFieldNames.COMPONENT_TYPE, FieldType.STRING, ZERO_OR_ONE);
+ public static final RecordField FLOWFILE_UUID = new SimpleRecordField(EventFieldNames.FLOWFILE_UUID, FieldType.STRING, EXACTLY_ONE);
+ public static final RecordField EVENT_DETAILS = new SimpleRecordField(EventFieldNames.EVENT_DETAILS, FieldType.STRING, ZERO_OR_ONE);
+ public static final RecordField SOURCE_QUEUE_IDENTIFIER = new SimpleRecordField(EventFieldNames.SOURCE_QUEUE_IDENTIFIER, FieldType.STRING, ZERO_OR_ONE);
// Attributes
- public static final RecordField ATTRIBUTE_NAME = new SimpleRecordField(Names.ATTRIBUTE_NAME, FieldType.LONG_STRING, EXACTLY_ONE);
- public static final RecordField ATTRIBUTE_VALUE_REQUIRED = new SimpleRecordField(Names.ATTRIBUTE_VALUE, FieldType.LONG_STRING, EXACTLY_ONE);
- public static final RecordField ATTRIBUTE_VALUE_OPTIONAL = new SimpleRecordField(Names.ATTRIBUTE_VALUE, FieldType.LONG_STRING, ZERO_OR_ONE);
+ public static final RecordField ATTRIBUTE_NAME = new SimpleRecordField(EventFieldNames.ATTRIBUTE_NAME, FieldType.LONG_STRING, EXACTLY_ONE);
+ public static final RecordField ATTRIBUTE_VALUE_REQUIRED = new SimpleRecordField(EventFieldNames.ATTRIBUTE_VALUE, FieldType.LONG_STRING, EXACTLY_ONE);
+ public static final RecordField ATTRIBUTE_VALUE_OPTIONAL = new SimpleRecordField(EventFieldNames.ATTRIBUTE_VALUE, FieldType.LONG_STRING, ZERO_OR_ONE);
- public static final RecordField PREVIOUS_ATTRIBUTES = new MapRecordField(Names.PREVIOUS_ATTRIBUTES, ATTRIBUTE_NAME, ATTRIBUTE_VALUE_REQUIRED, EXACTLY_ONE);
- public static final RecordField UPDATED_ATTRIBUTES = new MapRecordField(Names.UPDATED_ATTRIBUTES, ATTRIBUTE_NAME, ATTRIBUTE_VALUE_OPTIONAL, EXACTLY_ONE);
+ public static final RecordField PREVIOUS_ATTRIBUTES = new MapRecordField(EventFieldNames.PREVIOUS_ATTRIBUTES, ATTRIBUTE_NAME, ATTRIBUTE_VALUE_REQUIRED, EXACTLY_ONE);
+ public static final RecordField UPDATED_ATTRIBUTES = new MapRecordField(EventFieldNames.UPDATED_ATTRIBUTES, ATTRIBUTE_NAME, ATTRIBUTE_VALUE_OPTIONAL, EXACTLY_ONE);
// Content Claims
- public static final RecordField CONTENT_CLAIM_CONTAINER = new SimpleRecordField(Names.CONTENT_CLAIM_CONTAINER, FieldType.STRING, EXACTLY_ONE);
- public static final RecordField CONTENT_CLAIM_SECTION = new SimpleRecordField(Names.CONTENT_CLAIM_SECTION, FieldType.STRING, EXACTLY_ONE);
- public static final RecordField CONTENT_CLAIM_IDENTIFIER = new SimpleRecordField(Names.CONTENT_CLAIM_IDENTIFIER, FieldType.STRING, EXACTLY_ONE);
- public static final RecordField CONTENT_CLAIM_OFFSET = new SimpleRecordField(Names.CONTENT_CLAIM_OFFSET, FieldType.LONG, EXACTLY_ONE);
- public static final RecordField CONTENT_CLAIM_SIZE = new SimpleRecordField(Names.CONTENT_CLAIM_SIZE, FieldType.LONG, EXACTLY_ONE);
- public static final RecordField CURRENT_CONTENT_CLAIM = new ComplexRecordField(Names.CONTENT_CLAIM, ZERO_OR_ONE,
+ public static final RecordField CONTENT_CLAIM_CONTAINER = new SimpleRecordField(EventFieldNames.CONTENT_CLAIM_CONTAINER, FieldType.STRING, EXACTLY_ONE);
+ public static final RecordField CONTENT_CLAIM_SECTION = new SimpleRecordField(EventFieldNames.CONTENT_CLAIM_SECTION, FieldType.STRING, EXACTLY_ONE);
+ public static final RecordField CONTENT_CLAIM_IDENTIFIER = new SimpleRecordField(EventFieldNames.CONTENT_CLAIM_IDENTIFIER, FieldType.STRING, EXACTLY_ONE);
+ public static final RecordField CONTENT_CLAIM_OFFSET = new SimpleRecordField(EventFieldNames.CONTENT_CLAIM_OFFSET, FieldType.LONG, EXACTLY_ONE);
+ public static final RecordField CONTENT_CLAIM_SIZE = new SimpleRecordField(EventFieldNames.CONTENT_CLAIM_SIZE, FieldType.LONG, EXACTLY_ONE);
+ public static final RecordField CURRENT_CONTENT_CLAIM = new ComplexRecordField(EventFieldNames.CONTENT_CLAIM, ZERO_OR_ONE,
CONTENT_CLAIM_CONTAINER, CONTENT_CLAIM_SECTION, CONTENT_CLAIM_IDENTIFIER, CONTENT_CLAIM_OFFSET, CONTENT_CLAIM_SIZE);
- public static final RecordField PREVIOUS_CONTENT_CLAIM = new ComplexRecordField(Names.PREVIOUS_CONTENT_CLAIM, ZERO_OR_ONE,
+ public static final RecordField PREVIOUS_CONTENT_CLAIM = new ComplexRecordField(EventFieldNames.PREVIOUS_CONTENT_CLAIM, ZERO_OR_ONE,
CONTENT_CLAIM_CONTAINER, CONTENT_CLAIM_SECTION, CONTENT_CLAIM_IDENTIFIER, CONTENT_CLAIM_OFFSET, CONTENT_CLAIM_SIZE);
// EventType-Specific fields
// for FORK, JOIN, CLONE, REPLAY
- public static final RecordField PARENT_UUIDS = new SimpleRecordField(Names.PARENT_UUIDS, FieldType.STRING, ZERO_OR_MORE);
- public static final RecordField CHILD_UUIDS = new SimpleRecordField(Names.CHILD_UUIDS, FieldType.STRING, ZERO_OR_MORE);
+ public static final RecordField PARENT_UUIDS = new SimpleRecordField(EventFieldNames.PARENT_UUIDS, FieldType.STRING, ZERO_OR_MORE);
+ public static final RecordField CHILD_UUIDS = new SimpleRecordField(EventFieldNames.CHILD_UUIDS, FieldType.STRING, ZERO_OR_MORE);
// for SEND/RECEIVE/FETCH
- public static final RecordField TRANSIT_URI = new SimpleRecordField(Names.TRANSIT_URI, FieldType.STRING, ZERO_OR_ONE);
- public static final RecordField SOURCE_SYSTEM_FLOWFILE_IDENTIFIER = new SimpleRecordField(Names.SOURCE_QUEUE_IDENTIFIER, FieldType.STRING, ZERO_OR_ONE);
+ public static final RecordField TRANSIT_URI = new SimpleRecordField(EventFieldNames.TRANSIT_URI, FieldType.STRING, ZERO_OR_ONE);
+ public static final RecordField SOURCE_SYSTEM_FLOWFILE_IDENTIFIER = new SimpleRecordField(EventFieldNames.SOURCE_SYSTEM_FLOWFILE_IDENTIFIER, FieldType.STRING, ZERO_OR_ONE);
// for ADD_INFO
- public static final RecordField ALTERNATE_IDENTIFIER = new SimpleRecordField(Names.ALTERNATE_IDENTIFIER, FieldType.STRING, ZERO_OR_ONE);
- public static final RecordField RELATIONSHIP = new SimpleRecordField(Names.RELATIONSHIP, FieldType.STRING, ZERO_OR_ONE);
+ public static final RecordField ALTERNATE_IDENTIFIER = new SimpleRecordField(EventFieldNames.ALTERNATE_IDENTIFIER, FieldType.STRING, ZERO_OR_ONE);
+ public static final RecordField RELATIONSHIP = new SimpleRecordField(EventFieldNames.RELATIONSHIP, FieldType.STRING, ZERO_OR_ONE);
}