This commit is contained in:
Matt Gilman 2015-06-04 13:10:27 -04:00
commit e2227cdafa
14 changed files with 106 additions and 130 deletions

View File

@ -48,7 +48,7 @@
<groupId>org.apache.rat</groupId>
<artifactId>apache-rat-plugin</artifactId>
<configuration>
<excludes>
<excludes combine.children="append">
<exclude>src/test/resources/hypoglycemia</exclude>
<exclude>src/test/resources/hyperglycemia</exclude>
</excludes>

View File

@ -107,7 +107,7 @@
<groupId>org.apache.rat</groupId>
<artifactId>apache-rat-plugin</artifactId>
<configuration>
<excludes>
<excludes combine.children="append">
<exclude>src/main/asciidoc/asciidoc-mod.css</exclude> <!-- MIT license confirmed. Excluding due to parse error-->
</excludes>
</configuration>

View File

@ -61,7 +61,7 @@
<groupId>org.apache.rat</groupId>
<artifactId>apache-rat-plugin</artifactId>
<configuration>
<excludes>
<excludes combine.children="append">
<exclude>src/test/resources/hello.txt</exclude>
</excludes>
</configuration>

View File

@ -137,7 +137,7 @@
<groupId>org.apache.rat</groupId>
<artifactId>apache-rat-plugin</artifactId>
<configuration>
<excludes>
<excludes combine.children="append">
<exclude>src/test/resources/org/apache/nifi/cluster/firewall/impl/empty.txt</exclude>
<exclude>src/test/resources/org/apache/nifi/cluster/firewall/impl/ips.txt</exclude>
</excludes>

View File

@ -127,7 +127,7 @@
<groupId>org.apache.rat</groupId>
<artifactId>apache-rat-plugin</artifactId>
<configuration>
<excludes>
<excludes combine.children="append">
<exclude>src/test/resources/conf/0bytes.xml</exclude>
<exclude>src/test/resources/conf/termination-only.xml</exclude>
<exclude>src/test/resources/hello.txt</exclude>

View File

