You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by bb...@apache.org on 2019/12/10 20:28:10 UTC
[nifi] branch master updated: NIFI-6924: When seeking to the
appropriate offset for a content claim,
ensure that if there are not enough bytes in the underlying resource claim
that a ContentNotFoundException is thrown. Also cleaned up error-handling
case in StandardProcessSession to ensure that we close the existing
InputStream before calling handleContenttNotFoundException,
since this method may itself throw an Exception
This is an automated email from the ASF dual-hosted git repository.
bbende pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/master by this push:
new 452ca98 NIFI-6924: When seeking to the appropriate offset for a content claim, ensure that if there are not enough bytes in the underlying resource claim that a ContentNotFoundException is thrown. Also cleaned up error-handling case in StandardProcessSession to ensure that we close the existing InputStream before calling handleContenttNotFoundException, since this method may itself throw an Exception
452ca98 is described below
commit 452ca98c29eeda6a89c60350e4577f82138d7ef0
Author: Mark Payne <ma...@hotmail.com>
AuthorDate: Tue Dec 10 11:20:01 2019 -0500
NIFI-6924: When seeking to the appropriate offset for a content claim, ensure that if there are not enough bytes in the underlying resource claim that a ContentNotFoundException is thrown. Also cleaned up error-handling case in StandardProcessSession to ensure that we close the existing InputStream before calling handleContenttNotFoundException, since this method may itself throw an Exception
This closes #3924.
Signed-off-by: Bryan Bende <bb...@apache.org>
---
.../repository/FileSystemRepository.java | 51 +++++++++++++---------
.../repository/StandardProcessSession.java | 18 +++++---
.../repository/TestFileSystemRepository.java | 48 ++++++++++++++++++++
3 files changed, 92 insertions(+), 25 deletions(-)
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 8eee57d..d6e5f35 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
@@ -16,7 +16,27 @@
*/
package org.apache.nifi.controller.repository;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.controller.repository.claim.ContentClaim;
+import org.apache.nifi.controller.repository.claim.ResourceClaim;
+import org.apache.nifi.controller.repository.claim.ResourceClaimManager;
+import org.apache.nifi.controller.repository.claim.StandardContentClaim;
+import org.apache.nifi.controller.repository.io.LimitedInputStream;
+import org.apache.nifi.engine.FlowEngine;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.stream.io.ByteCountingOutputStream;
+import org.apache.nifi.stream.io.StreamUtils;
+import org.apache.nifi.stream.io.SynchronizedByteCountingOutputStream;
+import org.apache.nifi.util.FormatUtils;
+import org.apache.nifi.util.NiFiProperties;
+import org.apache.nifi.util.StopWatch;
+import org.apache.nifi.util.file.FileUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.io.ByteArrayInputStream;
+import java.io.EOFException;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
@@ -57,24 +77,6 @@ import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.regex.Pattern;
-import org.apache.commons.io.IOUtils;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.nifi.controller.repository.claim.ContentClaim;
-import org.apache.nifi.controller.repository.claim.ResourceClaim;
-import org.apache.nifi.controller.repository.claim.ResourceClaimManager;
-import org.apache.nifi.controller.repository.claim.StandardContentClaim;
-import org.apache.nifi.controller.repository.io.LimitedInputStream;
-import org.apache.nifi.engine.FlowEngine;
-import org.apache.nifi.processor.DataUnit;
-import org.apache.nifi.stream.io.ByteCountingOutputStream;
-import org.apache.nifi.stream.io.StreamUtils;
-import org.apache.nifi.stream.io.SynchronizedByteCountingOutputStream;
-import org.apache.nifi.util.FormatUtils;
-import org.apache.nifi.util.NiFiProperties;
-import org.apache.nifi.util.StopWatch;
-import org.apache.nifi.util.file.FileUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* Is thread safe
@@ -882,11 +884,20 @@ public class FileSystemRepository implements ContentRepository {
if (claim.getOffset() > 0L) {
try {
StreamUtils.skip(fis, claim.getOffset());
- } catch (IOException ioe) {
+ } catch (final EOFException eof) {
+ final long resourceClaimBytes;
+ try {
+ resourceClaimBytes = Files.size(path);
+ } catch (final IOException e) {
+ throw new ContentNotFoundException(claim, "Content Claim has an offset of " + claim.getOffset()
+ + " but Resource Claim has fewer than this many bytes (actual length of the resource claim could not be determined)");
+ }
+
+ throw new ContentNotFoundException(claim, "Content Claim has an offset of " + claim.getOffset() + " but Resource Claim " + path + " is only " + resourceClaimBytes + " bytes");
+ } catch (final IOException ioe) {
IOUtils.closeQuietly(fis);
throw ioe;
}
-
}
// A claim length of -1 indicates that the claim is still being written to and we don't know
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
index f7249c0..c758c3d 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
@@ -56,6 +56,7 @@ import org.apache.nifi.provenance.StandardProvenanceEventRecord;
import org.apache.nifi.stream.io.ByteCountingInputStream;
import org.apache.nifi.stream.io.ByteCountingOutputStream;
import org.apache.nifi.stream.io.StreamUtils;
+import org.rocksdb.Checkpoint;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -2361,7 +2362,14 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
throw new FlowFileAccessException("Failed to access ContentClaim for " + source.toString(), e);
}
- final InputStream rawIn = getInputStream(source, record.getCurrentClaim(), record.getCurrentClaimOffset(), true);
+ final InputStream rawIn;
+ try {
+ rawIn = getInputStream(source, record.getCurrentClaim(), record.getCurrentClaimOffset(), true);
+ } catch (final ContentNotFoundException nfe) {
+ handleContentNotFound(nfe, record);
+ throw nfe;
+ }
+
final InputStream limitedIn = new LimitedInputStream(rawIn, source.getSize());
final ByteCountingInputStream countingStream = new ByteCountingInputStream(limitedIn);
final FlowFileAccessInputStream ffais = new FlowFileAccessInputStream(countingStream, source, record.getCurrentClaim());
@@ -2375,13 +2383,13 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
try {
return ffais.read();
} catch (final ContentNotFoundException cnfe) {
- handleContentNotFound(cnfe, record);
close();
+ handleContentNotFound(cnfe, record);
throw cnfe;
} catch (final FlowFileAccessException ffae) {
LOG.error("Failed to read content from " + sourceFlowFile + "; rolling back session", ffae);
- rollback(true);
close();
+ rollback(true);
throw ffae;
}
}
@@ -2396,13 +2404,13 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
try {
return ffais.read(b, off, len);
} catch (final ContentNotFoundException cnfe) {
- handleContentNotFound(cnfe, record);
close();
+ handleContentNotFound(cnfe, record);
throw cnfe;
} catch (final FlowFileAccessException ffae) {
LOG.error("Failed to read content from " + sourceFlowFile + "; rolling back session", ffae);
- rollback(true);
close();
+ rollback(true);
throw ffae;
}
}
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestFileSystemRepository.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestFileSystemRepository.java
index bf1a579..6ac10d5 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestFileSystemRepository.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestFileSystemRepository.java
@@ -21,6 +21,7 @@ import ch.qos.logback.classic.Logger;
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.read.ListAppender;
import org.apache.nifi.controller.repository.claim.ContentClaim;
+import org.apache.nifi.controller.repository.claim.ResourceClaim;
import org.apache.nifi.controller.repository.claim.StandardContentClaim;
import org.apache.nifi.controller.repository.claim.StandardResourceClaim;
import org.apache.nifi.controller.repository.claim.StandardResourceClaimManager;
@@ -29,6 +30,7 @@ import org.apache.nifi.processor.DataUnit;
import org.apache.nifi.stream.io.StreamUtils;
import org.apache.nifi.util.NiFiProperties;
import org.junit.After;
+import org.junit.Assert;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
@@ -37,10 +39,12 @@ import org.slf4j.LoggerFactory;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
+import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.lang.reflect.Method;
+import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
@@ -165,6 +169,50 @@ public class TestFileSystemRepository {
}
@Test
+ public void testContentNotFoundExceptionThrownIfResourceClaimTooShort() throws IOException {
+ final File contentFile = new File("target/content_repository/0/0.bin");
+ try (final OutputStream fos = new FileOutputStream(contentFile)) {
+ fos.write("Hello World".getBytes(StandardCharsets.UTF_8));
+ }
+
+ final ResourceClaim resourceClaim = new StandardResourceClaim(claimManager, "default", "0", "0.bin", false);
+ final StandardContentClaim existingContentClaim = new StandardContentClaim(resourceClaim, 0);
+ existingContentClaim.setLength(11);
+
+ try (final InputStream in = repository.read(existingContentClaim)) {
+ final byte[] buff = new byte[11];
+ StreamUtils.fillBuffer(in, buff);
+ assertEquals("Hello World", new String(buff, StandardCharsets.UTF_8));
+ }
+
+ final StandardContentClaim halfContentClaim = new StandardContentClaim(resourceClaim, 6);
+ halfContentClaim.setLength(5);
+
+ try (final InputStream in = repository.read(halfContentClaim)) {
+ final byte[] buff = new byte[5];
+ StreamUtils.fillBuffer(in, buff);
+ assertEquals("World", new String(buff, StandardCharsets.UTF_8));
+ }
+
+ final StandardContentClaim emptyContentClaim = new StandardContentClaim(resourceClaim, 11);
+ existingContentClaim.setLength(0);
+
+ try (final InputStream in = repository.read(emptyContentClaim)) {
+ assertEquals(-1, in.read());
+ }
+
+ final StandardContentClaim missingContentClaim = new StandardContentClaim(resourceClaim, 12);
+ missingContentClaim.setLength(1);
+
+ try {
+ repository.read(missingContentClaim);
+ Assert.fail("Did not throw ContentNotFoundException");
+ } catch (final ContentNotFoundException cnfe) {
+ // Expected
+ }
+ }
+
+ @Test
public void testBogusFile() throws IOException {
repository.shutdown();
System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, TestFileSystemRepository.class.getResource("/conf/nifi.properties").getFile());