provenanceLogFiles) throws IOException {
final File originalFile = file;
-
- if (!file.exists()) {
- if (provenanceLogFiles == null) {
- throw new FileNotFoundException(file.toString());
- }
-
- final String baseName = LuceneUtil.substringBefore(file.getName(), ".") + ".";
- for (final Path path : provenanceLogFiles) {
- if (path.toFile().getName().startsWith(baseName)) {
- file = path.toFile();
- break;
- }
- }
- }
-
InputStream fis = null;
- if ( file.exists() ) {
- try {
- fis = new FileInputStream(file);
- } catch (final FileNotFoundException fnfe) {
- fis = null;
- }
- }
-
- openStream: while ( fis == null ) {
- final File dir = file.getParentFile();
- final String baseName = LuceneUtil.substringBefore(file.getName(), ".");
-
- // depending on which rollover actions have occurred, we could have 3 possibilities for the
- // filename that we need. The majority of the time, we will use the extension ".prov.indexed.gz"
- // because most often we are compressing on rollover and most often we have already finished
- // compressing by the time that we are querying the data.
- for ( final String extension : new String[] {".indexed.prov.gz", ".indexed.prov", ".prov"} ) {
- file = new File(dir, baseName + extension);
- if ( file.exists() ) {
- try {
- fis = new FileInputStream(file);
- break openStream;
- } catch (final FileNotFoundException fnfe) {
- // file was modified by a RolloverAction after we verified that it exists but before we could
- // create an InputStream for it. Start over.
- fis = null;
- continue openStream;
- }
- }
- }
-
- break;
- }
- if ( fis == null ) {
- throw new FileNotFoundException("Unable to locate file " + originalFile);
+ try {
+ if (!file.exists()) {
+ if (provenanceLogFiles != null) {
+ final String baseName = LuceneUtil.substringBefore(file.getName(), ".") + ".";
+ for (final Path path : provenanceLogFiles) {
+ if (path.toFile().getName().startsWith(baseName)) {
+ file = path.toFile();
+ break;
+ }
+ }
+ }
+ }
+
+ if ( file.exists() ) {
+ try {
+ fis = new FileInputStream(file);
+ } catch (final FileNotFoundException fnfe) {
+ fis = null;
+ }
+ }
+
+ String filename = file.getName();
+ openStream: while ( fis == null ) {
+ final File dir = file.getParentFile();
+ final String baseName = LuceneUtil.substringBefore(file.getName(), ".");
+
+ // depending on which rollover actions have occurred, we could have 3 possibilities for the
+ // filename that we need. The majority of the time, we will use the extension ".prov.indexed.gz"
+ // because most often we are compressing on rollover and most often we have already finished
+ // compressing by the time that we are querying the data.
+ for ( final String extension : new String[] {".prov.gz", ".prov"} ) {
+ file = new File(dir, baseName + extension);
+ if ( file.exists() ) {
+ try {
+ fis = new FileInputStream(file);
+ filename = baseName + extension;
+ break openStream;
+ } catch (final FileNotFoundException fnfe) {
+ // file was modified by a RolloverAction after we verified that it exists but before we could
+ // create an InputStream for it. Start over.
+ fis = null;
+ continue openStream;
+ }
+ }
+ }
+
+ break;
+ }
+
+ if ( fis == null ) {
+ throw new FileNotFoundException("Unable to locate file " + originalFile);
+ }
+
+ final File tocFile = TocUtil.getTocFile(file);
+ if ( tocFile.exists() ) {
+ final TocReader tocReader = new StandardTocReader(tocFile);
+ return new StandardRecordReader(fis, filename, tocReader);
+ } else {
+ return new StandardRecordReader(fis, filename);
+ }
+ } catch (final IOException ioe) {
+ if ( fis != null ) {
+ try {
+ fis.close();
+ } catch (final IOException inner) {
+ ioe.addSuppressed(inner);
+ }
+ }
+
+ throw ioe;
}
- final InputStream readableStream;
- if (file.getName().endsWith(".gz")) {
- readableStream = new BufferedInputStream(new GZIPInputStream(fis));
- } else {
- readableStream = new BufferedInputStream(fis);
- }
-
- final DataInputStream dis = new DataInputStream(readableStream);
- @SuppressWarnings("unused")
- final String repoClassName = dis.readUTF();
- final int serializationVersion = dis.readInt();
-
- return new StandardRecordReader(dis, serializationVersion, file.getName());
}
}
diff --git a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/serialization/RecordWriter.java b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/serialization/RecordWriter.java
index de98ab9cc5..58f4dc281e 100644
--- a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/serialization/RecordWriter.java
+++ b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/serialization/RecordWriter.java
@@ -21,6 +21,7 @@ import java.io.File;
import java.io.IOException;
import org.apache.nifi.provenance.ProvenanceEventRecord;
+import org.apache.nifi.provenance.toc.TocWriter;
public interface RecordWriter extends Closeable {
@@ -82,4 +83,9 @@ public interface RecordWriter extends Closeable {
*/
void sync() throws IOException;
+ /**
+ * Returns the TOC Writer that is being used to write the Table of Contents for this journal
+ * @return
+ */
+ TocWriter getTocWriter();
}
diff --git a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/serialization/RecordWriters.java b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/serialization/RecordWriters.java
index 15349de18a..47b7c7e5d7 100644
--- a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/serialization/RecordWriters.java
+++ b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/serialization/RecordWriters.java
@@ -20,11 +20,20 @@ import java.io.File;
import java.io.IOException;
import org.apache.nifi.provenance.StandardRecordWriter;
+import org.apache.nifi.provenance.toc.StandardTocWriter;
+import org.apache.nifi.provenance.toc.TocUtil;
+import org.apache.nifi.provenance.toc.TocWriter;
public class RecordWriters {
+ private static final int DEFAULT_COMPRESSION_BLOCK_SIZE = 1024 * 1024; // 1 MB
- public static RecordWriter newRecordWriter(final File file) throws IOException {
- return new StandardRecordWriter(file);
+ public static RecordWriter newRecordWriter(final File file, final boolean compressed, final boolean createToc) throws IOException {
+ return newRecordWriter(file, compressed, createToc, DEFAULT_COMPRESSION_BLOCK_SIZE);
+ }
+
+ public static RecordWriter newRecordWriter(final File file, final boolean compressed, final boolean createToc, final int compressionBlockBytes) throws IOException {
+ final TocWriter tocWriter = createToc ? new StandardTocWriter(TocUtil.getTocFile(file), false, false) : null;
+ return new StandardRecordWriter(file, tocWriter, compressed, compressionBlockBytes);
}
}
diff --git a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/StandardTocReader.java b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/StandardTocReader.java
new file mode 100644
index 0000000000..8944cec298
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/StandardTocReader.java
@@ -0,0 +1,108 @@
+/*
+ * 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.toc;
+
+import java.io.DataInputStream;
+import java.io.EOFException;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+
+/**
+ * Standard implementation of TocReader.
+ *
+ * Expects .toc file to be in the following format;
+ *
+ * byte 0: version
+ * byte 1: boolean: compressionFlag -> 0 = journal is NOT compressed, 1 = journal is compressed
+ * byte 2-9: long: offset of block 0
+ * byte 10-17: long: offset of block 1
+ * ...
+ * byte (N*8+2)-(N*8+9): long: offset of block N
+ */
+public class StandardTocReader implements TocReader {
+ private final boolean compressed;
+ private final long[] offsets;
+
+ public StandardTocReader(final File file) throws IOException {
+ try (final FileInputStream fis = new FileInputStream(file);
+ final DataInputStream dis = new DataInputStream(fis)) {
+
+ final int version = dis.read();
+ if ( version < 0 ) {
+ throw new EOFException();
+ }
+
+ final int compressionFlag = dis.read();
+ if ( compressionFlag < 0 ) {
+ throw new EOFException();
+ }
+
+ if ( compressionFlag == 0 ) {
+ compressed = false;
+ } else if ( compressionFlag == 1 ) {
+ compressed = true;
+ } else {
+ throw new IOException("Table of Contents appears to be corrupt: could not read 'compression flag' from header; expected value of 0 or 1 but got " + compressionFlag);
+ }
+
+ final int numBlocks = (int) ((file.length() - 2) / 8);
+ offsets = new long[numBlocks];
+
+ for (int i=0; i < numBlocks; i++) {
+ offsets[i] = dis.readLong();
+ }
+ }
+ }
+
+ @Override
+ public boolean isCompressed() {
+ return compressed;
+ }
+
+ @Override
+ public long getBlockOffset(final int blockIndex) {
+ if ( blockIndex >= offsets.length ) {
+ return -1L;
+ }
+ return offsets[blockIndex];
+ }
+
+ @Override
+ public long getLastBlockOffset() {
+ if ( offsets.length == 0 ) {
+ return 0L;
+ }
+ return offsets[offsets.length - 1];
+ }
+
+ @Override
+ public void close() throws IOException {
+ }
+
+ @Override
+ public int getBlockIndex(final long blockOffset) {
+ for (int i=0; i < offsets.length; i++) {
+ if ( offsets[i] > blockOffset ) {
+ return i-1;
+ }
+ }
+
+ return offsets.length - 1;
+ }
+
+}
diff --git a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/StandardTocWriter.java b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/StandardTocWriter.java
new file mode 100644
index 0000000000..488f225242
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/StandardTocWriter.java
@@ -0,0 +1,120 @@
+/*
+ * 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.toc;
+
+import java.io.BufferedOutputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.nio.file.Files;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Standard implementation of {@link TocWriter}.
+ *
+ * Format of .toc file:
+ * byte 0: version
+ * byte 1: compressed: 0 -> not compressed, 1 -> compressed
+ * byte 2-9: long: offset of block 0
+ * byte 10-17: long: offset of block 1
+ * ...
+ * byte (N*8+2)-(N*8+9): long: offset of block N
+ */
+public class StandardTocWriter implements TocWriter {
+ private static final Logger logger = LoggerFactory.getLogger(StandardTocWriter.class);
+
+ public static final byte VERSION = 1;
+
+ private final File file;
+ private final FileOutputStream fos;
+ private final boolean alwaysSync;
+ private int index = -1;
+
+ /**
+ * Creates a StandardTocWriter that writes to the given file.
+ * @param file the file to write to
+ * @param compressionFlag whether or not the journal is compressed
+ * @throws FileNotFoundException
+ */
+ public StandardTocWriter(final File file, final boolean compressionFlag, final boolean alwaysSync) throws IOException {
+ final File tocDir = file.getParentFile();
+ if ( !tocDir.exists() ) {
+ Files.createDirectories(tocDir.toPath());
+ }
+
+ this.file = file;
+ fos = new FileOutputStream(file);
+ this.alwaysSync = alwaysSync;
+
+ final byte[] header = new byte[2];
+ header[0] = VERSION;
+ header[1] = (byte) (compressionFlag ? 1 : 0);
+ fos.write(header);
+ fos.flush();
+
+ if ( alwaysSync ) {
+ sync();
+ }
+ }
+
+ @Override
+ public void addBlockOffset(final long offset) throws IOException {
+ final BufferedOutputStream bos = new BufferedOutputStream(fos);
+ final DataOutputStream dos = new DataOutputStream(bos);
+ dos.writeLong(offset);
+ dos.flush();
+ index++;
+ logger.debug("Adding block {} at offset {}", index, offset);
+
+ if ( alwaysSync ) {
+ sync();
+ }
+ }
+
+ @Override
+ public void sync() throws IOException {
+ fos.getFD().sync();
+ }
+
+ @Override
+ public int getCurrentBlockIndex() {
+ return index;
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (alwaysSync) {
+ fos.getFD().sync();
+ }
+
+ fos.close();
+ }
+
+ @Override
+ public File getFile() {
+ return file;
+ }
+
+ @Override
+ public String toString() {
+ return "TOC Writer for " + file;
+ }
+}
diff --git a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/TocReader.java b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/TocReader.java
new file mode 100644
index 0000000000..7c197be9d5
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/TocReader.java
@@ -0,0 +1,58 @@
+/*
+ * 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.toc;
+
+import java.io.Closeable;
+
+/**
+ *
+ * Reads a Table of Contents (.toc file) for a corresponding Journal File. We use a Table of Contents
+ * to map a Block Index to an offset into the Journal file where that Block begins. We do this so that
+ * we can then persist a Block Index for an event and then compress the Journal later. This way, we can
+ * get good compression by compressing a large batch of events at once, and this way we can also look up
+ * an event in a Journal that has not been compressed by looking in the Table of Contents or lookup the
+ * event in a Journal post-compression by simply rewriting the TOC while we compress the data.
+ *
+ */
+public interface TocReader extends Closeable {
+
+ /**
+ * Indicates whether or not the corresponding Journal file is compressed
+ * @return
+ */
+ boolean isCompressed();
+
+ /**
+ * Returns the byte offset into the Journal File for the Block with the given index.
+ * @param blockIndex
+ * @return
+ */
+ long getBlockOffset(int blockIndex);
+
+ /**
+ * Returns the byte offset into the Journal File of the last Block in the given index
+ * @return
+ */
+ long getLastBlockOffset();
+
+ /**
+ * Returns the index of the block that contains the given offset
+ * @param blockOffset
+ * @return
+ */
+ int getBlockIndex(long blockOffset);
+}
diff --git a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/TocUtil.java b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/TocUtil.java
new file mode 100644
index 0000000000..c30ac98830
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/TocUtil.java
@@ -0,0 +1,37 @@
+/*
+ * 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.toc;
+
+import java.io.File;
+
+import org.apache.nifi.provenance.lucene.LuceneUtil;
+
+public class TocUtil {
+
+ /**
+ * Returns the file that should be used as the Table of Contents for the given Journal File
+ * @param journalFile
+ * @return
+ */
+ public static File getTocFile(final File journalFile) {
+ final File tocDir = new File(journalFile.getParentFile(), "toc");
+ final String basename = LuceneUtil.substringBefore(journalFile.getName(), ".");
+ final File tocFile = new File(tocDir, basename + ".toc");
+ return tocFile;
+ }
+
+}
diff --git a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/TocWriter.java b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/TocWriter.java
new file mode 100644
index 0000000000..c6780531c5
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/TocWriter.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.toc;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+
+/**
+ * Writes a .toc file
+ */
+public interface TocWriter extends Closeable {
+
+ /**
+ * Adds the given block offset as the next Block Offset in the Table of Contents
+ * @param offset
+ * @throws IOException
+ */
+ void addBlockOffset(long offset) throws IOException;
+
+ /**
+ * Returns the index of the current Block
+ * @return
+ */
+ int getCurrentBlockIndex();
+
+ /**
+ * Returns the file that is currently being written to
+ * @return
+ */
+ File getFile();
+
+ /**
+ * Synchronizes the data with the underlying storage device
+ * @throws IOException
+ */
+ void sync() throws IOException;
+}
diff --git a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/TestPersistentProvenanceRepository.java b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/TestPersistentProvenanceRepository.java
index 5be208bd4e..5541ab56ad 100644
--- a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/TestPersistentProvenanceRepository.java
+++ b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/TestPersistentProvenanceRepository.java
@@ -16,6 +16,7 @@
*/
package org.apache.nifi.provenance;
+import static org.apache.nifi.provenance.TestUtil.createFlowFile;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
@@ -25,14 +26,14 @@ import java.io.FileFilter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.analysis.core.SimpleAnalyzer;
@@ -45,7 +46,6 @@ import org.apache.lucene.search.ScoreDoc;
import org.apache.lucene.search.TopDocs;
import org.apache.lucene.store.FSDirectory;
import org.apache.nifi.events.EventReporter;
-import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.provenance.lineage.EventNode;
import org.apache.nifi.provenance.lineage.Lineage;
import org.apache.nifi.provenance.lineage.LineageEdge;
@@ -59,8 +59,10 @@ import org.apache.nifi.provenance.search.SearchableField;
import org.apache.nifi.provenance.serialization.RecordReader;
import org.apache.nifi.provenance.serialization.RecordReaders;
import org.apache.nifi.reporting.Severity;
+import org.apache.nifi.util.file.FileUtils;
import org.junit.After;
import org.junit.Before;
+import org.junit.BeforeClass;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
@@ -72,87 +74,64 @@ public class TestPersistentProvenanceRepository {
public TestName name = new TestName();
private PersistentProvenanceRepository repo;
+ private RepositoryConfiguration config;
public static final int DEFAULT_ROLLOVER_MILLIS = 2000;
private RepositoryConfiguration createConfiguration() {
- final RepositoryConfiguration config = new RepositoryConfiguration();
+ config = new RepositoryConfiguration();
config.addStorageDirectory(new File("target/storage/" + UUID.randomUUID().toString()));
- config.setCompressOnRollover(false);
+ config.setCompressOnRollover(true);
config.setMaxEventFileLife(2000L, TimeUnit.SECONDS);
+ config.setCompressionBlockBytes(100);
return config;
}
+ @BeforeClass
+ public static void setLogLevel() {
+ System.setProperty("org.slf4j.simpleLogger.log.org.apache.nifi.provenance", "DEBUG");
+ }
+
@Before
public void printTestName() {
System.out.println("\n\n\n*********************** " + name.getMethodName() + " *****************************");
}
@After
- public void closeRepo() {
+ public void closeRepo() throws IOException {
if (repo != null) {
try {
repo.close();
} catch (final IOException ioe) {
}
}
+
+ // Delete all of the storage files. We do this in order to clean up the tons of files that
+ // we create but also to ensure that we have closed all of the file handles. If we leave any
+ // streams open, for instance, this will throw an IOException, causing our unit test to fail.
+ for ( final File storageDir : config.getStorageDirectories() ) {
+ int i;
+ for (i=0; i < 3; i++) {
+ try {
+ FileUtils.deleteFile(storageDir, true);
+ break;
+ } catch (final IOException ioe) {
+ // if there is a virus scanner, etc. running in the background we may not be able to
+ // delete the file. Wait a sec and try again.
+ if ( i == 2 ) {
+ throw ioe;
+ } else {
+ try {
+ Thread.sleep(1000L);
+ } catch (final InterruptedException ie) {
+ }
+ }
+ }
+ }
+ }
}
- private FlowFile createFlowFile(final long id, final long fileSize, final Map attributes) {
- final Map attrCopy = new HashMap<>(attributes);
-
- return new FlowFile() {
- @Override
- public long getId() {
- return id;
- }
-
- @Override
- public long getEntryDate() {
- return System.currentTimeMillis();
- }
-
- @Override
- public Set getLineageIdentifiers() {
- return new HashSet();
- }
-
- @Override
- public long getLineageStartDate() {
- return System.currentTimeMillis();
- }
-
- @Override
- public Long getLastQueueDate() {
- return System.currentTimeMillis();
- }
-
- @Override
- public boolean isPenalized() {
- return false;
- }
-
- @Override
- public String getAttribute(final String s) {
- return attrCopy.get(s);
- }
-
- @Override
- public long getSize() {
- return fileSize;
- }
-
- @Override
- public Map getAttributes() {
- return attrCopy;
- }
-
- @Override
- public int compareTo(final FlowFile o) {
- return 0;
- }
- };
- }
+
private EventReporter getEventReporter() {
return new EventReporter() {
@@ -261,6 +240,8 @@ public class TestPersistentProvenanceRepository {
repo.registerEvent(record);
}
+ Thread.sleep(1000L);
+
repo.close();
Thread.sleep(500L); // Give the repo time to shutdown (i.e., close all file handles, etc.)
@@ -417,10 +398,10 @@ public class TestPersistentProvenanceRepository {
@Test
public void testIndexAndCompressOnRolloverAndSubsequentSearch() throws IOException, InterruptedException, ParseException {
final RepositoryConfiguration config = createConfiguration();
- config.setMaxRecordLife(3, TimeUnit.SECONDS);
- config.setMaxStorageCapacity(1024L * 1024L);
+ config.setMaxRecordLife(30, TimeUnit.SECONDS);
+ config.setMaxStorageCapacity(1024L * 1024L * 10);
config.setMaxEventFileLife(500, TimeUnit.MILLISECONDS);
- config.setMaxEventFileCapacity(1024L * 1024L);
+ config.setMaxEventFileCapacity(1024L * 1024L * 10);
config.setSearchableFields(new ArrayList<>(SearchableFields.getStandardFields()));
repo = new PersistentProvenanceRepository(config, DEFAULT_ROLLOVER_MILLIS);
@@ -923,12 +904,16 @@ public class TestPersistentProvenanceRepository {
final PersistentProvenanceRepository secondRepo = new PersistentProvenanceRepository(config, DEFAULT_ROLLOVER_MILLIS);
secondRepo.initialize(getEventReporter());
- final ProvenanceEventRecord event11 = builder.build();
- secondRepo.registerEvent(event11);
- secondRepo.waitForRollover();
- final ProvenanceEventRecord event11Retrieved = secondRepo.getEvent(10L);
- assertNotNull(event11Retrieved);
- assertEquals(10, event11Retrieved.getEventId());
+ try {
+ final ProvenanceEventRecord event11 = builder.build();
+ secondRepo.registerEvent(event11);
+ secondRepo.waitForRollover();
+ final ProvenanceEventRecord event11Retrieved = secondRepo.getEvent(10L);
+ assertNotNull(event11Retrieved);
+ assertEquals(10, event11Retrieved.getEventId());
+ } finally {
+ secondRepo.close();
+ }
}
@Test
@@ -998,6 +983,73 @@ public class TestPersistentProvenanceRepository {
storageDirFiles = config.getStorageDirectories().get(0).listFiles(indexFileFilter);
assertEquals(0, storageDirFiles.length);
}
+
+
+ @Test
+ public void testBackPressure() throws IOException, InterruptedException {
+ final RepositoryConfiguration config = createConfiguration();
+ config.setMaxEventFileCapacity(1L); // force rollover on each record.
+ config.setJournalCount(1);
+
+ final AtomicInteger journalCountRef = new AtomicInteger(0);
+
+ repo = new PersistentProvenanceRepository(config, DEFAULT_ROLLOVER_MILLIS) {
+ @Override
+ protected int getJournalCount() {
+ return journalCountRef.get();
+ }
+ };
+ repo.initialize(getEventReporter());
+
+ final Map attributes = new HashMap<>();
+ final ProvenanceEventBuilder builder = new StandardProvenanceEventRecord.Builder();
+ builder.setEventTime(System.currentTimeMillis());
+ builder.setEventType(ProvenanceEventType.RECEIVE);
+ builder.setTransitUri("nifi://unit-test");
+ attributes.put("uuid", UUID.randomUUID().toString());
+ builder.fromFlowFile(createFlowFile(3L, 3000L, attributes));
+ builder.setComponentId("1234");
+ builder.setComponentType("dummy processor");
+
+ // ensure that we can register the events.
+ for (int i = 0; i < 10; i++) {
+ builder.fromFlowFile(createFlowFile(i, 3000L, attributes));
+ attributes.put("uuid", "00000000-0000-0000-0000-00000000000" + i);
+ repo.registerEvent(builder.build());
+ }
+
+ // set number of journals to 6 so that we will block.
+ journalCountRef.set(6);
+
+ final AtomicLong threadNanos = new AtomicLong(0L);
+ final Thread t = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ final long start = System.nanoTime();
+ builder.fromFlowFile(createFlowFile(13, 3000L, attributes));
+ attributes.put("uuid", "00000000-0000-0000-0000-00000000000" + 13);
+ repo.registerEvent(builder.build());
+ threadNanos.set(System.nanoTime() - start);
+ }
+ });
+ t.start();
+
+ Thread.sleep(1500L);
+
+ journalCountRef.set(1);
+ t.join();
+
+ final int threadMillis = (int) TimeUnit.NANOSECONDS.toMillis(threadNanos.get());
+ assertTrue(threadMillis > 1200); // use 1200 to account for the fact that the timing is not exact
+
+ builder.fromFlowFile(createFlowFile(15, 3000L, attributes));
+ attributes.put("uuid", "00000000-0000-0000-0000-00000000000" + 15);
+ repo.registerEvent(builder.build());
+ }
+
+
+ // TODO: test EOF on merge
+ // TODO: Test journal with no records
@Test
public void testTextualQuery() throws InterruptedException, IOException, ParseException {
diff --git a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/TestStandardRecordReaderWriter.java b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/TestStandardRecordReaderWriter.java
new file mode 100644
index 0000000000..6f85b94f15
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/TestStandardRecordReaderWriter.java
@@ -0,0 +1,189 @@
+/*
+ * 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;
+
+import static org.apache.nifi.provenance.TestUtil.createFlowFile;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+
+import org.apache.nifi.provenance.toc.StandardTocReader;
+import org.apache.nifi.provenance.toc.StandardTocWriter;
+import org.apache.nifi.provenance.toc.TocReader;
+import org.apache.nifi.provenance.toc.TocUtil;
+import org.apache.nifi.provenance.toc.TocWriter;
+import org.apache.nifi.util.file.FileUtils;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class TestStandardRecordReaderWriter {
+ @BeforeClass
+ public static void setLogLevel() {
+ System.setProperty("org.slf4j.simpleLogger.log.org.apache.nifi.provenance", "DEBUG");
+ }
+
+ private ProvenanceEventRecord createEvent() {
+ final Map attributes = new HashMap<>();
+ attributes.put("filename", "1.txt");
+ attributes.put("uuid", UUID.randomUUID().toString());
+
+ final ProvenanceEventBuilder builder = new StandardProvenanceEventRecord.Builder();
+ builder.setEventTime(System.currentTimeMillis());
+ builder.setEventType(ProvenanceEventType.RECEIVE);
+ builder.setTransitUri("nifi://unit-test");
+ builder.fromFlowFile(createFlowFile(3L, 3000L, attributes));
+ builder.setComponentId("1234");
+ builder.setComponentType("dummy processor");
+ final ProvenanceEventRecord record = builder.build();
+
+ return record;
+ }
+
+ @Test
+ public void testSimpleWriteWithToc() throws IOException {
+ final File journalFile = new File("target/storage/" + UUID.randomUUID().toString() + "/testSimpleWrite");
+ final File tocFile = TocUtil.getTocFile(journalFile);
+ final TocWriter tocWriter = new StandardTocWriter(tocFile, false, false);
+ final StandardRecordWriter writer = new StandardRecordWriter(journalFile, tocWriter, false, 1024 * 1024);
+
+ writer.writeHeader();
+ writer.writeRecord(createEvent(), 1L);
+ writer.close();
+
+ final TocReader tocReader = new StandardTocReader(tocFile);
+
+ try (final FileInputStream fis = new FileInputStream(journalFile);
+ final StandardRecordReader reader = new StandardRecordReader(fis, journalFile.getName(), tocReader)) {
+ assertEquals(0, reader.getBlockIndex());
+ reader.skipToBlock(0);
+ StandardProvenanceEventRecord recovered = reader.nextRecord();
+ assertNotNull(recovered);
+
+ assertEquals("nifi://unit-test", recovered.getTransitUri());
+ assertNull(reader.nextRecord());
+ }
+
+ FileUtils.deleteFile(journalFile.getParentFile(), true);
+ }
+
+
+ @Test
+ public void testSingleRecordCompressed() throws IOException {
+ final File journalFile = new File("target/storage/" + UUID.randomUUID().toString() + "/testSimpleWrite.gz");
+ final File tocFile = TocUtil.getTocFile(journalFile);
+ final TocWriter tocWriter = new StandardTocWriter(tocFile, false, false);
+ final StandardRecordWriter writer = new StandardRecordWriter(journalFile, tocWriter, true, 100);
+
+ writer.writeHeader();
+ writer.writeRecord(createEvent(), 1L);
+ writer.close();
+
+ final TocReader tocReader = new StandardTocReader(tocFile);
+
+ try (final FileInputStream fis = new FileInputStream(journalFile);
+ final StandardRecordReader reader = new StandardRecordReader(fis, journalFile.getName(), tocReader)) {
+ assertEquals(0, reader.getBlockIndex());
+ reader.skipToBlock(0);
+ StandardProvenanceEventRecord recovered = reader.nextRecord();
+ assertNotNull(recovered);
+
+ assertEquals("nifi://unit-test", recovered.getTransitUri());
+ assertNull(reader.nextRecord());
+ }
+
+ FileUtils.deleteFile(journalFile.getParentFile(), true);
+ }
+
+
+ @Test
+ public void testMultipleRecordsSameBlockCompressed() throws IOException {
+ final File journalFile = new File("target/storage/" + UUID.randomUUID().toString() + "/testSimpleWrite.gz");
+ final File tocFile = TocUtil.getTocFile(journalFile);
+ final TocWriter tocWriter = new StandardTocWriter(tocFile, false, false);
+ // new record each 1 MB of uncompressed data
+ final StandardRecordWriter writer = new StandardRecordWriter(journalFile, tocWriter, true, 1024 * 1024);
+
+ writer.writeHeader();
+ for (int i=0; i < 10; i++) {
+ writer.writeRecord(createEvent(), i);
+ }
+ writer.close();
+
+ final TocReader tocReader = new StandardTocReader(tocFile);
+
+ try (final FileInputStream fis = new FileInputStream(journalFile);
+ final StandardRecordReader reader = new StandardRecordReader(fis, journalFile.getName(), tocReader)) {
+ for (int i=0; i < 10; i++) {
+ assertEquals(0, reader.getBlockIndex());
+
+ // call skipToBlock half the time to ensure that we can; avoid calling it
+ // the other half of the time to ensure that it's okay.
+ if (i <= 5) {
+ reader.skipToBlock(0);
+ }
+
+ StandardProvenanceEventRecord recovered = reader.nextRecord();
+ assertNotNull(recovered);
+ assertEquals("nifi://unit-test", recovered.getTransitUri());
+ }
+
+ assertNull(reader.nextRecord());
+ }
+
+ FileUtils.deleteFile(journalFile.getParentFile(), true);
+ }
+
+
+ @Test
+ public void testMultipleRecordsMultipleBlocksCompressed() throws IOException {
+ final File journalFile = new File("target/storage/" + UUID.randomUUID().toString() + "/testSimpleWrite.gz");
+ final File tocFile = TocUtil.getTocFile(journalFile);
+ final TocWriter tocWriter = new StandardTocWriter(tocFile, false, false);
+ // new block each 10 bytes
+ final StandardRecordWriter writer = new StandardRecordWriter(journalFile, tocWriter, true, 100);
+
+ writer.writeHeader();
+ for (int i=0; i < 10; i++) {
+ writer.writeRecord(createEvent(), i);
+ }
+ writer.close();
+
+ final TocReader tocReader = new StandardTocReader(tocFile);
+
+ try (final FileInputStream fis = new FileInputStream(journalFile);
+ final StandardRecordReader reader = new StandardRecordReader(fis, journalFile.getName(), tocReader)) {
+ for (int i=0; i < 10; i++) {
+ StandardProvenanceEventRecord recovered = reader.nextRecord();
+ System.out.println(recovered);
+ assertNotNull(recovered);
+ assertEquals((long) i, recovered.getEventId());
+ assertEquals("nifi://unit-test", recovered.getTransitUri());
+ }
+
+ assertNull(reader.nextRecord());
+ }
+
+ FileUtils.deleteFile(journalFile.getParentFile(), true);
+ }
+}
diff --git a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/TestUtil.java b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/TestUtil.java
new file mode 100644
index 0000000000..7459fe8a8c
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/TestUtil.java
@@ -0,0 +1,82 @@
+/*
+ * 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;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.nifi.flowfile.FlowFile;
+
+public class TestUtil {
+ public static FlowFile createFlowFile(final long id, final long fileSize, final Map attributes) {
+ final Map attrCopy = new HashMap<>(attributes);
+
+ return new FlowFile() {
+ @Override
+ public long getId() {
+ return id;
+ }
+
+ @Override
+ public long getEntryDate() {
+ return System.currentTimeMillis();
+ }
+
+ @Override
+ public Set getLineageIdentifiers() {
+ return new HashSet();
+ }
+
+ @Override
+ public long getLineageStartDate() {
+ return System.currentTimeMillis();
+ }
+
+ @Override
+ public Long getLastQueueDate() {
+ return System.currentTimeMillis();
+ }
+
+ @Override
+ public boolean isPenalized() {
+ return false;
+ }
+
+ @Override
+ public String getAttribute(final String s) {
+ return attrCopy.get(s);
+ }
+
+ @Override
+ public long getSize() {
+ return fileSize;
+ }
+
+ @Override
+ public Map getAttributes() {
+ return attrCopy;
+ }
+
+ @Override
+ public int compareTo(final FlowFile o) {
+ return 0;
+ }
+ };
+ }
+}
diff --git a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/toc/TestStandardTocReader.java b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/toc/TestStandardTocReader.java
new file mode 100644
index 0000000000..30326e7264
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/toc/TestStandardTocReader.java
@@ -0,0 +1,91 @@
+/*
+ * 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.toc;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.UUID;
+
+import org.junit.Test;
+
+public class TestStandardTocReader {
+
+ @Test
+ public void testDetectsCompression() throws IOException {
+ final File file = new File("target/" + UUID.randomUUID().toString());
+ try (final OutputStream out = new FileOutputStream(file)) {
+ out.write(0);
+ out.write(0);
+ }
+
+ try {
+ try(final StandardTocReader reader = new StandardTocReader(file)) {
+ assertFalse(reader.isCompressed());
+ }
+ } finally {
+ file.delete();
+ }
+
+
+ try (final OutputStream out = new FileOutputStream(file)) {
+ out.write(0);
+ out.write(1);
+ }
+
+ try {
+ try(final StandardTocReader reader = new StandardTocReader(file)) {
+ assertTrue(reader.isCompressed());
+ }
+ } finally {
+ file.delete();
+ }
+ }
+
+
+ @Test
+ public void testGetBlockIndex() throws IOException {
+ final File file = new File("target/" + UUID.randomUUID().toString());
+ try (final OutputStream out = new FileOutputStream(file);
+ final DataOutputStream dos = new DataOutputStream(out)) {
+ out.write(0);
+ out.write(0);
+
+ for (int i=0; i < 1024; i++) {
+ dos.writeLong(i * 1024L);
+ }
+ }
+
+ try {
+ try(final StandardTocReader reader = new StandardTocReader(file)) {
+ assertFalse(reader.isCompressed());
+
+ for (int i=0; i < 1024; i++) {
+ assertEquals(i * 1024, reader.getBlockOffset(i));
+ }
+ }
+ } finally {
+ file.delete();
+ }
+ }
+}
diff --git a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/toc/TestStandardTocWriter.java b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/toc/TestStandardTocWriter.java
new file mode 100644
index 0000000000..70f55a2c6c
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/toc/TestStandardTocWriter.java
@@ -0,0 +1,42 @@
+/*
+ * 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.toc;
+
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.UUID;
+
+import org.apache.nifi.util.file.FileUtils;
+import org.junit.Test;
+
+public class TestStandardTocWriter {
+ @Test
+ public void testOverwriteEmptyFile() throws IOException {
+ final File tocFile = new File("target/" + UUID.randomUUID().toString() + ".toc");
+ try {
+ assertTrue( tocFile.createNewFile() );
+
+ try (final StandardTocWriter writer = new StandardTocWriter(tocFile, false, false)) {
+ }
+ } finally {
+ FileUtils.deleteFile(tocFile, false);
+ }
+ }
+
+}
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml
index c3c05df34e..be0fc67585 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml
@@ -176,6 +176,7 @@
apache-rat-plugin
+ src/test/resources/localhost.cer
src/test/resources/hello.txt
src/test/resources/CharacterSetConversionSamples/Converted.txt
src/test/resources/CharacterSetConversionSamples/Original.txt
@@ -231,6 +232,8 @@
src/test/resources/TestTransformXml/tokens.xml
src/test/resources/TestUnpackContent/folder/cal.txt
src/test/resources/TestUnpackContent/folder/date.txt
+ src/test/resources/TestUnpackContent/data.flowfilev2
+ src/test/resources/TestUnpackContent/data.flowfilev3
src/test/resources/TestXml/xml-bundle-1
src/test/resources/CompressedData/SampleFile.txt.bz2
src/test/resources/CompressedData/SampleFile.txt.gz