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/09/23 21:32:14 UTC
[1/4] nifi git commit: NIFI-713: Infer Hadoop Compression
Automatically
Repository: nifi
Updated Branches:
refs/heads/master 8615941c8 -> 26f80095b
NIFI-713: Infer Hadoop Compression Automatically
- Three main types of compression options:
NONE : no compression
AUTOMATIC : infers codec by extension
SPECIFIED : specified codec (e.g. snappy, gzip, bzip, or lz4)
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/c72fb201
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/c72fb201
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/c72fb201
Branch: refs/heads/master
Commit: c72fb201d52113a16b176be3e0337b4303124244
Parents: 2bb7853
Author: ricky <ri...@cloudera.com>
Authored: Mon Aug 31 14:38:15 2015 -0400
Committer: ricky <ri...@cloudera.com>
Committed: Mon Aug 31 14:38:15 2015 -0400
----------------------------------------------------------------------
.../hadoop/AbstractHadoopProcessor.java | 39 +++++++++++--
.../hadoop/CreateHadoopSequenceFile.java | 4 +-
.../apache/nifi/processors/hadoop/GetHDFS.java | 32 +++++++++--
.../apache/nifi/processors/hadoop/PutHDFS.java | 15 ++---
.../nifi/processors/hadoop/GetHDFSTest.java | 59 ++++++++++++++------
5 files changed, 111 insertions(+), 38 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/c72fb201/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java
index 8519d2c..0102b1f 100644
--- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java
+++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java
@@ -34,7 +34,6 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.compress.BZip2Codec;
-import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.io.compress.GzipCodec;
@@ -59,6 +58,34 @@ import org.apache.nifi.util.Tuple;
* This is a base class that is helpful when building processors interacting with HDFS.
*/
public abstract class AbstractHadoopProcessor extends AbstractProcessor {
+ /**
+ * Compression Type Enum
+ */
+ public enum CompressionType {
+ NONE,
+ DEFAULT,
+ BZIP,
+ GZIP,
+ LZ4,
+ SNAPPY,
+ AUTOMATIC;
+
+ @Override
+ public String toString() {
+ switch (this) {
+ case NONE: return "NONE";
+ case DEFAULT: return DefaultCodec.class.getName();
+ case BZIP: return BZip2Codec.class.getName();
+ case GZIP: return GzipCodec.class.getName();
+ case LZ4: return Lz4Codec.class.getName();
+ case SNAPPY: return SnappyCodec.class.getName();
+ case AUTOMATIC: return "Automatically Detected";
+ }
+ return null;
+ }
+ }
+
+
private static final Validator KERBEROS_CONFIG_VALIDATOR = new Validator() {
@Override
public ValidationResult validate(String subject, String input, ValidationContext context) {
@@ -94,8 +121,8 @@ public abstract class AbstractHadoopProcessor extends AbstractProcessor {
public static final String DIRECTORY_PROP_NAME = "Directory";
- public static final PropertyDescriptor COMPRESSION_CODEC = new PropertyDescriptor.Builder().name("Compression codec").required(false)
- .allowableValues(BZip2Codec.class.getName(), DefaultCodec.class.getName(), GzipCodec.class.getName(), Lz4Codec.class.getName(), SnappyCodec.class.getName()).build();
+ public static final PropertyDescriptor COMPRESSION_CODEC = new PropertyDescriptor.Builder().name("Compression codec").required(true)
+ .allowableValues(CompressionType.values()).defaultValue(CompressionType.NONE.toString()).build();
public static final PropertyDescriptor KERBEROS_PRINCIPAL = new PropertyDescriptor.Builder().name("Kerberos Principal").required(false)
.description("Kerberos principal to authenticate as. Requires nifi.kerberos.krb5.file to be set " + "in your nifi.properties").addValidator(Validator.VALID)
@@ -324,10 +351,10 @@ public abstract class AbstractHadoopProcessor extends AbstractProcessor {
* the Hadoop Configuration
* @return CompressionCodec or null
*/
- protected CompressionCodec getCompressionCodec(ProcessContext context, Configuration configuration) {
- CompressionCodec codec = null;
+ protected org.apache.hadoop.io.compress.CompressionCodec getCompressionCodec(ProcessContext context, Configuration configuration) {
+ org.apache.hadoop.io.compress.CompressionCodec codec = null;
if (context.getProperty(COMPRESSION_CODEC).isSet()) {
- String compressionClassname = context.getProperty(COMPRESSION_CODEC).getValue();
+ String compressionClassname = CompressionType.valueOf(context.getProperty(COMPRESSION_CODEC).getValue()).toString();
CompressionCodecFactory ccf = new CompressionCodecFactory(configuration);
codec = ccf.getCodecByClassName(compressionClassname);
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/c72fb201/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/CreateHadoopSequenceFile.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/CreateHadoopSequenceFile.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/CreateHadoopSequenceFile.java
index 186a290..1f55164 100644
--- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/CreateHadoopSequenceFile.java
+++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/CreateHadoopSequenceFile.java
@@ -22,6 +22,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.SequenceFile.CompressionType;
import org.apache.nifi.annotation.behavior.SideEffectFree;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
@@ -152,7 +153,8 @@ public class CreateHadoopSequenceFile extends AbstractHadoopProcessor {
sequenceFileWriter = new SequenceFileWriterImpl();
}
String value = context.getProperty(COMPRESSION_TYPE).getValue();
- CompressionType compressionType = value == null ? CompressionType.valueOf(DEFAULT_COMPRESSION_TYPE) : CompressionType.valueOf(value);
+ SequenceFile.CompressionType compressionType = value == null ?
+ SequenceFile.CompressionType.valueOf(DEFAULT_COMPRESSION_TYPE) : SequenceFile.CompressionType.valueOf(value);
final String fileName = flowFile.getAttribute(CoreAttributes.FILENAME.key()) + ".sf";
flowFile = session.putAttribute(flowFile, CoreAttributes.FILENAME.key(), fileName);
try {
http://git-wip-us.apache.org/repos/asf/nifi/blob/c72fb201/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/GetHDFS.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/GetHDFS.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/GetHDFS.java
index cac61b0..de776d4 100644
--- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/GetHDFS.java
+++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/GetHDFS.java
@@ -33,18 +33,20 @@ import java.util.concurrent.locks.ReentrantLock;
import java.util.regex.Pattern;
import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.nifi.annotation.behavior.TriggerWhenEmpty;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
-import org.apache.nifi.annotation.behavior.WritesAttribute;
-import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.ValidationContext;
@@ -332,6 +334,7 @@ public class GetHDFS extends AbstractHadoopProcessor {
protected void processBatchOfFiles(final List<Path> files, final ProcessContext context, final ProcessSession session) {
// process the batch of files
InputStream stream = null;
+ CompressionCodec codec = null;
Configuration conf = getConfiguration();
FileSystem hdfs = getFileSystem();
final boolean keepSourceFiles = context.getProperty(KEEP_SOURCE_FILE).asBoolean();
@@ -339,19 +342,36 @@ public class GetHDFS extends AbstractHadoopProcessor {
int bufferSize = bufferSizeProp != null ? bufferSizeProp.intValue() : conf.getInt(BUFFER_SIZE_KEY,
BUFFER_SIZE_DEFAULT);
final Path rootDir = new Path(context.getProperty(DIRECTORY).getValue());
- final CompressionCodec codec = getCompressionCodec(context, conf);
+
+ final CompressionType compressionType = CompressionType.valueOf(context.getProperty(COMPRESSION_CODEC).toString());
+ final boolean inferCompressionCodec = compressionType == CompressionType.AUTOMATIC;
+ if (inferCompressionCodec || compressionType != CompressionType.NONE) {
+ codec = getCompressionCodec(context, getConfiguration());
+ }
+ final CompressionCodecFactory compressionCodecFactory = new CompressionCodecFactory(conf);
for (final Path file : files) {
try {
if (!hdfs.exists(file)) {
continue; // if file is no longer there then move on
}
- final String filename = file.getName();
+ final String originalFilename = file.getName();
final String relativePath = getPathDifference(rootDir, file);
stream = hdfs.open(file, bufferSize);
+
+ final String outputFilename;
+ // Check if we should infer compression codec
+ if (inferCompressionCodec) {
+ codec = compressionCodecFactory.getCodec(file);
+ }
+ // Check if compression codec is defined (inferred or otherwise)
if (codec != null) {
stream = codec.createInputStream(stream);
+ outputFilename = StringUtils.removeEnd(originalFilename, codec.getDefaultExtension());
+ } else {
+ outputFilename = originalFilename;
}
+
FlowFile flowFile = session.create();
final StopWatch stopWatch = new StopWatch(true);
@@ -361,7 +381,7 @@ public class GetHDFS extends AbstractHadoopProcessor {
final long millis = stopWatch.getDuration(TimeUnit.MILLISECONDS);
flowFile = session.putAttribute(flowFile, CoreAttributes.PATH.key(), relativePath);
- flowFile = session.putAttribute(flowFile, CoreAttributes.FILENAME.key(), filename);
+ flowFile = session.putAttribute(flowFile, CoreAttributes.FILENAME.key(), outputFilename);
if (!keepSourceFiles && !hdfs.delete(file, false)) {
getLogger().warn("Could not remove {} from HDFS. Not ingesting this file ...",
@@ -370,7 +390,7 @@ public class GetHDFS extends AbstractHadoopProcessor {
continue;
}
- final String transitUri = (filename.startsWith("/")) ? "hdfs:/" + filename : "hdfs://" + filename;
+ final String transitUri = (originalFilename.startsWith("/")) ? "hdfs:/" + originalFilename : "hdfs://" + originalFilename;
session.getProvenanceReporter().receive(flowFile, transitUri);
session.transfer(flowFile, REL_SUCCESS);
getLogger().info("retrieved {} from HDFS {} in {} milliseconds at a rate of {}",
http://git-wip-us.apache.org/repos/asf/nifi/blob/c72fb201/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/PutHDFS.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/PutHDFS.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/PutHDFS.java
index 419b1de..901159b 100644
--- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/PutHDFS.java
+++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/PutHDFS.java
@@ -219,13 +219,14 @@ public class PutHDFS extends AbstractHadoopProcessor {
final CompressionCodec codec = getCompressionCodec(context, configuration);
+ final String filename = codec != null
+ ? flowFile.getAttribute(CoreAttributes.FILENAME.key()) + codec.getDefaultExtension()
+ : flowFile.getAttribute(CoreAttributes.FILENAME.key());
+
Path tempDotCopyFile = null;
try {
- final Path tempCopyFile;
- final Path copyFile;
-
- tempCopyFile = new Path(configuredRootDirPath, "." + flowFile.getAttribute(CoreAttributes.FILENAME.key()));
- copyFile = new Path(configuredRootDirPath, flowFile.getAttribute(CoreAttributes.FILENAME.key()));
+ final Path tempCopyFile = new Path(configuredRootDirPath, "." + filename);
+ final Path copyFile = new Path(configuredRootDirPath, filename);
// Create destination directory if it does not exist
try {
@@ -327,8 +328,8 @@ public class PutHDFS extends AbstractHadoopProcessor {
getLogger().info("copied {} to HDFS at {} in {} milliseconds at a rate of {}",
new Object[]{flowFile, copyFile, millis, dataRate});
- final String filename = copyFile.toString();
- final String transitUri = (filename.startsWith("/")) ? "hdfs:/" + filename : "hdfs://" + filename;
+ final String outputPath = copyFile.toString();
+ final String transitUri = (outputPath.startsWith("/")) ? "hdfs:/" + outputPath : "hdfs://" + outputPath;
session.getProvenanceReporter().send(flowFile, transitUri);
session.transfer(flowFile, REL_SUCCESS);
http://git-wip-us.apache.org/repos/asf/nifi/blob/c72fb201/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/GetHDFSTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/GetHDFSTest.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/GetHDFSTest.java
index b0dd17f..e8714dd 100644
--- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/GetHDFSTest.java
+++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/GetHDFSTest.java
@@ -33,8 +33,6 @@ import org.apache.nifi.util.MockProcessContext;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.compress.CompressionCodec;
-import org.apache.hadoop.io.compress.GzipCodec;
import org.junit.Assert;
import org.junit.Test;
@@ -108,19 +106,6 @@ public class GetHDFSTest {
for (ValidationResult vr : results) {
Assert.assertTrue(vr.toString().contains("is invalid because Minimum File Age cannot be greater than Maximum File Age"));
}
-
- results = new HashSet<>();
- runner.setProperty(GetHDFS.DIRECTORY, "/target");
- runner.setProperty(GetHDFS.COMPRESSION_CODEC, CompressionCodec.class.getName());
- runner.enqueue(new byte[0]);
- pc = runner.getProcessContext();
- if (pc instanceof MockProcessContext) {
- results = ((MockProcessContext) pc).validate();
- }
- Assert.assertEquals(1, results.size());
- for (ValidationResult vr : results) {
- Assert.assertTrue(vr.toString().contains("is invalid because Given value not found in allowed set"));
- }
}
@Test
@@ -138,18 +123,56 @@ public class GetHDFSTest {
}
@Test
- public void testGetFilesWithCompression() throws IOException {
+ public void testAutomaticDecompression() throws IOException {
TestRunner runner = TestRunners.newTestRunner(GetHDFS.class);
runner.setProperty(PutHDFS.DIRECTORY, "src/test/resources/testdata");
runner.setProperty(GetHDFS.FILE_FILTER_REGEX, "random.*.gz");
- runner.setProperty(GetHDFS.COMPRESSION_CODEC, GzipCodec.class.getName());
runner.setProperty(GetHDFS.KEEP_SOURCE_FILE, "true");
+ runner.setProperty(GetHDFS.COMPRESSION_CODEC, "AUTOMATIC");
runner.run();
+
List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(GetHDFS.REL_SUCCESS);
assertEquals(1, flowFiles.size());
+
MockFlowFile flowFile = flowFiles.get(0);
- assertTrue(flowFile.getAttribute(CoreAttributes.FILENAME.key()).startsWith("randombytes-1.gz"));
+ assertTrue(flowFile.getAttribute(CoreAttributes.FILENAME.key()).equals("randombytes-1"));
InputStream expected = getClass().getResourceAsStream("/testdata/randombytes-1");
flowFile.assertContentEquals(expected);
}
+
+ @Test
+ public void testInferCompressionCodecDisabled() throws IOException {
+ TestRunner runner = TestRunners.newTestRunner(GetHDFS.class);
+ runner.setProperty(PutHDFS.DIRECTORY, "src/test/resources/testdata");
+ runner.setProperty(GetHDFS.FILE_FILTER_REGEX, "random.*.gz");
+ runner.setProperty(GetHDFS.KEEP_SOURCE_FILE, "true");
+ runner.setProperty(GetHDFS.COMPRESSION_CODEC, "NONE");
+ runner.run();
+
+ List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(GetHDFS.REL_SUCCESS);
+ assertEquals(1, flowFiles.size());
+
+ MockFlowFile flowFile = flowFiles.get(0);
+ assertTrue(flowFile.getAttribute(CoreAttributes.FILENAME.key()).equals("randombytes-1.gz"));
+ InputStream expected = getClass().getResourceAsStream("/testdata/randombytes-1.gz");
+ flowFile.assertContentEquals(expected);
+ }
+
+ @Test
+ public void testFileExtensionNotACompressionCodec() throws IOException {
+ TestRunner runner = TestRunners.newTestRunner(GetHDFS.class);
+ runner.setProperty(PutHDFS.DIRECTORY, "src/test/resources/testdata");
+ runner.setProperty(GetHDFS.FILE_FILTER_REGEX, ".*.zip");
+ runner.setProperty(GetHDFS.KEEP_SOURCE_FILE, "true");
+ runner.setProperty(GetHDFS.COMPRESSION_CODEC, "AUTOMATIC");
+ runner.run();
+
+ List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(GetHDFS.REL_SUCCESS);
+ assertEquals(1, flowFiles.size());
+
+ MockFlowFile flowFile = flowFiles.get(0);
+ assertTrue(flowFile.getAttribute(CoreAttributes.FILENAME.key()).equals("13545423550275052.zip"));
+ InputStream expected = getClass().getResourceAsStream("/testdata/13545423550275052.zip");
+ flowFile.assertContentEquals(expected);
+ }
}
[3/4] nifi git commit: NIFI-713: Fixed checkstyle issue
Posted by ma...@apache.org.
NIFI-713: Fixed checkstyle issue
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/f884086f
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/f884086f
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/f884086f
Branch: refs/heads/master
Commit: f884086fc2085cc0dbf5e1a80bca4a6a5138e42a
Parents: da56b49
Author: Mark Payne <ma...@hotmail.com>
Authored: Wed Sep 23 15:20:25 2015 -0400
Committer: Mark Payne <ma...@hotmail.com>
Committed: Wed Sep 23 15:20:25 2015 -0400
----------------------------------------------------------------------
.../apache/nifi/processors/hadoop/CreateHadoopSequenceFile.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/f884086f/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/CreateHadoopSequenceFile.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/CreateHadoopSequenceFile.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/CreateHadoopSequenceFile.java
index 1f55164..cbcc54d 100644
--- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/CreateHadoopSequenceFile.java
+++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/CreateHadoopSequenceFile.java
@@ -153,8 +153,8 @@ public class CreateHadoopSequenceFile extends AbstractHadoopProcessor {
sequenceFileWriter = new SequenceFileWriterImpl();
}
String value = context.getProperty(COMPRESSION_TYPE).getValue();
- SequenceFile.CompressionType compressionType = value == null ?
- SequenceFile.CompressionType.valueOf(DEFAULT_COMPRESSION_TYPE) : SequenceFile.CompressionType.valueOf(value);
+ SequenceFile.CompressionType compressionType = value == null
+ ? SequenceFile.CompressionType.valueOf(DEFAULT_COMPRESSION_TYPE) : SequenceFile.CompressionType.valueOf(value);
final String fileName = flowFile.getAttribute(CoreAttributes.FILENAME.key()) + ".sf";
flowFile = session.putAttribute(flowFile, CoreAttributes.FILENAME.key(), fileName);
try {
[2/4] nifi git commit: Merge branch 'NIFI-713' of
https://github.com/rickysaltzer/nifi into NIFI-713
Posted by ma...@apache.org.
Merge branch 'NIFI-713' of https://github.com/rickysaltzer/nifi into NIFI-713
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/da56b49b
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/da56b49b
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/da56b49b
Branch: refs/heads/master
Commit: da56b49b88ee2e6491d893d2a6571cf457bb032c
Parents: 1fd4595 c72fb20
Author: Mark Payne <ma...@hotmail.com>
Authored: Wed Sep 23 12:06:50 2015 -0400
Committer: Mark Payne <ma...@hotmail.com>
Committed: Wed Sep 23 12:06:50 2015 -0400
----------------------------------------------------------------------
.../hadoop/AbstractHadoopProcessor.java | 39 +++++++++++--
.../hadoop/CreateHadoopSequenceFile.java | 4 +-
.../apache/nifi/processors/hadoop/GetHDFS.java | 32 +++++++++--
.../apache/nifi/processors/hadoop/PutHDFS.java | 15 ++---
.../nifi/processors/hadoop/GetHDFSTest.java | 59 ++++++++++++++------
5 files changed, 111 insertions(+), 38 deletions(-)
----------------------------------------------------------------------
[4/4] nifi git commit: Merge branch 'master' of
http://git-wip-us.apache.org/repos/asf/nifi
Posted by ma...@apache.org.
Merge branch 'master' of http://git-wip-us.apache.org/repos/asf/nifi
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/26f80095
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/26f80095
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/26f80095
Branch: refs/heads/master
Commit: 26f80095b7f7db4b41f66e15b81268a2fbb19682
Parents: f884086 8615941
Author: Mark Payne <ma...@hotmail.com>
Authored: Wed Sep 23 15:31:19 2015 -0400
Committer: Mark Payne <ma...@hotmail.com>
Committed: Wed Sep 23 15:31:19 2015 -0400
----------------------------------------------------------------------
.../apache/nifi/processor/ProcessContext.java | 13 +++
.../apache/nifi/util/MockProcessContext.java | 33 ++++++++
.../nifi/util/StandardProcessorTestRunner.java | 25 ++++++
.../java/org/apache/nifi/util/TestRunner.java | 37 +++++++++
.../documentation/mock/MockProcessContext.java | 12 +++
.../apache/nifi/controller/ProcessorNode.java | 1 +
.../scheduling/ConnectableProcessContext.java | 13 +++
.../nifi/processor/StandardProcessContext.java | 11 +++
.../processor/StandardSchedulingContext.java | 10 +++
.../nifi-image-bundle/nifi-image-nar/pom.xml | 6 ++
.../nifi-image-bundle/nifi-image-viewer/pom.xml | 48 +++++++++++
.../apache/nifi/web/ImageViewerController.java | 52 ++++++++++++
.../main/webapp/META-INF/nifi-content-viewer | 17 ++++
.../src/main/webapp/WEB-INF/jsp/image.jsp | 39 +++++++++
.../src/main/webapp/WEB-INF/web.xml | 29 +++++++
nifi-nar-bundles/nifi-image-bundle/pom.xml | 1 +
.../processors/solr/PutSolrContentStream.java | 2 +-
.../solr/TestPutSolrContentStream.java | 54 +++++++++++++
.../nifi/processors/standard/ExecuteSQL.java | 17 ++--
.../nifi/processors/standard/MergeContent.java | 5 +-
.../standard/TestDetectDuplicate.java | 2 +-
.../processors/standard/TestExecuteSQL.java | 84 ++++++++++++++------
.../processors/standard/TestMergeContent.java | 21 +++++
23 files changed, 498 insertions(+), 34 deletions(-)
----------------------------------------------------------------------