You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2015/04/22 15:57:35 UTC
[6/7] incubator-nifi git commit: NIFI-527: Added unit test to verify
backpressure
NIFI-527: Added unit test to verify backpressure
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/4dafd65a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/4dafd65a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/4dafd65a
Branch: refs/heads/improve-prov-performance
Commit: 4dafd65a2e9296af7a229026fdf99e28b9c53f37
Parents: 2f7b2ef
Author: Mark Payne <ma...@hotmail.com>
Authored: Tue Apr 21 11:11:58 2015 -0400
Committer: Mark Payne <ma...@hotmail.com>
Committed: Tue Apr 21 11:11:58 2015 -0400
----------------------------------------------------------------------
.../PersistentProvenanceRepository.java | 10 +--
.../TestPersistentProvenanceRepository.java | 66 ++++++++++++++++++++
2 files changed, 72 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/4dafd65a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/PersistentProvenanceRepository.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/PersistentProvenanceRepository.java b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/PersistentProvenanceRepository.java
index 7bf7f8a..ab3c2df 100644
--- a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/PersistentProvenanceRepository.java
+++ b/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/PersistentProvenanceRepository.java
@@ -608,8 +608,10 @@ public class PersistentProvenanceRepository implements ProvenanceEventRepository
indexManager.close();
- for (final RecordWriter writer : writers) {
- writer.close();
+ if ( writers != null ) {
+ for (final RecordWriter writer : writers) {
+ writer.close();
+ }
}
} finally {
writeLock.unlock();
@@ -905,8 +907,8 @@ public class PersistentProvenanceRepository implements ProvenanceEventRepository
}
}
-
- private int getJournalCount() {
+ // made protected for testing purposes
+ protected int getJournalCount() {
// determine how many 'journals' we have in the journals directories
int journalFileCount = 0;
for ( final File storageDir : configuration.getStorageDirectories() ) {
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/4dafd65a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/TestPersistentProvenanceRepository.java
----------------------------------------------------------------------
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 7822e5c..25a363f 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
@@ -32,6 +32,8 @@ 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;
@@ -965,6 +967,70 @@ public class TestPersistentProvenanceRepository {
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<String, String> 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