Update CDH dependency to CDH3u3

.gitignore

@@ -1 +1,5 @@
 target
+/.settings/
+.DS_Store
+catalog.xml
+nbactions.xml

README.md

@@ -25,7 +25,7 @@ without the dependencies and one with the dependencies included.
 
 The [Spring Source bundle repository][2] is used for Spring dependencies. The
 (currently unpublished) Cloudera repository is used for Hadoop and its
-dependencies. Specifically, this builds against CDH2u1. Some dependencies are
+dependencies. Specifically, this builds against CDH3u3. Some dependencies are
 pulled from Maven Central, but this will move to the Spring Source repository
 to reduce potential metadata issues.
 

pom.xml

@@ -11,7 +11,7 @@
 
 		<repository>
 			<id>com.cloudera.repository.public</id>
-			<url>http://repository.cloudera.com/nexus/content/groups/public/</url>
+			<url>https://repository.cloudera.com/artifactory/public</url>
 			<releases>
 				<enabled>true</enabled>
 			</releases>
@@ -59,10 +59,9 @@
 				<artifactId>maven-assembly-plugin</artifactId>
 				<version>2.2-beta-5</version>
 				<configuration>
-					<descriptorRefs>
-						<descriptorRef>bin</descriptorRef>
-						<descriptorRef>jar-with-dependencies</descriptorRef>
-					</descriptorRefs>
+					<descriptors>
+						<descriptor>src/main/assembly/descriptor.xml</descriptor>
+					</descriptors>
 				</configuration>
 			</plugin>
 
@@ -158,7 +157,7 @@
 		<dependency>
 			<groupId>org.apache.hadoop</groupId>
 			<artifactId>hadoop-core</artifactId>
-			<version>0.20.1-CDH.169.88</version>
+			<version>0.20.2-cdh3u3</version>
 		</dependency>
 
 	</dependencies>

src/main/assembly/descriptor.xml

@@ -0,0 +1,24 @@
+
+<assembly xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0" 
+  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
+  <id>bin</id>
+  <formats>
+    <format>tar.gz</format>
+  </formats>
+  <includeBaseDirectory>true</includeBaseDirectory>
+  <fileSets>
+    <fileSet>
+      <directory>${project.build.directory}</directory>
+      <outputDirectory>/lib</outputDirectory>
+      <includes>
+        <include>*.jar</include>
+      </includes>
+    </fileSet>
+  </fileSets>
+  <dependencySets>
+    <dependencySet>
+      <outputDirectory>/lib</outputDirectory>
+    </dependencySet>
+  </dependencySets>
+</assembly>

src/main/java/com/cloudera/fbus/Application.java