@ -93,7 +93,7 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
// determines how many things must be transferred, removed, modified in order to avoid logging the FlowFile ID's on commit/rollback
public static final int VERBOSE_LOG_THRESHOLD = 10;
private static final long MAX_APPENDABLE_CLAIM_SIZE = DataUnit.parseDataSize(
NiFiProperties.getInstance().getMaxAppendableClaimSize(), DataUnit.B).longValue();
NiFiProperties.getInstance().getMaxAppendableClaimSize(), DataUnit.B).longValue();
private static final int MAX_FLOWFILES_PER_CLAIM = NiFiProperties.getInstance().getMaxFlowFilesPerClaim();
public static final String DEFAULT_FLOWFILE_PATH = "./";
@ -154,28 +154,28 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
String description = connectable.toString();
switch (connectable.getConnectableType()) {
case PROCESSOR:
final ProcessorNode procNode = (ProcessorNode) connectable;
componentType = procNode.getProcessor().getClass().getSimpleName();
description = procNode.getProcessor().toString();
break;
case INPUT_PORT:
componentType = "Input Port";
break;
case OUTPUT_PORT:
componentType = "Output Port";
break;
case REMOTE_INPUT_PORT:
componentType = "Remote Input Port";
break;
case REMOTE_OUTPUT_PORT:
componentType = "Remote Output Port";
break;
case FUNNEL:
componentType = "Funnel";
break;
default:
throw new AssertionError("Connectable type is " + connectable.getConnectableType());
case PROCESSOR:
final ProcessorNode procNode = (ProcessorNode) connectable;
componentType = procNode.getProcessor().getClass().getSimpleName();
description = procNode.getProcessor().toString();
break;
case INPUT_PORT:
componentType = "Input Port";
break;
case OUTPUT_PORT:
componentType = "Output Port";
break;
case REMOTE_INPUT_PORT:
componentType = "Remote Input Port";
break;
case REMOTE_OUTPUT_PORT:
componentType = "Remote Output Port";
break;
case FUNNEL:
componentType = "Funnel";
break;
default:
throw new AssertionError("Connectable type is " + connectable.getConnectableType());
}
this.provenanceReporter = new StandardProvenanceReporter(connectable.getIdentifier(), componentType, context.getProvenanceRepository(), this);
@ -437,16 +437,16 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
}
switch (event.getEventType()) {
case SEND:
flowFilesSent++;
bytesSent += event.getFileSize();
break;
case RECEIVE:
flowFilesReceived++;
bytesReceived += event.getFileSize();
break;
default:
break;
case SEND:
flowFilesSent++;
bytesSent += event.getFileSize();
break;
case RECEIVE:
flowFilesReceived++;
bytesReceived += event.getFileSize();
break;
default:
break;
}
}
@ -519,9 +519,13 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
updateEventContentClaims(builder, flowFile, checkpoint.records.get(flowFile));
final ProvenanceEventRecord event = builder.build();
if (!event.getChildUuids().isEmpty() && !isSpuriousForkEvent(event, checkpoint.removedFlowFiles) && !processorGenerated.contains(event)) {
recordsToSubmit.add(event);
if (!event.getChildUuids().isEmpty() && !isSpuriousForkEvent(event, checkpoint.removedFlowFiles)) {
// If framework generated the event, add it to the 'recordsToSubmit' Set.
if (!processorGenerated.contains(event)) {
recordsToSubmit.add(event);
}
// Register the FORK event for each child and each parent.
for (final String childUuid : event.getChildUuids()) {
addEventType(eventTypesPerFlowFileId, childUuid, event.getEventType());
}
@ -536,13 +540,12 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
if (isSpuriousForkEvent(event, checkpoint.removedFlowFiles)) {
continue;
}
if (isSpuriousRouteEvent(event, checkpoint.records)) {
continue;
}
// Check if the event indicates that the FlowFile was routed to the same
// connection from which it was pulled (and only this connection). If so, discard the event.
isSpuriousRouteEvent(event, checkpoint.records);
if (isSpuriousRouteEvent(event, checkpoint.records)) {
continue;
}
recordsToSubmit.add(event);
addEventType(eventTypesPerFlowFileId, event.getFlowFileUuid(), event.getEventType());
@ -596,9 +599,9 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
boolean creationEventRegistered = false;
if (registeredTypes != null) {
if (registeredTypes.contains(ProvenanceEventType.CREATE)
|| registeredTypes.contains(ProvenanceEventType.FORK)
|| registeredTypes.contains(ProvenanceEventType.JOIN)
|| registeredTypes.contains(ProvenanceEventType.RECEIVE)) {
|| registeredTypes.contains(ProvenanceEventType.FORK)
|| registeredTypes.contains(ProvenanceEventType.JOIN)
|| registeredTypes.contains(ProvenanceEventType.RECEIVE)) {
creationEventRegistered = true;
}
}
@ -687,11 +690,11 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
builder.setCurrentContentClaim(null, null, null, null, 0L);
} else {
builder.setCurrentContentClaim(
originalClaim.getContainer(), originalClaim.getSection(), originalClaim.getId(), repoRecord.getOriginal().getContentClaimOffset(), repoRecord.getOriginal().getSize()
);
originalClaim.getContainer(), originalClaim.getSection(), originalClaim.getId(), repoRecord.getOriginal().getContentClaimOffset(), repoRecord.getOriginal().getSize()
);
builder.setPreviousContentClaim(
originalClaim.getContainer(), originalClaim.getSection(), originalClaim.getId(), repoRecord.getOriginal().getContentClaimOffset(), repoRecord.getOriginal().getSize()
);
originalClaim.getContainer(), originalClaim.getSection(), originalClaim.getId(), repoRecord.getOriginal().getContentClaimOffset(), repoRecord.getOriginal().getSize()
);
}
}
@ -727,7 +730,7 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
}
private StandardProvenanceEventRecord enrich(
final ProvenanceEventRecord rawEvent, final Map<String, FlowFileRecord> flowFileRecordMap, final Map<FlowFileRecord, StandardRepositoryRecord> records, final boolean updateAttributes) {
final ProvenanceEventRecord rawEvent, final Map<String, FlowFileRecord> flowFileRecordMap, final Map<FlowFileRecord, StandardRepositoryRecord> records, final boolean updateAttributes) {
final StandardProvenanceEventRecord.Builder recordBuilder = new StandardProvenanceEventRecord.Builder().fromEvent(rawEvent);
final FlowFileRecord eventFlowFile = flowFileRecordMap.get(rawEvent.getFlowFileUuid());
if (eventFlowFile != null) {
@ -996,7 +999,7 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
final StringBuilder sb = new StringBuilder(512);
if (!LOG.isDebugEnabled() && (largestTransferSetSize > VERBOSE_LOG_THRESHOLD
|| numModified > VERBOSE_LOG_THRESHOLD || numCreated > VERBOSE_LOG_THRESHOLD || numRemoved > VERBOSE_LOG_THRESHOLD)) {
|| numModified > VERBOSE_LOG_THRESHOLD || numCreated > VERBOSE_LOG_THRESHOLD || numRemoved > VERBOSE_LOG_THRESHOLD)) {
if (numCreated > 0) {
sb.append("created ").append(numCreated).append(" FlowFiles, ");
}
@ -1244,8 +1247,8 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
attrs.put(CoreAttributes.UUID.key(), UUID.randomUUID().toString());
final FlowFileRecord fFile = new StandardFlowFileRecord.Builder().id(context.getNextFlowFileSequence())
.addAttributes(attrs)
.build();
.addAttributes(attrs)
.build();
final StandardRepositoryRecord record = new StandardRepositoryRecord(null);
record.setWorking(fFile, attrs);
records.put(fFile, record);
@ -1607,7 +1610,7 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
}
final StandardProvenanceReporter expiredReporter = new StandardProvenanceReporter(connectable.getIdentifier(),
processorType, context.getProvenanceRepository(), this);
processorType, context.getProvenanceRepository(), this);
final Map<String, FlowFileRecord> recordIdMap = new HashMap<>();
for (final FlowFileRecord flowFile : flowFiles) {
@ -1734,9 +1737,9 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
}
try (final InputStream rawIn = getInputStream(source, record.getCurrentClaim(), record.getCurrentClaimOffset());
final InputStream limitedIn = new LimitedInputStream(rawIn, source.getSize());
final InputStream disableOnCloseIn = new DisableOnCloseInputStream(limitedIn);
final ByteCountingInputStream countingStream = new ByteCountingInputStream(disableOnCloseIn, this.bytesRead)) {
final InputStream limitedIn = new LimitedInputStream(rawIn, source.getSize());
final InputStream disableOnCloseIn = new DisableOnCloseInputStream(limitedIn);
final ByteCountingInputStream countingStream = new ByteCountingInputStream(disableOnCloseIn, this.bytesRead)) {
// We want to differentiate between IOExceptions thrown by the repository and IOExceptions thrown from
// Processor code. As a result, as have the FlowFileAccessInputStream that catches IOException from the repository
@ -1807,7 +1810,7 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
try {
try (final OutputStream rawOut = contentRepo.write(newClaim);
final OutputStream out = new BufferedOutputStream(rawOut)) {
final OutputStream out = new BufferedOutputStream(rawOut)) {
if (header != null && header.length > 0) {
out.write(header);
@ -1914,7 +1917,7 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
ensureNotAppending(newClaim);
try (final OutputStream disableOnClose = new DisableOnCloseOutputStream(currentWriteClaimStream);
final OutputStream countingOut = new ByteCountingOutputStream(disableOnClose, writtenHolder)) {
final OutputStream countingOut = new ByteCountingOutputStream(disableOnClose, writtenHolder)) {
recursionSet.add(source);
@ -1937,7 +1940,7 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
ensureNotAppending(newClaim);
try (final OutputStream stream = context.getContentRepository().write(newClaim);
final OutputStream countingOut = new ByteCountingOutputStream(stream, writtenHolder)) {
final OutputStream countingOut = new ByteCountingOutputStream(stream, writtenHolder)) {
recursionSet.add(source);
writeRecursionLevel++;
@ -2173,11 +2176,11 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
ensureNotAppending(newClaim);
try (final InputStream rawIn = getInputStream(source, currClaim, record.getCurrentClaimOffset());
final InputStream limitedIn = new LimitedInputStream(rawIn, source.getSize());
final InputStream disableOnCloseIn = new DisableOnCloseInputStream(limitedIn);
final InputStream countingIn = new ByteCountingInputStream(disableOnCloseIn, bytesRead);
final OutputStream disableOnCloseOut = new DisableOnCloseOutputStream(currentWriteClaimStream);
final OutputStream countingOut = new ByteCountingOutputStream(disableOnCloseOut, writtenHolder)) {
final InputStream limitedIn = new LimitedInputStream(rawIn, source.getSize());
final InputStream disableOnCloseIn = new DisableOnCloseInputStream(limitedIn);
final InputStream countingIn = new ByteCountingInputStream(disableOnCloseIn, bytesRead);
final OutputStream disableOnCloseOut = new DisableOnCloseOutputStream(currentWriteClaimStream);
final OutputStream countingOut = new ByteCountingOutputStream(disableOnCloseOut, writtenHolder)) {
recursionSet.add(source);
@ -2216,10 +2219,10 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
ensureNotAppending(newClaim);
try (final InputStream is = getInputStream(source, currClaim, record.getCurrentClaimOffset());
final InputStream limitedIn = new LimitedInputStream(is, source.getSize());
final InputStream countingIn = new ByteCountingInputStream(limitedIn, bytesRead);
final OutputStream os = context.getContentRepository().write(newClaim);
final OutputStream countingOut = new ByteCountingOutputStream(os, writtenHolder)) {
final InputStream limitedIn = new LimitedInputStream(is, source.getSize());
final InputStream countingIn = new ByteCountingInputStream(limitedIn, bytesRead);
final OutputStream os = context.getContentRepository().write(newClaim);
final OutputStream countingOut = new ByteCountingOutputStream(os, writtenHolder)) {
recursionSet.add(source);
@ -2332,9 +2335,9 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
removeTemporaryClaim(record);
final FlowFileRecord newFile = new StandardFlowFileRecord.Builder().fromFlowFile(record.getCurrent())
.contentClaim(newClaim).contentClaimOffset(claimOffset).size(newSize)
.addAttribute(CoreAttributes.FILENAME.key(), source.toFile().getName())
.build();
.contentClaim(newClaim).contentClaimOffset(claimOffset).size(newSize)
.addAttribute(CoreAttributes.FILENAME.key(), source.toFile().getName())
.build();
record.setWorking(newFile, CoreAttributes.FILENAME.key(), source.toFile().getName());
if (!keepSourceFile) {
deleteOnCommit.add(source);
@ -2501,8 +2504,8 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
final String key = entry.getKey();
final String value = entry.getValue();
if (CoreAttributes.ALTERNATE_IDENTIFIER.key().equals(key)
|| CoreAttributes.DISCARD_REASON.key().equals(key)
|| CoreAttributes.UUID.key().equals(key)) {
|| CoreAttributes.DISCARD_REASON.key().equals(key)
|| CoreAttributes.UUID.key().equals(key)) {
continue;
}
newAttributes.put(key, value);
@ -2547,10 +2550,10 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
newAttributes.put(CoreAttributes.UUID.key(), UUID.randomUUID().toString());
final FlowFileRecord fFile = new StandardFlowFileRecord.Builder().id(context.getNextFlowFileSequence())
.addAttributes(newAttributes)
.lineageIdentifiers(lineageIdentifiers)
.lineageStartDate(lineageStartDate)
.build();
.addAttributes(newAttributes)
.lineageIdentifiers(lineageIdentifiers)
.lineageStartDate(lineageStartDate)
.build();
final StandardRepositoryRecord record = new StandardRepositoryRecord(null);
record.setWorking(fFile, newAttributes);
@ -2584,18 +2587,18 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
final Map<String, String> firstMap = flowFileList.iterator().next().getAttributes();
outer:
for (final Map.Entry<String, String> mapEntry : firstMap.entrySet()) {
final String key = mapEntry.getKey();
final String value = mapEntry.getValue();
for (final FlowFile flowFile : flowFileList) {
final Map<String, String> currMap = flowFile.getAttributes();
final String curVal = currMap.get(key);
if (curVal == null || !curVal.equals(value)) {
continue outer;
for (final Map.Entry<String, String> mapEntry : firstMap.entrySet()) {
final String key = mapEntry.getKey();
final String value = mapEntry.getValue();
for (final FlowFile flowFile : flowFileList) {
final Map<String, String> currMap = flowFile.getAttributes();
final String curVal = currMap.get(key);
if (curVal == null || !curVal.equals(value)) {
continue outer;
}
}
result.put(key, value);
}
result.put(key, value);
}
return result;
}

View File

@ -94,7 +94,7 @@
<groupId>org.apache.rat</groupId>
<artifactId>apache-rat-plugin</artifactId>
<configuration>
<excludes>
<excludes combine.children="append">
<exclude>src/main/webapp/js/hexview/hexview.default.css</exclude>
<exclude>src/main/webapp/js/hexview/hexview.js</exclude>
</excludes>

View File

@ -595,7 +595,7 @@
<groupId>org.apache.rat</groupId>
<artifactId>apache-rat-plugin</artifactId>
<configuration>
<excludes>
<excludes combine.children="append">
<exclude>nbactions.xml</exclude>
<exclude>src/main/webapp/js/json2.js</exclude>
<exclude>src/main/webapp/js/jquery/</exclude>

View File

@ -31,7 +31,7 @@
<groupId>org.apache.rat</groupId>
<artifactId>apache-rat-plugin</artifactId>
<configuration>
<excludes>
<excludes combine.children="append">
<exclude>src/test/resources/hypoglycemia.hl7</exclude>
</excludes>
</configuration>

View File

@ -64,35 +64,4 @@
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>com.github.joelittlejohn.embedmongo</groupId>
<artifactId>embedmongo-maven-plugin</artifactId>
<version>0.1.12</version>
<executions>
<execution>
<id>start</id>
<goals>
<goal>start</goal>
</goals>
<phase>test-compile</phase>
<configuration>
<databaseDirectory>${project.build.directory}/embedmongo/db</databaseDirectory>
<logging>file</logging>
<logFile>${project.build.directory}/embedmongo.log</logFile>
</configuration>
</execution>
<execution>
<id>stop</id>
<goals>
<goal>stop</goal>
</goals>
<phase>prepare-package</phase>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

View File

@ -33,6 +33,7 @@ import org.bson.Document;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import com.google.common.collect.Lists;
@ -40,6 +41,7 @@ import com.mongodb.MongoClient;
import com.mongodb.MongoClientURI;
import com.mongodb.client.MongoCollection;
@Ignore("Integration tests that cause failures in some environments. Require that they be run from Maven to run the embedded mongo maven plugin. Maven Plugin also fails in my CentOS 7 environment.")
public class GetMongoTest {
private static final String MONGO_URI = "mongodb://localhost";
private static final String DB_NAME = GetMongoTest.class.getSimpleName().toLowerCase();
@ -49,7 +51,7 @@ public class GetMongoTest {
new Document("_id", "doc_1").append("a", 1).append("b", 2).append("c", 3),
new Document("_id", "doc_2").append("a", 1).append("b", 2).append("c", 4),
new Document("_id", "doc_3").append("a", 1).append("b", 3)
);
);
private TestRunner runner;
private MongoClient mongoClient;

View File

@ -34,6 +34,7 @@ import org.bson.Document;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import com.google.common.collect.Lists;
@ -41,15 +42,16 @@ import com.mongodb.MongoClient;
import com.mongodb.MongoClientURI;
import com.mongodb.client.MongoCollection;
@Ignore("Integration tests that cause failures in some environments. Require that they be run from Maven to run the embedded mongo maven plugin. Maven Plugin also fails in my CentOS 7 environment.")
public class PutMongoTest {
private static final String MONGO_URI = "mongodb://localhost";
private static final String DATABASE_NAME = PutMongoTest.class.getSimpleName().toLowerCase();
private static final String COLLECTION_NAME = "test";
private static final List<Document> DOCUMENTS = Lists.newArrayList(
new Document("_id", "doc_1").append("a", 1).append("b", 2).append("c", 3),
new Document("_id", "doc_2").append("a", 1).append("b", 2).append("c", 4),
new Document("_id", "doc_3").append("a", 1).append("b", 3)
new Document("_id", "doc_1").append("a", 1).append("b", 2).append("c", 3),
new Document("_id", "doc_2").append("a", 1).append("b", 2).append("c", 4),
new Document("_id", "doc_3").append("a", 1).append("b", 3)
);
private TestRunner runner;

View File

@ -95,7 +95,7 @@
<groupId>org.apache.rat</groupId>
<artifactId>apache-rat-plugin</artifactId>
<configuration>
<excludes>
<excludes combine.children="append">
<exclude>src/test/resources/solr/solr.xml</exclude>
<exclude>src/test/resources/testCollection/core.properties</exclude>
<exclude>src/test/resources/testCollection/conf/_rest_managed.json</exclude>

View File

@ -179,7 +179,7 @@
<groupId>org.apache.rat</groupId>
<artifactId>apache-rat-plugin</artifactId>
<configuration>
<excludes>
<excludes combine.children="append">
<exclude>src/test/resources/localhost.cer</exclude>
<exclude>src/test/resources/hello.txt</exclude>
<exclude>src/test/resources/CharacterSetConversionSamples/Converted.txt</exclude>