You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by jp...@apache.org on 2016/11/15 21:49:48 UTC
nifi git commit: NIFI-3036: When we replay a FlowFile,
ensure that we are using the 'golden copy' of the associated Resource
Claim,
if the claim is still writable. Ensure that StandardResourceClaimManager
retains the 'golden copy' of a Resource Claim unt
Repository: nifi
Updated Branches:
refs/heads/master 81357d445 -> 721964b7d
NIFI-3036: When we replay a FlowFile, ensure that we are using the 'golden copy' of the associated Resource Claim, if the claim is still writable. Ensure that StandardResourceClaimManager retains the 'golden copy' of a Resource Claim until it is no longer writable and has a claim count of 0
This closes #1223
Signed-off-by: jpercivall <JP...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/721964b7
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/721964b7
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/721964b7
Branch: refs/heads/master
Commit: 721964b7d87c696f293f7c769b274fc6364ff034
Parents: 81357d4
Author: Mark Payne <ma...@hotmail.com>
Authored: Tue Nov 15 09:15:37 2016 -0500
Committer: jpercivall <JP...@apache.org>
Committed: Tue Nov 15 16:51:34 2016 -0500
----------------------------------------------------------------------
.../org/apache/nifi/controller/FlowController.java | 17 ++++++++++++++---
.../repository/FileSystemRepository.java | 10 +++++++++-
.../claim/StandardResourceClaimManager.java | 11 ++++++++++-
3 files changed, 33 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/721964b7/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
index 6927944..faa4230 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
@@ -3795,9 +3795,20 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
throw new IllegalStateException("Cannot replay data from Provenance Event because the Source FlowFile Queue with ID " + event.getSourceQueueIdentifier() + " no longer exists");
}
- // Create the ContentClaim
- final ResourceClaim resourceClaim = resourceClaimManager.newResourceClaim(event.getPreviousContentClaimContainer(),
- event.getPreviousContentClaimSection(), event.getPreviousContentClaimIdentifier(), false, false);
+ // Create the ContentClaim. To do so, we first need the appropriate Resource Claim. Because we don't know whether or
+ // not the Resource Claim is still active, we first call ResourceClaimManager.getResourceClaim. If this returns
+ // null, then we know that the Resource Claim is no longer active and can just create a new one that is not writable.
+ // It's critical though that we first call getResourceClaim because otherwise, if the Resource Claim is active and we
+ // create a new one that is not writable, we could end up archiving or destroying the Resource Claim while it's still
+ // being written to by the Content Repository. This is important only because we are creating a FlowFile with this Resource
+ // Claim. If, for instance, we are simply creating the claim to request its content, as in #getContentAvailability, etc.
+ // then this is not necessary.
+ ResourceClaim resourceClaim = resourceClaimManager.getResourceClaim(event.getPreviousContentClaimContainer(),
+ event.getPreviousContentClaimSection(), event.getPreviousContentClaimIdentifier());
+ if (resourceClaim == null) {
+ resourceClaim = resourceClaimManager.newResourceClaim(event.getPreviousContentClaimContainer(),
+ event.getPreviousContentClaimSection(), event.getPreviousContentClaimIdentifier(), false, false);
+ }
// Increment Claimant Count, since we will now be referencing the Content Claim
resourceClaimManager.incrementClaimantCount(resourceClaim);
http://git-wip-us.apache.org/repos/asf/nifi/blob/721964b7/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/FileSystemRepository.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/FileSystemRepository.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/FileSystemRepository.java
index e45852a..5eb9c5a 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/FileSystemRepository.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/FileSystemRepository.java
@@ -943,7 +943,15 @@ public class FileSystemRepository implements ContentRepository {
final long resourceClaimLength = scc.getOffset() + scc.getLength();
if (recycle && resourceClaimLength < MAX_APPENDABLE_CLAIM_LENGTH) {
final ClaimLengthPair pair = new ClaimLengthPair(scc.getResourceClaim(), resourceClaimLength);
- final boolean enqueued = writableClaimQueue.offer(pair);
+
+ // We are checking that writableClaimStreams contains the resource claim as a key, as a sanity check.
+ // It should always be there. However, we have encountered a bug before where we archived content before
+ // we should have. As a result, the Resource Claim and the associated OutputStream were removed from the
+ // writableClaimStreams map, and this caused a NullPointerException. Worse, the call here to
+ // writableClaimQueue.offer() means that the ResourceClaim was then reused, which resulted in an endless
+ // loop of NullPointerException's being thrown. As a result, we simply ensure that the Resource Claim does
+ // in fact have an OutputStream associated with it before adding it back to the writableClaimQueue.
+ final boolean enqueued = writableClaimStreams.get(scc.getResourceClaim()) != null && writableClaimQueue.offer(pair);
if (enqueued) {
LOG.debug("Claim length less than max; Adding {} back to Writable Claim Queue", this);
http://git-wip-us.apache.org/repos/asf/nifi/blob/721964b7/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/claim/StandardResourceClaimManager.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/claim/StandardResourceClaimManager.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/claim/StandardResourceClaimManager.java
index be0e8b8..7d554b1 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/claim/StandardResourceClaimManager.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/claim/StandardResourceClaimManager.java
@@ -97,7 +97,10 @@ public class StandardResourceClaimManager implements ResourceClaimManager {
logger.debug("Decrementing claimant count for {} to {}", claim, newClaimantCount);
}
- if (newClaimantCount == 0) {
+ // If the claim is no longer referenced, we want to remove it. We consider the claim to be "no longer referenced"
+ // if the count is 0 and it is no longer writable (if it's writable, it may still be writable by the Content Repository,
+ // even though no existing FlowFile is referencing the claim).
+ if (newClaimantCount == 0 && !claim.isWritable()) {
removeClaimantCount(claim);
}
return newClaimantCount;
@@ -188,6 +191,12 @@ public class StandardResourceClaimManager implements ResourceClaimManager {
}
((StandardResourceClaim) claim).freeze();
+
+ synchronized (claim) {
+ if (getClaimantCount(claim) == 0) {
+ claimantCounts.remove(claim);
+ }
+ }
}