@@ -16,6 +16,7 @@ public class Application {
 
   private static final Logger logger = LoggerFactory
       .getLogger(Application.class);
+  private static final String defaultConfigLocation = "classpath:fbus-default.xml";
   private static final String configLocation = "classpath:fbus.xml";
 
   public static void main(String[] args) {
@@ -34,7 +35,7 @@ public static void main(String[] args) {
   public void run(String[] args) {
     AbstractApplicationContext context;
 
-    context = new ClassPathXmlApplicationContext(configLocation);
+    context = new ClassPathXmlApplicationContext(defaultConfigLocation, configLocation);
 
     context.registerShutdownHook();
   }

src/main/java/com/cloudera/fbus/HDFSSequenceFileDirectoryDestination.java

@@ -0,0 +1,124 @@
+package com.cloudera.fbus;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileReader;
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.io.compress.SnappyCodec;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.integration.message.MessageBuilder;
+import org.springframework.util.Assert;
+
+/**
+ * A Spring Integration end point suitable for delivering messages with a
+ * {@link File} payload to the Hadoop Distributed File System.
+ * 
+ */
+public class HDFSSequenceFileDirectoryDestination {
+
+  private static final String tmpSuffix = ".tmp";
+  private static final Logger logger = LoggerFactory.getLogger(HDFSDirectoryDestination.class);
+  private File directory;
+  private FileSystem fileSystem;
+  private CompressionCodec compressionCodec = new SnappyCodec(); //Snappy is the default compression
+  private Configuration config;
+
+  public HDFSSequenceFileDirectoryDestination() throws IOException {
+    config = new Configuration();
+    fileSystem = FileSystem.get(config);
+  }
+
+  public void deliver(File file) throws DeliveryException {
+    Path source;
+    Path destinationTmp;
+    Path destination;
+
+    Assert.notNull(file, "File may not be null");
+    Assert.isTrue(file.canRead(), "File " + file
+        + " does not exist or is unreadable");
+
+    source = new Path(file.getPath());
+    destinationTmp = new Path(directory.getPath(), source.getName() + tmpSuffix);
+    destination = new Path(directory.getPath(), source.getName());
+
+    try {
+
+      copyFromLocalToSeqFile(file, destinationTmp);
+
+      if (!fileSystem.rename(destinationTmp, destination)) {
+        if (!fileSystem.delete(destinationTmp, false)) {
+          logger.warn(
+              "Failed to clean up temporary file at {} after rename failed. File will be retried.",
+              destinationTmp);
+        }
+
+        throw new IOException("Can't rename " + destinationTmp + " to "
+            + destination);
+      } else {
+        if (!file.delete()) {
+          /*
+           * Purposefully do not throw an exception here as it would cause retry
+           * as well. -esammer
+           */
+          logger.error(
+              "Unable to delete file {} after moving it to HDFS @ {}. File may be retransfered!",
+              file, destination);
+        }
+      }
+    } catch (Throwable e) {
+      throw DeliveryException.newWith(MessageBuilder.withPayload(file).build(),
+          e);
+    }
+  }
+
+  private void copyFromLocalToSeqFile(File file, Path destinationTmp) throws IOException, FileNotFoundException {
+    SequenceFile.Writer writer = null;
+    BufferedReader reader = null;
+
+
+    try {
+      writer = SequenceFile.createWriter(fileSystem, config, destinationTmp, NullWritable.class, Text.class, SequenceFile.CompressionType.BLOCK, compressionCodec);
+      reader = new BufferedReader(new FileReader(file));
+
+      String line = null;
+      Text text = new Text();
+      while ((line = reader.readLine()) != null) {
+        text.set(line);
+        writer.append(NullWritable.get(), text);
+      }
+    } finally {
+      if (writer != null) {
+        writer.close();
+      }
+      if (reader != null) {
+        reader.close();
+      }
+    }
+  }
+
+  public File getDirectory() {
+    return directory;
+  }
+
+  public void setDirectory(File directory) {
+    this.directory = directory;
+  }
+
+  public CompressionCodec getCompressionCodec() {
+    return compressionCodec;
+  }
+
+  public void setCompressionCodec(CompressionCodec compressionCodec) {
+    this.compressionCodec = compressionCodec;
+  }
+}

src/main/resources/fbus-default.xml

@@ -0,0 +1,15 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:integration="http://www.springframework.org/schema/integration"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
+
+	<!-- Compression codec beans. -->
+
+	<bean id="snappy" class="org.apache.hadoop.io.compress.SnappyCodec"/>
+	<bean id="gzip" class="org.apache.hadoop.io.compress.GzipCodec"/>
+	<bean id="deflate" class="org.apache.hadoop.io.compress.DeflateCodec"/>
+	<bean id="bzip2" class="org.apache.hadoop.io.compress.BZip2Codec"/>
+
+</beans>

src/test/resources/fbus.xml

@@ -29,7 +29,7 @@
 	<!-- Example HDFS destination. -->
 
 	<bean id="hdfsDirectoryDestination" class="com.cloudera.fbus.HDFSDirectoryDestination">
-		<property name="directory" value="/user/${user.name}/output" />
+		<property name="directory" value="/user/#{ systemProperties['user.name'] }/output" />
 	</bean>
 
 	<!-- Channel adapters attach end points to channels. -->