camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ggrzy...@apache.org
Subject [1/2] camel git commit: [CAMEL-8434] Allow consuming empty files in camel-hdfs
Date Wed, 04 Mar 2015 11:00:02 GMT
Repository: camel
Updated Branches:
  refs/heads/master a547de166 -> d84f9b6e4


[CAMEL-8434] Allow consuming empty files in camel-hdfs


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/d84f9b6e
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/d84f9b6e
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/d84f9b6e

Branch: refs/heads/master
Commit: d84f9b6e44c05c9954d57630903af6fde59336ac
Parents: 4e314d2
Author: Grzegorz Grzybek <gr.grzybek@gmail.com>
Authored: Wed Mar 4 11:57:59 2015 +0100
Committer: Grzegorz Grzybek <gr.grzybek@gmail.com>
Committed: Wed Mar 4 11:58:12 2015 +0100

----------------------------------------------------------------------
 .../camel/component/hdfs/HdfsConsumer.java      |  2 +-
 .../camel/component/hdfs/HdfsFileType.java      |  3 +-
 .../camel/component/hdfs/HdfsInputStream.java   | 11 ++-
 .../camel/component/hdfs/HdfsConsumerTest.java  | 84 ++++++++++++++++++++
 4 files changed, 96 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/d84f9b6e/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsConsumer.java
b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsConsumer.java
index 367f418..f718238 100644
--- a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsConsumer.java
+++ b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsConsumer.java
@@ -145,7 +145,7 @@ public final class HdfsConsumer extends ScheduledPollConsumer {
             try {
                 Holder<Object> key = new Holder<Object>();
                 Holder<Object> value = new Holder<Object>();
-                while (this.istream.next(key, value) != 0) {
+                while (this.istream.next(key, value) >= 0) {
                     Exchange exchange = this.getEndpoint().createExchange();
                     Message message = new DefaultMessage();
                     String fileName = StringUtils.substringAfterLast(status.getPath().toString(),
"/");

http://git-wip-us.apache.org/repos/asf/camel/blob/d84f9b6e/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsFileType.java
----------------------------------------------------------------------
diff --git a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsFileType.java
b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsFileType.java
index 97c174f..ff4036d 100644
--- a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsFileType.java
+++ b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsFileType.java
@@ -84,7 +84,8 @@ public enum HdfsFileType {
                     return bytesRead;
                 } else {
                     key.value = null;
-                    value.value = null;
+                    // indication that we may have read from empty file
+                    value.value = bos;
                     return 0;
                 }
             } catch (IOException ex) {

http://git-wip-us.apache.org/repos/asf/camel/blob/d84f9b6e/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsInputStream.java
----------------------------------------------------------------------
diff --git a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsInputStream.java
b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsInputStream.java
index ac1bff9..24342f6 100644
--- a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsInputStream.java
+++ b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsInputStream.java
@@ -64,11 +64,18 @@ public class HdfsInputStream implements Closeable {
 
     public final long next(Holder<Object> key, Holder<Object> value) {
         long nb = fileType.next(this, key, value);
-        if (nb > 0) {
+        // when zero bytes was read from given type of file, we may still have a record (e.g.,
empty file)
+        // null value.value is the only indication that no (new) record/chunk was read
+        if (nb == 0 && numOfReadBytes.get() > 0) {
+            // we've read all chunks from file, which size is exact multiple the chunk size
+            return -1;
+        }
+        if (value.value != null) {
             numOfReadBytes.addAndGet(nb);
             numOfReadMessages.incrementAndGet();
+            return nb;
         }
-        return nb;
+        return -1;
     }
 
     public final long getNumOfReadBytes() {

http://git-wip-us.apache.org/repos/asf/camel/blob/d84f9b6e/components/camel-hdfs/src/test/java/org/apache/camel/component/hdfs/HdfsConsumerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-hdfs/src/test/java/org/apache/camel/component/hdfs/HdfsConsumerTest.java
b/components/camel-hdfs/src/test/java/org/apache/camel/component/hdfs/HdfsConsumerTest.java
index 368b1d8..1cda88c 100644
--- a/components/camel-hdfs/src/test/java/org/apache/camel/component/hdfs/HdfsConsumerTest.java
+++ b/components/camel-hdfs/src/test/java/org/apache/camel/component/hdfs/HdfsConsumerTest.java
@@ -16,6 +16,7 @@
  */
 package org.apache.camel.component.hdfs;
 
+import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -107,6 +108,89 @@ public class HdfsConsumerTest extends HdfsTestSupport {
     }
 
     @Test
+    public void testSimpleConsumerWithEmptyFile() throws Exception {
+        if (!canTest()) {
+            return;
+        }
+
+        final Path file = new Path(new File("target/test/test-camel-normal-file").getAbsolutePath());
+        Configuration conf = new Configuration();
+        FileSystem fs = FileSystem.get(file.toUri(), conf);
+        FSDataOutputStream out = fs.create(file);
+        out.close();
+
+        MockEndpoint resultEndpoint = context.getEndpoint("mock:result", MockEndpoint.class);
+        resultEndpoint.expectedMessageCount(1);
+
+        context.addRoutes(new RouteBuilder() {
+            public void configure() {
+                from("hdfs:///" + file.toUri() + "?fileSystemType=LOCAL&chunkSize=4096&initialDelay=0").to("mock:result");
+            }
+        });
+        context.start();
+
+        resultEndpoint.assertIsSatisfied();
+        assertThat(resultEndpoint.getReceivedExchanges().get(0).getIn().getBody(ByteArrayOutputStream.class).toByteArray().length,
equalTo(0));
+    }
+
+    @Test
+    public void testSimpleConsumerFileWithSizeEqualToNChunks() throws Exception {
+        if (!canTest()) {
+            return;
+        }
+
+        final Path file = new Path(new File("target/test/test-camel-normal-file").getAbsolutePath());
+        Configuration conf = new Configuration();
+        FileSystem fs = FileSystem.get(file.toUri(), conf);
+        FSDataOutputStream out = fs.create(file);
+        // size = 5 times chunk size = 210 bytes
+        for (int i = 0; i < 42; ++i) {
+            out.write(new byte[] { 0x61, 0x62, 0x63, 0x64, 0x65 });
+            out.flush();
+        }
+        out.close();
+
+        MockEndpoint resultEndpoint = context.getEndpoint("mock:result", MockEndpoint.class);
+        resultEndpoint.expectedMessageCount(5);
+
+        context.addRoutes(new RouteBuilder() {
+            public void configure() {
+                from("hdfs:///" + file.toUri() + "?fileSystemType=LOCAL&chunkSize=42&initialDelay=0").to("mock:result");
+            }
+        });
+        context.start();
+
+        resultEndpoint.assertIsSatisfied();
+        assertThat(resultEndpoint.getReceivedExchanges().get(0).getIn().getBody(ByteArrayOutputStream.class).toByteArray().length,
equalTo(42));
+    }
+
+    @Test
+    public void testSimpleConsumerWithEmptySequenceFile() throws Exception {
+        if (!canTest()) {
+            return;
+        }
+
+        final Path file = new Path(new File("target/test/test-camel-sequence-file").getAbsolutePath());
+        Configuration conf = new Configuration();
+        FileSystem fs1 = FileSystem.get(file.toUri(), conf);
+        SequenceFile.Writer writer = createWriter(fs1, conf, file, NullWritable.class, BooleanWritable.class);
+        writer.sync();
+        writer.close();
+
+        MockEndpoint resultEndpoint = context.getEndpoint("mock:result", MockEndpoint.class);
+        resultEndpoint.expectedMessageCount(0);
+
+        context.addRoutes(new RouteBuilder() {
+            public void configure() {
+                from("hdfs:///" + file.toUri() + "?fileSystemType=LOCAL&fileType=SEQUENCE_FILE&chunkSize=4096&initialDelay=0").to("mock:result");
+            }
+        });
+        context.start();
+
+        resultEndpoint.assertIsSatisfied();
+    }
+
+    @Test
     public void testReadWithReadSuffix() throws Exception {
         if (!canTest()) {
             return;


Mime
View raw message