You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by pt...@apache.org on 2014/08/08 21:48:20 UTC
[09/50] [abbrv] git commit: add storm-hdfs as an external module
add storm-hdfs as an external module
Project: http://git-wip-us.apache.org/repos/asf/incubator-storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-storm/commit/2881ac9c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-storm/tree/2881ac9c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-storm/diff/2881ac9c
Branch: refs/heads/master
Commit: 2881ac9c85cb61bec44f8e6e3f1d638e2968abf2
Parents: cd0b118 4051c8a
Author: P. Taylor Goetz <pt...@gmail.com>
Authored: Thu May 29 15:24:13 2014 -0400
Committer: P. Taylor Goetz <pt...@gmail.com>
Committed: Thu May 29 15:24:13 2014 -0400
----------------------------------------------------------------------
external/storm-hdfs/.gitignore | 4 +
external/storm-hdfs/LICENSE | 202 +++++++++++
external/storm-hdfs/README.md | 338 +++++++++++++++++++
external/storm-hdfs/pom.xml | 81 +++++
.../storm/hdfs/bolt/AbstractHdfsBolt.java | 110 ++++++
.../org/apache/storm/hdfs/bolt/HdfsBolt.java | 129 +++++++
.../storm/hdfs/bolt/SequenceFileBolt.java | 146 ++++++++
.../hdfs/bolt/format/DefaultFileNameFormat.java | 86 +++++
.../hdfs/bolt/format/DefaultSequenceFormat.java | 52 +++
.../hdfs/bolt/format/DelimitedRecordFormat.java | 88 +++++
.../storm/hdfs/bolt/format/FileNameFormat.java | 42 +++
.../storm/hdfs/bolt/format/RecordFormat.java | 32 ++
.../storm/hdfs/bolt/format/SequenceFormat.java | 40 +++
.../hdfs/bolt/rotation/FileRotationPolicy.java | 51 +++
.../bolt/rotation/FileSizeRotationPolicy.java | 81 +++++
.../hdfs/bolt/rotation/NoRotationPolicy.java | 35 ++
.../storm/hdfs/bolt/sync/CountSyncPolicy.java | 46 +++
.../apache/storm/hdfs/bolt/sync/SyncPolicy.java | 46 +++
.../hdfs/common/rotation/MoveFileAction.java | 27 ++
.../hdfs/common/rotation/RotationAction.java | 12 +
.../hdfs/common/security/HdfsSecurityUtil.java | 48 +++
.../apache/storm/hdfs/trident/HdfsState.java | 295 ++++++++++++++++
.../storm/hdfs/trident/HdfsStateFactory.java | 29 ++
.../apache/storm/hdfs/trident/HdfsUpdater.java | 14 +
.../trident/format/DefaultFileNameFormat.java | 83 +++++
.../trident/format/DefaultSequenceFormat.java | 54 +++
.../trident/format/DelimitedRecordFormat.java | 83 +++++
.../hdfs/trident/format/FileNameFormat.java | 40 +++
.../storm/hdfs/trident/format/RecordFormat.java | 32 ++
.../hdfs/trident/format/SequenceFormat.java | 40 +++
.../trident/rotation/FileRotationPolicy.java | 50 +++
.../rotation/FileSizeRotationPolicy.java | 81 +++++
.../hdfs/trident/rotation/NoRotationPolicy.java | 35 ++
.../hdfs/trident/sync/CountSyncPolicy.java | 46 +++
.../storm/hdfs/trident/sync/SyncPolicy.java | 46 +++
.../storm/hdfs/bolt/HdfsFileTopology.java | 197 +++++++++++
.../storm/hdfs/bolt/SequenceFileTopology.java | 193 +++++++++++
.../storm/hdfs/trident/FixedBatchSpout.java | 80 +++++
.../storm/hdfs/trident/TridentFileTopology.java | 73 ++++
.../hdfs/trident/TridentSequenceTopology.java | 71 ++++
40 files changed, 3238 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2881ac9c/external/storm-hdfs/.gitignore
----------------------------------------------------------------------
diff --cc external/storm-hdfs/.gitignore
index 0000000,0000000..b1e997f
new file mode 100644
--- /dev/null
+++ b/external/storm-hdfs/.gitignore
@@@ -1,0 -1,0 +1,4 @@@
++target/
++*.iml
++*.ipr
++*.iws
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2881ac9c/external/storm-hdfs/LICENSE
----------------------------------------------------------------------
diff --cc external/storm-hdfs/LICENSE
index 0000000,0000000..e06d208
new file mode 100644
--- /dev/null
+++ b/external/storm-hdfs/LICENSE
@@@ -1,0 -1,0 +1,202 @@@
++Apache License
++ Version 2.0, January 2004
++ http://www.apache.org/licenses/
++
++ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
++
++ 1. Definitions.
++
++ "License" shall mean the terms and conditions for use, reproduction,
++ and distribution as defined by Sections 1 through 9 of this document.
++
++ "Licensor" shall mean the copyright owner or entity authorized by
++ the copyright owner that is granting the License.
++
++ "Legal Entity" shall mean the union of the acting entity and all
++ other entities that control, are controlled by, or are under common
++ control with that entity. For the purposes of this definition,
++ "control" means (i) the power, direct or indirect, to cause the
++ direction or management of such entity, whether by contract or
++ otherwise, or (ii) ownership of fifty percent (50%) or more of the
++ outstanding shares, or (iii) beneficial ownership of such entity.
++
++ "You" (or "Your") shall mean an individual or Legal Entity
++ exercising permissions granted by this License.
++
++ "Source" form shall mean the preferred form for making modifications,
++ including but not limited to software source code, documentation
++ source, and configuration files.
++
++ "Object" form shall mean any form resulting from mechanical
++ transformation or translation of a Source form, including but
++ not limited to compiled object code, generated documentation,
++ and conversions to other media types.
++
++ "Work" shall mean the work of authorship, whether in Source or
++ Object form, made available under the License, as indicated by a
++ copyright notice that is included in or attached to the work
++ (an example is provided in the Appendix below).
++
++ "Derivative Works" shall mean any work, whether in Source or Object
++ form, that is based on (or derived from) the Work and for which the
++ editorial revisions, annotations, elaborations, or other modifications
++ represent, as a whole, an original work of authorship. For the purposes
++ of this License, Derivative Works shall not include works that remain
++ separable from, or merely link (or bind by name) to the interfaces of,
++ the Work and Derivative Works thereof.
++
++ "Contribution" shall mean any work of authorship, including
++ the original version of the Work and any modifications or additions
++ to that Work or Derivative Works thereof, that is intentionally
++ submitted to Licensor for inclusion in the Work by the copyright owner
++ or by an individual or Legal Entity authorized to submit on behalf of
++ the copyright owner. For the purposes of this definition, "submitted"
++ means any form of electronic, verbal, or written communication sent
++ to the Licensor or its representatives, including but not limited to
++ communication on electronic mailing lists, source code control systems,
++ and issue tracking systems that are managed by, or on behalf of, the
++ Licensor for the purpose of discussing and improving the Work, but
++ excluding communication that is conspicuously marked or otherwise
++ designated in writing by the copyright owner as "Not a Contribution."
++
++ "Contributor" shall mean Licensor and any individual or Legal Entity
++ on behalf of whom a Contribution has been received by Licensor and
++ subsequently incorporated within the Work.
++
++ 2. Grant of Copyright License. Subject to the terms and conditions of
++ this License, each Contributor hereby grants to You a perpetual,
++ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
++ copyright license to reproduce, prepare Derivative Works of,
++ publicly display, publicly perform, sublicense, and distribute the
++ Work and such Derivative Works in Source or Object form.
++
++ 3. Grant of Patent License. Subject to the terms and conditions of
++ this License, each Contributor hereby grants to You a perpetual,
++ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
++ (except as stated in this section) patent license to make, have made,
++ use, offer to sell, sell, import, and otherwise transfer the Work,
++ where such license applies only to those patent claims licensable
++ by such Contributor that are necessarily infringed by their
++ Contribution(s) alone or by combination of their Contribution(s)
++ with the Work to which such Contribution(s) was submitted. If You
++ institute patent litigation against any entity (including a
++ cross-claim or counterclaim in a lawsuit) alleging that the Work
++ or a Contribution incorporated within the Work constitutes direct
++ or contributory patent infringement, then any patent licenses
++ granted to You under this License for that Work shall terminate
++ as of the date such litigation is filed.
++
++ 4. Redistribution. You may reproduce and distribute copies of the
++ Work or Derivative Works thereof in any medium, with or without
++ modifications, and in Source or Object form, provided that You
++ meet the following conditions:
++
++ (a) You must give any other recipients of the Work or
++ Derivative Works a copy of this License; and
++
++ (b) You must cause any modified files to carry prominent notices
++ stating that You changed the files; and
++
++ (c) You must retain, in the Source form of any Derivative Works
++ that You distribute, all copyright, patent, trademark, and
++ attribution notices from the Source form of the Work,
++ excluding those notices that do not pertain to any part of
++ the Derivative Works; and
++
++ (d) If the Work includes a "NOTICE" text file as part of its
++ distribution, then any Derivative Works that You distribute must
++ include a readable copy of the attribution notices contained
++ within such NOTICE file, excluding those notices that do not
++ pertain to any part of the Derivative Works, in at least one
++ of the following places: within a NOTICE text file distributed
++ as part of the Derivative Works; within the Source form or
++ documentation, if provided along with the Derivative Works; or,
++ within a display generated by the Derivative Works, if and
++ wherever such third-party notices normally appear. The contents
++ of the NOTICE file are for informational purposes only and
++ do not modify the License. You may add Your own attribution
++ notices within Derivative Works that You distribute, alongside
++ or as an addendum to the NOTICE text from the Work, provided
++ that such additional attribution notices cannot be construed
++ as modifying the License.
++
++ You may add Your own copyright statement to Your modifications and
++ may provide additional or different license terms and conditions
++ for use, reproduction, or distribution of Your modifications, or
++ for any such Derivative Works as a whole, provided Your use,
++ reproduction, and distribution of the Work otherwise complies with
++ the conditions stated in this License.
++
++ 5. Submission of Contributions. Unless You explicitly state otherwise,
++ any Contribution intentionally submitted for inclusion in the Work
++ by You to the Licensor shall be under the terms and conditions of
++ this License, without any additional terms or conditions.
++ Notwithstanding the above, nothing herein shall supersede or modify
++ the terms of any separate license agreement you may have executed
++ with Licensor regarding such Contributions.
++
++ 6. Trademarks. This License does not grant permission to use the trade
++ names, trademarks, service marks, or product names of the Licensor,
++ except as required for reasonable and customary use in describing the
++ origin of the Work and reproducing the content of the NOTICE file.
++
++ 7. Disclaimer of Warranty. Unless required by applicable law or
++ agreed to in writing, Licensor provides the Work (and each
++ Contributor provides its Contributions) on an "AS IS" BASIS,
++ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
++ implied, including, without limitation, any warranties or conditions
++ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
++ PARTICULAR PURPOSE. You are solely responsible for determining the
++ appropriateness of using or redistributing the Work and assume any
++ risks associated with Your exercise of permissions under this License.
++
++ 8. Limitation of Liability. In no event and under no legal theory,
++ whether in tort (including negligence), contract, or otherwise,
++ unless required by applicable law (such as deliberate and grossly
++ negligent acts) or agreed to in writing, shall any Contributor be
++ liable to You for damages, including any direct, indirect, special,
++ incidental, or consequential damages of any character arising as a
++ result of this License or out of the use or inability to use the
++ Work (including but not limited to damages for loss of goodwill,
++ work stoppage, computer failure or malfunction, or any and all
++ other commercial damages or losses), even if such Contributor
++ has been advised of the possibility of such damages.
++
++ 9. Accepting Warranty or Additional Liability. While redistributing
++ the Work or Derivative Works thereof, You may choose to offer,
++ and charge a fee for, acceptance of support, warranty, indemnity,
++ or other liability obligations and/or rights consistent with this
++ License. However, in accepting such obligations, You may act only
++ on Your own behalf and on Your sole responsibility, not on behalf
++ of any other Contributor, and only if You agree to indemnify,
++ defend, and hold each Contributor harmless for any liability
++ incurred by, or claims asserted against, such Contributor by reason
++ of your accepting any such warranty or additional liability.
++
++ END OF TERMS AND CONDITIONS
++
++ APPENDIX: How to apply the Apache License to your work.
++
++ To apply the Apache License to your work, attach the following
++ boilerplate notice, with the fields enclosed by brackets "{}"
++ replaced with your own identifying information. (Don't include
++ the brackets!) The text should be enclosed in the appropriate
++ comment syntax for the file format. We also recommend that a
++ file or class name and description of purpose be included on the
++ same "printed page" as the copyright notice for easier
++ identification within third-party archives.
++
++ Copyright {yyyy} {name of copyright owner}
++
++ Licensed under the Apache License, Version 2.0 (the "License");
++ you may not use this file except in compliance with the License.
++ You may obtain a copy of the License at
++
++ http://www.apache.org/licenses/LICENSE-2.0
++
++ Unless required by applicable law or agreed to in writing, software
++ distributed under the License is distributed on an "AS IS" BASIS,
++ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
++ See the License for the specific language governing permissions and
++ limitations under the License.
++
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2881ac9c/external/storm-hdfs/README.md
----------------------------------------------------------------------
diff --cc external/storm-hdfs/README.md
index 0000000,0000000..1bf1d11
new file mode 100644
--- /dev/null
+++ b/external/storm-hdfs/README.md
@@@ -1,0 -1,0 +1,338 @@@
++# Storm HDFS
++
++Storm components for interacting with HDFS file systems
++
++
++## Usage
++The following example will write pipe("|")-delimited files to the HDFS path hdfs://localhost:54310/foo. After every
++1,000 tuples it will sync filesystem, making that data visible to other HDFS clients. It will rotate files when they
++reach 5 megabytes in size.
++
++```java
++// use "|" instead of "," for field delimiter
++RecordFormat format = new DelimitedRecordFormat()
++ .withFieldDelimiter("|");
++
++// sync the filesystem after every 1k tuples
++SyncPolicy syncPolicy = new CountSyncPolicy(1000);
++
++// rotate files when they reach 5MB
++FileRotationPolicy rotationPolicy = new FileSizeRotationPolicy(5.0f, Units.MB);
++
++FileNameFormat fileNameFormat = new DefaultFileNameFormat()
++ .withPath("/foo/");
++
++HdfsBolt bolt = new HdfsBolt()
++ .withFsUrl("hdfs://localhost:54310")
++ .withFileNameFormat(fileNameFormat)
++ .withRecordFormat(format)
++ .withRotationPolicy(rotationPolicy)
++ .withSyncPolicy(syncPolicy);
++```
++
++### Packaging a Topology
++When packaging your topology, it's important that you use the [maven-shade-plugin]() as opposed to the
++[maven-assempbly-plugin]().
++
++The shade plugin provides facilities for merging JAR manifest entries, which the hadoop client leverages for URL scheme
++resolution.
++
++If you experience errors such as the following:
++
++```
++java.lang.RuntimeException: Error preparing HdfsBolt: No FileSystem for scheme: hdfs
++```
++
++it's an indication that your topology jar file isn't packaged properly.
++
++If you are using maven to create your topology jar, you should use the following `maven-shade-plugin` configuration to
++create your topology jar:
++
++```xml
++<plugin>
++ <groupId>org.apache.maven.plugins</groupId>
++ <artifactId>maven-shade-plugin</artifactId>
++ <version>1.4</version>
++ <configuration>
++ <createDependencyReducedPom>true</createDependencyReducedPom>
++ </configuration>
++ <executions>
++ <execution>
++ <phase>package</phase>
++ <goals>
++ <goal>shade</goal>
++ </goals>
++ <configuration>
++ <transformers>
++ <transformer
++ implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
++ <transformer
++ implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
++ <mainClass></mainClass>
++ </transformer>
++ </transformers>
++ </configuration>
++ </execution>
++ </executions>
++</plugin>
++
++```
++
++### Specifying a Hadoop Version
++By default, storm-hdfs uses the following Hadoop dependencies:
++
++```xml
++<dependency>
++ <groupId>org.apache.hadoop</groupId>
++ <artifactId>hadoop-client</artifactId>
++ <version>2.2.0</version>
++ <exclusions>
++ <exclusion>
++ <groupId>org.slf4j</groupId>
++ <artifactId>slf4j-log4j12</artifactId>
++ </exclusion>
++ </exclusions>
++</dependency>
++<dependency>
++ <groupId>org.apache.hadoop</groupId>
++ <artifactId>hadoop-hdfs</artifactId>
++ <version>2.2.0</version>
++ <exclusions>
++ <exclusion>
++ <groupId>org.slf4j</groupId>
++ <artifactId>slf4j-log4j12</artifactId>
++ </exclusion>
++ </exclusions>
++</dependency>
++```
++
++If you are using a different version of Hadoop, you should exclude the Hadoop libraries from the storm-hdfs dependency
++and add the dependencies for your preferred version in your pom.
++
++Hadoop client version incompatibilites can manifest as errors like:
++
++```
++com.google.protobuf.InvalidProtocolBufferException: Protocol message contained an invalid tag (zero)
++```
++
++## Customization
++
++### Record Formats
++Record format can be controlled by providing an implementation of the `org.apache.storm.hdfs.format.RecordFormat`
++interface:
++
++```java
++public interface RecordFormat extends Serializable {
++ byte[] format(Tuple tuple);
++}
++```
++
++The provided `org.apache.storm.hdfs.format.DelimitedRecordFormat` is capable of producing formats such as CSV and
++tab-delimited files.
++
++
++### File Naming
++File naming can be controlled by providing an implementation of the `org.apache.storm.hdfs.format.FileNameFormat`
++interface:
++
++```java
++public interface FileNameFormat extends Serializable {
++ void prepare(Map conf, TopologyContext topologyContext);
++ String getName(long rotation, long timeStamp);
++ String getPath();
++}
++```
++
++The provided `org.apache.storm.hdfs.format.DefaultFileNameFormat` will create file names with the following format:
++
++ {prefix}{componentId}-{taskId}-{rotationNum}-{timestamp}{extension}
++
++For example:
++
++ MyBolt-5-7-1390579837830.txt
++
++By default, prefix is empty and extenstion is ".txt".
++
++
++
++### Sync Policies
++Sync policies allow you to control when buffered data is flushed to the underlying filesystem (thus making it available
++to clients reading the data) by implementing the `org.apache.storm.hdfs.sync.SyncPolicy` interface:
++
++```java
++public interface SyncPolicy extends Serializable {
++ boolean mark(Tuple tuple, long offset);
++ void reset();
++}
++```
++The `HdfsBolt` will call the `mark()` method for every tuple it processes. Returning `true` will trigger the `HdfsBolt`
++to perform a sync/flush, after which it will call the `reset()` method.
++
++The `org.apache.storm.hdfs.sync.CountSyncPolicy` class simply triggers a sync after the specified number of tuples have
++been processed.
++
++### File Rotation Policies
++Similar to sync policies, file rotation policies allow you to control when data files are rotated by providing a
++`org.apache.storm.hdfs.rotation.FileRotation` interface:
++
++```java
++public interface FileRotationPolicy extends Serializable {
++ boolean mark(Tuple tuple, long offset);
++ void reset();
++}
++```
++
++The `org.apache.storm.hdfs.rotation.FileSizeRotationPolicy` implementation allows you to trigger file rotation when
++data files reach a specific file size:
++
++```java
++FileRotationPolicy rotationPolicy = new FileSizeRotationPolicy(5.0f, Units.MB);
++```
++
++### File Rotation Actions
++Both the HDFS bolt and Trident State implementation allow you to register any number of `RotationAction`s.
++What `RotationAction`s do is provide a hook to allow you to perform some action right after a file is rotated. For
++example, moving a file to a different location or renaming it.
++
++
++```java
++public interface RotationAction extends Serializable {
++ void execute(FileSystem fileSystem, Path filePath) throws IOException;
++}
++```
++
++Storm-HDFS includes a simple action that will move a file after rotation:
++
++```java
++public class MoveFileAction implements RotationAction {
++ private static final Logger LOG = LoggerFactory.getLogger(MoveFileAction.class);
++
++ private String destination;
++
++ public MoveFileAction withDestination(String destDir){
++ destination = destDir;
++ return this;
++ }
++
++ @Override
++ public void execute(FileSystem fileSystem, Path filePath) throws IOException {
++ Path destPath = new Path(destination, filePath.getName());
++ LOG.info("Moving file {} to {}", filePath, destPath);
++ boolean success = fileSystem.rename(filePath, destPath);
++ return;
++ }
++}
++```
++
++If you are using Trident and sequence files you can do something like this:
++
++```java
++ HdfsState.Options seqOpts = new HdfsState.SequenceFileOptions()
++ .withFileNameFormat(fileNameFormat)
++ .withSequenceFormat(new DefaultSequenceFormat("key", "data"))
++ .withRotationPolicy(rotationPolicy)
++ .withFsUrl("hdfs://localhost:54310")
++ .addRotationAction(new MoveFileAction().withDestination("/dest2/"));
++```
++
++
++## Support for HDFS Sequence Files
++
++The `org.apache.storm.hdfs.bolt.SequenceFileBolt` class allows you to write storm data to HDFS sequence files:
++
++```java
++ // sync the filesystem after every 1k tuples
++ SyncPolicy syncPolicy = new CountSyncPolicy(1000);
++
++ // rotate files when they reach 5MB
++ FileRotationPolicy rotationPolicy = new FileSizeRotationPolicy(5.0f, Units.MB);
++
++ FileNameFormat fileNameFormat = new DefaultFileNameFormat()
++ .withExtension(".seq")
++ .withPath("/data/");
++
++ // create sequence format instance.
++ DefaultSequenceFormat format = new DefaultSequenceFormat("timestamp", "sentence");
++
++ SequenceFileBolt bolt = new SequenceFileBolt()
++ .withFsUrl("hdfs://localhost:54310")
++ .withFileNameFormat(fileNameFormat)
++ .withSequenceFormat(format)
++ .withRotationPolicy(rotationPolicy)
++ .withSyncPolicy(syncPolicy)
++ .withCompressionType(SequenceFile.CompressionType.RECORD)
++ .withCompressionCodec("deflate");
++```
++
++The `SequenceFileBolt` requires that you provide a `org.apache.storm.hdfs.bolt.format.SequenceFormat` that maps tuples to
++key/value pairs:
++
++```java
++public interface SequenceFormat extends Serializable {
++ Class keyClass();
++ Class valueClass();
++
++ Writable key(Tuple tuple);
++ Writable value(Tuple tuple);
++}
++```
++
++## Trident API
++storm-hdfs also includes a Trident `state` implementation for writing data to HDFS, with an API that closely mirrors
++that of the bolts.
++
++ ```java
++ Fields hdfsFields = new Fields("field1", "field2");
++
++ FileNameFormat fileNameFormat = new DefaultFileNameFormat()
++ .withPath("/trident")
++ .withPrefix("trident")
++ .withExtension(".txt");
++
++ RecordFormat recordFormat = new DelimitedRecordFormat()
++ .withFields(hdfsFields);
++
++ FileRotationPolicy rotationPolicy = new FileSizeRotationPolicy(5.0f, FileSizeRotationPolicy.Units.MB);
++
++ HdfsState.Options options = new HdfsState.HdfsFileOptions()
++ .withFileNameFormat(fileNameFormat)
++ .withRecordFormat(recordFormat)
++ .withRotationPolicy(rotationPolicy)
++ .withFsUrl("hdfs://localhost:54310");
++
++ StateFactory factory = new HdfsStateFactory().withOptions(options);
++
++ TridentState state = stream
++ .partitionPersist(factory, hdfsFields, new HdfsUpdater(), new Fields());
++ ```
++
++ To use the sequence file `State` implementation, use the `HdfsState.SequenceFileOptions`:
++
++ ```java
++ HdfsState.Options seqOpts = new HdfsState.SequenceFileOptions()
++ .withFileNameFormat(fileNameFormat)
++ .withSequenceFormat(new DefaultSequenceFormat("key", "data"))
++ .withRotationPolicy(rotationPolicy)
++ .withFsUrl("hdfs://localhost:54310")
++ .addRotationAction(new MoveFileAction().toDestination("/dest2/"));
++```
++
++
++## License
++
++Licensed to the Apache Software Foundation (ASF) under one
++or more contributor license agreements. See the NOTICE file
++distributed with this work for additional information
++regarding copyright ownership. The ASF licenses this file
++to you under the Apache License, Version 2.0 (the
++"License"); you may not use this file except in compliance
++with the License. You may obtain a copy of the License at
++
++ http://www.apache.org/licenses/LICENSE-2.0
++
++Unless required by applicable law or agreed to in writing,
++software distributed under the License is distributed on an
++"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
++KIND, either express or implied. See the License for the
++specific language governing permissions and limitations
++under the License.
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2881ac9c/external/storm-hdfs/pom.xml
----------------------------------------------------------------------
diff --cc external/storm-hdfs/pom.xml
index 0000000,0000000..e4c62b3
new file mode 100644
--- /dev/null
+++ b/external/storm-hdfs/pom.xml
@@@ -1,0 -1,0 +1,81 @@@
++<?xml version="1.0" encoding="UTF-8"?>
++<!--
++ Licensed to the Apache Software Foundation (ASF) under one or more
++ contributor license agreements. See the NOTICE file distributed with
++ this work for additional information regarding copyright ownership.
++ The ASF licenses this file to You under the Apache License, Version 2.0
++ (the "License"); you may not use this file except in compliance with
++ the License. You may obtain a copy of the License at
++
++ http://www.apache.org/licenses/LICENSE-2.0
++
++ Unless required by applicable law or agreed to in writing, software
++ distributed under the License is distributed on an "AS IS" BASIS,
++ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
++ See the License for the specific language governing permissions and
++ limitations under the License.
++-->
++<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
++ <modelVersion>4.0.0</modelVersion>
++
++ <parent>
++ <groupId>org.sonatype.oss</groupId>
++ <artifactId>oss-parent</artifactId>
++ <version>7</version>
++ </parent>
++
++ <groupId>com.github.ptgoetz</groupId>
++ <artifactId>storm-hdfs</artifactId>
++ <version>0.1.3-SNAPSHOT</version>
++
++ <licenses>
++ <license>
++ <name>The Apache Software License, Version 2.0</name>
++ <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
++ </license>
++ </licenses>
++ <scm>
++ <connection>scm:git:git@github.com:ptgoetz/storm-hdfs.git</connection>
++ <developerConnection>scm:git:git@github.com:ptgoetz/storm-hdfs.git</developerConnection>
++ <url>:git@github.com:ptgoetz/storm-hdfs.git</url>
++ </scm>
++
++ <developers>
++ <developer>
++ <id>ptgoetz</id>
++ <name>P. Taylor Goetz</name>
++ <email>ptgoetz@gmail.com</email>
++ </developer>
++ </developers>
++
++ <dependencies>
++ <dependency>
++ <groupId>org.apache.storm</groupId>
++ <artifactId>storm-core</artifactId>
++ <version>0.9.1-incubating</version>
++ <scope>provided</scope>
++ </dependency>
++ <dependency>
++ <groupId>org.apache.hadoop</groupId>
++ <artifactId>hadoop-client</artifactId>
++ <version>2.2.0</version>
++ <exclusions>
++ <exclusion>
++ <groupId>org.slf4j</groupId>
++ <artifactId>slf4j-log4j12</artifactId>
++ </exclusion>
++ </exclusions>
++ </dependency>
++ <dependency>
++ <groupId>org.apache.hadoop</groupId>
++ <artifactId>hadoop-hdfs</artifactId>
++ <version>2.2.0</version>
++ <exclusions>
++ <exclusion>
++ <groupId>org.slf4j</groupId>
++ <artifactId>slf4j-log4j12</artifactId>
++ </exclusion>
++ </exclusions>
++ </dependency>
++ </dependencies>
++</project>
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2881ac9c/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/AbstractHdfsBolt.java
----------------------------------------------------------------------
diff --cc external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/AbstractHdfsBolt.java
index 0000000,0000000..3d3a3c1
new file mode 100644
--- /dev/null
+++ b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/AbstractHdfsBolt.java
@@@ -1,0 -1,0 +1,110 @@@
++/**
++ * Licensed to the Apache Software Foundation (ASF) under one
++ * or more contributor license agreements. See the NOTICE file
++ * distributed with this work for additional information
++ * regarding copyright ownership. The ASF licenses this file
++ * to you under the Apache License, Version 2.0 (the
++ * "License"); you may not use this file except in compliance
++ * with the License. You may obtain a copy of the License at
++ *
++ * http://www.apache.org/licenses/LICENSE-2.0
++ *
++ * Unless required by applicable law or agreed to in writing, software
++ * distributed under the License is distributed on an "AS IS" BASIS,
++ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
++ * See the License for the specific language governing permissions and
++ * limitations under the License.
++ */
++package org.apache.storm.hdfs.bolt;
++
++import backtype.storm.task.OutputCollector;
++import backtype.storm.task.TopologyContext;
++import backtype.storm.topology.OutputFieldsDeclarer;
++import backtype.storm.topology.base.BaseRichBolt;
++import org.apache.hadoop.conf.Configuration;
++import org.apache.hadoop.fs.FileSystem;
++import org.apache.hadoop.fs.Path;
++import org.apache.storm.hdfs.bolt.format.FileNameFormat;
++import org.apache.storm.hdfs.bolt.rotation.FileRotationPolicy;
++import org.apache.storm.hdfs.bolt.sync.SyncPolicy;
++import org.apache.storm.hdfs.common.rotation.RotationAction;
++import org.apache.storm.hdfs.common.security.HdfsSecurityUtil;
++import org.slf4j.Logger;
++import org.slf4j.LoggerFactory;
++
++import java.io.IOException;
++import java.util.ArrayList;
++import java.util.Map;
++
++public abstract class AbstractHdfsBolt extends BaseRichBolt {
++ private static final Logger LOG = LoggerFactory.getLogger(AbstractHdfsBolt.class);
++
++ protected ArrayList<RotationAction> rotationActions = new ArrayList<RotationAction>();
++ private Path currentFile;
++ protected OutputCollector collector;
++ protected FileSystem fs;
++ protected SyncPolicy syncPolicy;
++ protected FileRotationPolicy rotationPolicy;
++ protected FileNameFormat fileNameFormat;
++ protected int rotation = 0;
++ protected String fsUrl;
++ protected String configKey;
++// protected String path;
++
++ protected Configuration hdfsConfig;
++
++ protected void rotateOutputFile() throws IOException {
++ LOG.info("Rotating output file...");
++ long start = System.currentTimeMillis();
++ closeOutputFile();
++ this.rotation++;
++
++ Path newFile = createOutputFile();
++ LOG.info("Performing {} file rotation actions.", this.rotationActions.size());
++ for(RotationAction action : this.rotationActions){
++ action.execute(this.fs, this.currentFile);
++ }
++ this.currentFile = newFile;
++ long time = System.currentTimeMillis() - start;
++ LOG.info("File rotation took {} ms.", time);
++ }
++
++ public void prepare(Map conf, TopologyContext topologyContext, OutputCollector collector){
++ if (this.syncPolicy == null) throw new IllegalStateException("SyncPolicy must be specified.");
++ if (this.rotationPolicy == null) throw new IllegalStateException("RotationPolicy must be specified.");
++ if (this.fsUrl == null) {
++ throw new IllegalStateException("File system URL must be specified.");
++ }
++
++ this.collector = collector;
++ this.fileNameFormat.prepare(conf, topologyContext);
++ this.hdfsConfig = new Configuration();
++ Map<String, Object> map = (Map<String, Object>)conf.get(this.configKey);
++ if(map != null){
++ for(String key : map.keySet()){
++ this.hdfsConfig.set(key, String.valueOf(map.get(key)));
++ }
++ }
++
++
++ try{
++ HdfsSecurityUtil.login(conf, hdfsConfig);
++ doPrepare(conf, topologyContext, collector);
++ this.currentFile = createOutputFile();
++
++ } catch (Exception e){
++ throw new RuntimeException("Error preparing HdfsBolt: " + e.getMessage(), e);
++ }
++ }
++
++ @Override
++ public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
++ }
++
++ abstract void closeOutputFile() throws IOException;
++
++ abstract Path createOutputFile() throws IOException;
++
++ abstract void doPrepare(Map conf, TopologyContext topologyContext, OutputCollector collector) throws IOException;
++
++}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2881ac9c/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/HdfsBolt.java
----------------------------------------------------------------------
diff --cc external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/HdfsBolt.java
index 0000000,0000000..489b0b0
new file mode 100644
--- /dev/null
+++ b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/HdfsBolt.java
@@@ -1,0 -1,0 +1,129 @@@
++/**
++ * Licensed to the Apache Software Foundation (ASF) under one
++ * or more contributor license agreements. See the NOTICE file
++ * distributed with this work for additional information
++ * regarding copyright ownership. The ASF licenses this file
++ * to you under the Apache License, Version 2.0 (the
++ * "License"); you may not use this file except in compliance
++ * with the License. You may obtain a copy of the License at
++ *
++ * http://www.apache.org/licenses/LICENSE-2.0
++ *
++ * Unless required by applicable law or agreed to in writing, software
++ * distributed under the License is distributed on an "AS IS" BASIS,
++ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
++ * See the License for the specific language governing permissions and
++ * limitations under the License.
++ */
++package org.apache.storm.hdfs.bolt;
++
++import backtype.storm.task.OutputCollector;
++import backtype.storm.task.TopologyContext;
++import backtype.storm.tuple.Tuple;
++import org.apache.hadoop.fs.FSDataOutputStream;
++import org.apache.hadoop.fs.FileSystem;
++import org.apache.hadoop.fs.Path;
++import org.apache.hadoop.hdfs.client.HdfsDataOutputStream;
++import org.apache.hadoop.hdfs.client.HdfsDataOutputStream.SyncFlag;
++import org.apache.storm.hdfs.bolt.format.FileNameFormat;
++import org.apache.storm.hdfs.bolt.format.RecordFormat;
++import org.apache.storm.hdfs.bolt.rotation.FileRotationPolicy;
++import org.apache.storm.hdfs.bolt.sync.SyncPolicy;
++import org.apache.storm.hdfs.common.rotation.RotationAction;
++import org.slf4j.Logger;
++import org.slf4j.LoggerFactory;
++
++import java.io.IOException;
++import java.net.URI;
++import java.util.EnumSet;
++import java.util.Map;
++
++public class HdfsBolt extends AbstractHdfsBolt{
++ private static final Logger LOG = LoggerFactory.getLogger(HdfsBolt.class);
++
++ private FSDataOutputStream out;
++ private RecordFormat format;
++ private long offset = 0;
++
++ public HdfsBolt withFsUrl(String fsUrl){
++ this.fsUrl = fsUrl;
++ return this;
++ }
++
++ public HdfsBolt withConfigKey(String configKey){
++ this.configKey = configKey;
++ return this;
++ }
++
++ public HdfsBolt withFileNameFormat(FileNameFormat fileNameFormat){
++ this.fileNameFormat = fileNameFormat;
++ return this;
++ }
++
++ public HdfsBolt withRecordFormat(RecordFormat format){
++ this.format = format;
++ return this;
++ }
++
++ public HdfsBolt withSyncPolicy(SyncPolicy syncPolicy){
++ this.syncPolicy = syncPolicy;
++ return this;
++ }
++
++ public HdfsBolt withRotationPolicy(FileRotationPolicy rotationPolicy){
++ this.rotationPolicy = rotationPolicy;
++ return this;
++ }
++
++ public HdfsBolt addRotationAction(RotationAction action){
++ this.rotationActions.add(action);
++ return this;
++ }
++
++ @Override
++ public void doPrepare(Map conf, TopologyContext topologyContext, OutputCollector collector) throws IOException {
++ LOG.info("Preparing HDFS Bolt...");
++ this.fs = FileSystem.get(URI.create(this.fsUrl), hdfsConfig);
++ }
++
++ @Override
++ public void execute(Tuple tuple) {
++ try {
++ byte[] bytes = this.format.format(tuple);
++ out.write(bytes);
++ this.offset += bytes.length;
++
++ this.collector.ack(tuple);
++
++ if(this.syncPolicy.mark(tuple, this.offset)){
++ if(this.out instanceof HdfsDataOutputStream){
++ ((HdfsDataOutputStream)this.out).hsync(EnumSet.of(SyncFlag.UPDATE_LENGTH));
++ } else {
++ this.out.hsync();
++ }
++ this.syncPolicy.reset();
++ }
++ if(this.rotationPolicy.mark(tuple, this.offset)){
++ rotateOutputFile();
++ this.offset = 0;
++ this.rotationPolicy.reset();
++ }
++ } catch (IOException e) {
++ LOG.warn("write/sync failed.", e);
++ this.collector.fail(tuple);
++ }
++ }
++
++ @Override
++ void closeOutputFile() throws IOException {
++ this.out.hsync();
++ this.out.close();
++ }
++
++ @Override
++ Path createOutputFile() throws IOException {
++ Path path = new Path(this.fileNameFormat.getPath(), this.fileNameFormat.getName(this.rotation, System.currentTimeMillis()));
++ this.out = this.fs.create(path);
++ return path;
++ }
++}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2881ac9c/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/SequenceFileBolt.java
----------------------------------------------------------------------
diff --cc external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/SequenceFileBolt.java
index 0000000,0000000..4f519de
new file mode 100644
--- /dev/null
+++ b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/SequenceFileBolt.java
@@@ -1,0 -1,0 +1,146 @@@
++/**
++ * Licensed to the Apache Software Foundation (ASF) under one
++ * or more contributor license agreements. See the NOTICE file
++ * distributed with this work for additional information
++ * regarding copyright ownership. The ASF licenses this file
++ * to you under the Apache License, Version 2.0 (the
++ * "License"); you may not use this file except in compliance
++ * with the License. You may obtain a copy of the License at
++ *
++ * http://www.apache.org/licenses/LICENSE-2.0
++ *
++ * Unless required by applicable law or agreed to in writing, software
++ * distributed under the License is distributed on an "AS IS" BASIS,
++ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
++ * See the License for the specific language governing permissions and
++ * limitations under the License.
++ */
++package org.apache.storm.hdfs.bolt;
++
++import backtype.storm.task.OutputCollector;
++import backtype.storm.task.TopologyContext;
++import backtype.storm.tuple.Tuple;
++import org.apache.hadoop.fs.FileSystem;
++import org.apache.hadoop.fs.Path;
++import org.apache.hadoop.io.SequenceFile;
++import org.apache.hadoop.io.compress.CompressionCodecFactory;
++import org.apache.storm.hdfs.bolt.format.FileNameFormat;
++import org.apache.storm.hdfs.bolt.format.SequenceFormat;
++import org.apache.storm.hdfs.bolt.rotation.FileRotationPolicy;
++import org.apache.storm.hdfs.bolt.sync.SyncPolicy;
++import org.apache.storm.hdfs.common.rotation.RotationAction;
++import org.slf4j.Logger;
++import org.slf4j.LoggerFactory;
++
++import java.io.IOException;
++import java.net.URI;
++import java.util.Map;
++
++public class SequenceFileBolt extends AbstractHdfsBolt {
++ private static final Logger LOG = LoggerFactory.getLogger(SequenceFileBolt.class);
++
++ private SequenceFormat format;
++ private SequenceFile.CompressionType compressionType = SequenceFile.CompressionType.RECORD;
++ private SequenceFile.Writer writer;
++
++ private String compressionCodec = "default";
++ private transient CompressionCodecFactory codecFactory;
++
++ public SequenceFileBolt() {
++ }
++
++ public SequenceFileBolt withCompressionCodec(String codec){
++ this.compressionCodec = codec;
++ return this;
++ }
++
++ public SequenceFileBolt withFsUrl(String fsUrl) {
++ this.fsUrl = fsUrl;
++ return this;
++ }
++
++ public SequenceFileBolt withConfigKey(String configKey){
++ this.configKey = configKey;
++ return this;
++ }
++
++ public SequenceFileBolt withFileNameFormat(FileNameFormat fileNameFormat) {
++ this.fileNameFormat = fileNameFormat;
++ return this;
++ }
++
++ public SequenceFileBolt withSequenceFormat(SequenceFormat format) {
++ this.format = format;
++ return this;
++ }
++
++ public SequenceFileBolt withSyncPolicy(SyncPolicy syncPolicy) {
++ this.syncPolicy = syncPolicy;
++ return this;
++ }
++
++ public SequenceFileBolt withRotationPolicy(FileRotationPolicy rotationPolicy) {
++ this.rotationPolicy = rotationPolicy;
++ return this;
++ }
++
++ public SequenceFileBolt withCompressionType(SequenceFile.CompressionType compressionType){
++ this.compressionType = compressionType;
++ return this;
++ }
++
++ public SequenceFileBolt addRotationAction(RotationAction action){
++ this.rotationActions.add(action);
++ return this;
++ }
++
++ @Override
++ public void doPrepare(Map conf, TopologyContext topologyContext, OutputCollector collector) throws IOException {
++ LOG.info("Preparing Sequence File Bolt...");
++ if (this.format == null) throw new IllegalStateException("SequenceFormat must be specified.");
++
++ this.fs = FileSystem.get(URI.create(this.fsUrl), hdfsConfig);
++ this.codecFactory = new CompressionCodecFactory(hdfsConfig);
++ }
++
++ @Override
++ public void execute(Tuple tuple) {
++ try {
++ this.writer.append(this.format.key(tuple), this.format.value(tuple));
++ long offset = this.writer.getLength();
++ this.collector.ack(tuple);
++
++ if (this.syncPolicy.mark(tuple, offset)) {
++ this.writer.hsync();
++ this.syncPolicy.reset();
++ }
++ if (this.rotationPolicy.mark(tuple, offset)) {
++ rotateOutputFile();
++ this.rotationPolicy.reset();
++ }
++ } catch (IOException e) {
++ LOG.warn("write/sync failed.", e);
++ this.collector.fail(tuple);
++ }
++
++ }
++
++ Path createOutputFile() throws IOException {
++ Path p = new Path(this.fsUrl + this.fileNameFormat.getPath(), this.fileNameFormat.getName(this.rotation, System.currentTimeMillis()));
++ this.writer = SequenceFile.createWriter(
++ this.hdfsConfig,
++ SequenceFile.Writer.file(p),
++ SequenceFile.Writer.keyClass(this.format.keyClass()),
++ SequenceFile.Writer.valueClass(this.format.valueClass()),
++ SequenceFile.Writer.compression(this.compressionType, this.codecFactory.getCodecByName(this.compressionCodec))
++ );
++ return p;
++ }
++
++ void closeOutputFile() throws IOException {
++ this.writer.hsync();
++ this.writer.close();
++ }
++
++
++}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2881ac9c/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/format/DefaultFileNameFormat.java
----------------------------------------------------------------------
diff --cc external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/format/DefaultFileNameFormat.java
index 0000000,0000000..e046f90
new file mode 100644
--- /dev/null
+++ b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/format/DefaultFileNameFormat.java
@@@ -1,0 -1,0 +1,86 @@@
++/**
++ * Licensed to the Apache Software Foundation (ASF) under one
++ * or more contributor license agreements. See the NOTICE file
++ * distributed with this work for additional information
++ * regarding copyright ownership. The ASF licenses this file
++ * to you under the Apache License, Version 2.0 (the
++ * "License"); you may not use this file except in compliance
++ * with the License. You may obtain a copy of the License at
++ *
++ * http://www.apache.org/licenses/LICENSE-2.0
++ *
++ * Unless required by applicable law or agreed to in writing, software
++ * distributed under the License is distributed on an "AS IS" BASIS,
++ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
++ * See the License for the specific language governing permissions and
++ * limitations under the License.
++ */
++package org.apache.storm.hdfs.bolt.format;
++
++import backtype.storm.task.TopologyContext;
++
++import java.util.Map;
++
++
++/**
++ * Creates file names with the following format:
++ * <pre>
++ * {prefix}{componentId}-{taskId}-{rotationNum}-{timestamp}{extension}
++ * </pre>
++ * For example:
++ * <pre>
++ * MyBolt-5-7-1390579837830.txt
++ * </pre>
++ *
++ * By default, prefix is empty and extenstion is ".txt".
++ *
++ */
++public class DefaultFileNameFormat implements FileNameFormat {
++ private String componentId;
++ private int taskId;
++ private String path = "/storm";
++ private String prefix = "";
++ private String extension = ".txt";
++
++ /**
++ * Overrides the default prefix.
++ *
++ * @param prefix
++ * @return
++ */
++ public DefaultFileNameFormat withPrefix(String prefix){
++ this.prefix = prefix;
++ return this;
++ }
++
++ /**
++ * Overrides the default file extension.
++ *
++ * @param extension
++ * @return
++ */
++ public DefaultFileNameFormat withExtension(String extension){
++ this.extension = extension;
++ return this;
++ }
++
++ public DefaultFileNameFormat withPath(String path){
++ this.path = path;
++ return this;
++ }
++
++ @Override
++ public void prepare(Map conf, TopologyContext topologyContext) {
++ this.componentId = topologyContext.getThisComponentId();
++ this.taskId = topologyContext.getThisTaskId();
++ }
++
++ @Override
++ public String getName(long rotation, long timeStamp) {
++ return this.prefix + this.componentId + "-" + this.taskId + "-" + rotation + "-" + timeStamp + this.extension;
++ }
++
++ public String getPath(){
++ return this.path;
++ }
++}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2881ac9c/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/format/DefaultSequenceFormat.java
----------------------------------------------------------------------
diff --cc external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/format/DefaultSequenceFormat.java
index 0000000,0000000..4bd0f26
new file mode 100644
--- /dev/null
+++ b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/format/DefaultSequenceFormat.java
@@@ -1,0 -1,0 +1,52 @@@
++package org.apache.storm.hdfs.bolt.format;
++
++import backtype.storm.tuple.Tuple;
++import org.apache.hadoop.io.LongWritable;
++import org.apache.hadoop.io.Text;
++import org.apache.hadoop.io.Writable;
++
++/**
++ * Basic <code>SequenceFormat</code> implementation that uses
++ * <code>LongWritable</code> for keys and <code>Text</code> for values.
++ *
++ */
++public class DefaultSequenceFormat implements SequenceFormat {
++ private transient LongWritable key;
++ private transient Text value;
++
++ private String keyField;
++ private String valueField;
++
++ public DefaultSequenceFormat(String keyField, String valueField){
++ this.keyField = keyField;
++ this.valueField = valueField;
++ }
++
++ @Override
++ public Class keyClass() {
++ return LongWritable.class;
++ }
++
++ @Override
++ public Class valueClass() {
++ return Text.class;
++ }
++
++ @Override
++ public Writable key(Tuple tuple) {
++ if(this.key == null){
++ this.key = new LongWritable();
++ }
++ this.key.set(tuple.getLongByField(this.keyField));
++ return this.key;
++ }
++
++ @Override
++ public Writable value(Tuple tuple) {
++ if(this.value == null){
++ this.value = new Text();
++ }
++ this.value.set(tuple.getStringByField(this.valueField));
++ return this.value;
++ }
++}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2881ac9c/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/format/DelimitedRecordFormat.java
----------------------------------------------------------------------
diff --cc external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/format/DelimitedRecordFormat.java
index 0000000,0000000..2bd3abb
new file mode 100644
--- /dev/null
+++ b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/format/DelimitedRecordFormat.java
@@@ -1,0 -1,0 +1,88 @@@
++/**
++ * Licensed to the Apache Software Foundation (ASF) under one
++ * or more contributor license agreements. See the NOTICE file
++ * distributed with this work for additional information
++ * regarding copyright ownership. The ASF licenses this file
++ * to you under the Apache License, Version 2.0 (the
++ * "License"); you may not use this file except in compliance
++ * with the License. You may obtain a copy of the License at
++ *
++ * http://www.apache.org/licenses/LICENSE-2.0
++ *
++ * Unless required by applicable law or agreed to in writing, software
++ * distributed under the License is distributed on an "AS IS" BASIS,
++ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
++ * See the License for the specific language governing permissions and
++ * limitations under the License.
++ */
++package org.apache.storm.hdfs.bolt.format;
++
++import backtype.storm.tuple.Fields;
++import backtype.storm.tuple.Tuple;
++
++/**
++ * RecordFormat implementation that uses field and record delimiters.
++ * By default uses a comma (",") as the field delimiter and a
++ * newline ("\n") as the record delimiter.
++ *
++ * Also by default, this implementation will output all the
++ * field values in the tuple in the order they were declared. To
++ * override this behavior, call <code>withFields()</code> to
++ * specify which tuple fields to output.
++ *
++ */
++public class DelimitedRecordFormat implements RecordFormat {
++ public static final String DEFAULT_FIELD_DELIMITER = ",";
++ public static final String DEFAULT_RECORD_DELIMITER = "\n";
++ private String fieldDelimiter = DEFAULT_FIELD_DELIMITER;
++ private String recordDelimiter = DEFAULT_RECORD_DELIMITER;
++ private Fields fields = null;
++
++ /**
++ * Only output the specified fields.
++ *
++ * @param fields
++ * @return
++ */
++ public DelimitedRecordFormat withFields(Fields fields){
++ this.fields = fields;
++ return this;
++ }
++
++ /**
++ * Overrides the default field delimiter.
++ *
++ * @param delimiter
++ * @return
++ */
++ public DelimitedRecordFormat withFieldDelimiter(String delimiter){
++ this.fieldDelimiter = delimiter;
++ return this;
++ }
++
++ /**
++ * Overrides the default record delimiter.
++ *
++ * @param delimiter
++ * @return
++ */
++ public DelimitedRecordFormat withRecordDelimiter(String delimiter){
++ this.recordDelimiter = delimiter;
++ return this;
++ }
++
++ @Override
++ public byte[] format(Tuple tuple) {
++ StringBuilder sb = new StringBuilder();
++ Fields fields = this.fields == null ? tuple.getFields() : this.fields;
++ int size = fields.size();
++ for(int i = 0; i < size; i++){
++ sb.append(tuple.getValueByField(fields.get(i)));
++ if(i != size - 1){
++ sb.append(this.fieldDelimiter);
++ }
++ }
++ sb.append(this.recordDelimiter);
++ return sb.toString().getBytes();
++ }
++}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2881ac9c/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/format/FileNameFormat.java
----------------------------------------------------------------------
diff --cc external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/format/FileNameFormat.java
index 0000000,0000000..8344d78
new file mode 100644
--- /dev/null
+++ b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/format/FileNameFormat.java
@@@ -1,0 -1,0 +1,42 @@@
++/**
++ * Licensed to the Apache Software Foundation (ASF) under one
++ * or more contributor license agreements. See the NOTICE file
++ * distributed with this work for additional information
++ * regarding copyright ownership. The ASF licenses this file
++ * to you under the Apache License, Version 2.0 (the
++ * "License"); you may not use this file except in compliance
++ * with the License. You may obtain a copy of the License at
++ *
++ * http://www.apache.org/licenses/LICENSE-2.0
++ *
++ * Unless required by applicable law or agreed to in writing, software
++ * distributed under the License is distributed on an "AS IS" BASIS,
++ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
++ * See the License for the specific language governing permissions and
++ * limitations under the License.
++ */
++package org.apache.storm.hdfs.bolt.format;
++
++import backtype.storm.task.TopologyContext;
++
++import java.io.Serializable;
++import java.util.Map;
++
++/**
++ * Formatter interface for determining HDFS file names.
++ *
++ */
++public interface FileNameFormat extends Serializable {
++
++ void prepare(Map conf, TopologyContext topologyContext);
++
++ /**
++ * Returns the filename the HdfsBolt will create.
++ * @param rotation the current file rotation number (incremented on every rotation)
++ * @param timeStamp current time in milliseconds when the rotation occurs
++ * @return
++ */
++ String getName(long rotation, long timeStamp);
++
++ String getPath();
++}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2881ac9c/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/format/RecordFormat.java
----------------------------------------------------------------------
diff --cc external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/format/RecordFormat.java
index 0000000,0000000..3a26a61
new file mode 100644
--- /dev/null
+++ b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/format/RecordFormat.java
@@@ -1,0 -1,0 +1,32 @@@
++/**
++ * Licensed to the Apache Software Foundation (ASF) under one
++ * or more contributor license agreements. See the NOTICE file
++ * distributed with this work for additional information
++ * regarding copyright ownership. The ASF licenses this file
++ * to you under the Apache License, Version 2.0 (the
++ * "License"); you may not use this file except in compliance
++ * with the License. You may obtain a copy of the License at
++ *
++ * http://www.apache.org/licenses/LICENSE-2.0
++ *
++ * Unless required by applicable law or agreed to in writing, software
++ * distributed under the License is distributed on an "AS IS" BASIS,
++ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
++ * See the License for the specific language governing permissions and
++ * limitations under the License.
++ */
++package org.apache.storm.hdfs.bolt.format;
++
++
++import backtype.storm.tuple.Tuple;
++
++import java.io.Serializable;
++
++/**
++ * Formats a Tuple object into a byte array
++ * that will be written to HDFS.
++ *
++ */
++public interface RecordFormat extends Serializable {
++ byte[] format(Tuple tuple);
++}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2881ac9c/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/format/SequenceFormat.java
----------------------------------------------------------------------
diff --cc external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/format/SequenceFormat.java
index 0000000,0000000..6b30e2f
new file mode 100644
--- /dev/null
+++ b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/format/SequenceFormat.java
@@@ -1,0 -1,0 +1,40 @@@
++package org.apache.storm.hdfs.bolt.format;
++
++import backtype.storm.tuple.Tuple;
++import org.apache.hadoop.io.Writable;
++
++import java.io.Serializable;
++
++/**
++ * Interface for converting <code>Tuple</code> objects to HDFS sequence file key-value pairs.
++ *
++ */
++public interface SequenceFormat extends Serializable {
++ /**
++ * Key class used by implementation (e.g. IntWritable.class, etc.)
++ *
++ * @return
++ */
++ Class keyClass();
++
++ /**
++ * Value class used by implementation (e.g. Text.class, etc.)
++ * @return
++ */
++ Class valueClass();
++
++ /**
++ * Given a tuple, return the key that should be written to the sequence file.
++ *
++ * @param tuple
++ * @return
++ */
++ Writable key(Tuple tuple);
++
++ /**
++ * Given a tuple, return the value that should be written to the sequence file.
++ * @param tuple
++ * @return
++ */
++ Writable value(Tuple tuple);
++}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2881ac9c/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/rotation/FileRotationPolicy.java
----------------------------------------------------------------------
diff --cc external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/rotation/FileRotationPolicy.java
index 0000000,0000000..658579e
new file mode 100644
--- /dev/null
+++ b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/rotation/FileRotationPolicy.java
@@@ -1,0 -1,0 +1,51 @@@
++/**
++ * Licensed to the Apache Software Foundation (ASF) under one
++ * or more contributor license agreements. See the NOTICE file
++ * distributed with this work for additional information
++ * regarding copyright ownership. The ASF licenses this file
++ * to you under the Apache License, Version 2.0 (the
++ * "License"); you may not use this file except in compliance
++ * with the License. You may obtain a copy of the License at
++ *
++ * http://www.apache.org/licenses/LICENSE-2.0
++ *
++ * Unless required by applicable law or agreed to in writing, software
++ * distributed under the License is distributed on an "AS IS" BASIS,
++ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
++ * See the License for the specific language governing permissions and
++ * limitations under the License.
++ */
++package org.apache.storm.hdfs.bolt.rotation;
++
++
++import backtype.storm.tuple.Tuple;
++
++import java.io.Serializable;
++
++/**
++ * Used by the HdfsBolt to decide when to rotate files.
++ *
++ * The HdfsBolt will call the <code>mark()</code> method for every
++ * tuple received. If the <code>mark()</code> method returns
++ * <code>true</code> the HdfsBolt will perform a file rotation.
++ *
++ * After file rotation, the HdfsBolt will call the <code>reset()</code>
++ * method.
++ */
++public interface FileRotationPolicy extends Serializable {
++ /**
++ * Called for every tuple the HdfsBolt executes.
++ *
++ * @param tuple The tuple executed.
++ * @param offset current offset of file being written
++ * @return true if a file rotation should be performed
++ */
++ boolean mark(Tuple tuple, long offset);
++
++
++ /**
++ * Called after the HdfsBolt rotates a file.
++ *
++ */
++ void reset();
++}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2881ac9c/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/rotation/FileSizeRotationPolicy.java
----------------------------------------------------------------------
diff --cc external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/rotation/FileSizeRotationPolicy.java
index 0000000,0000000..70d170f
new file mode 100644
--- /dev/null
+++ b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/rotation/FileSizeRotationPolicy.java
@@@ -1,0 -1,0 +1,81 @@@
++/**
++ * Licensed to the Apache Software Foundation (ASF) under one
++ * or more contributor license agreements. See the NOTICE file
++ * distributed with this work for additional information
++ * regarding copyright ownership. The ASF licenses this file
++ * to you under the Apache License, Version 2.0 (the
++ * "License"); you may not use this file except in compliance
++ * with the License. You may obtain a copy of the License at
++ *
++ * http://www.apache.org/licenses/LICENSE-2.0
++ *
++ * Unless required by applicable law or agreed to in writing, software
++ * distributed under the License is distributed on an "AS IS" BASIS,
++ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
++ * See the License for the specific language governing permissions and
++ * limitations under the License.
++ */
++package org.apache.storm.hdfs.bolt.rotation;
++
++
++import backtype.storm.tuple.Tuple;
++import org.slf4j.Logger;
++import org.slf4j.LoggerFactory;
++
++/**
++ * File rotation policy that will rotate files when a certain
++ * file size is reached.
++ *
++ * For example:
++ * <pre>
++ * // rotate when files reach 5MB
++ * FileSizeRotationPolicy policy =
++ * new FileSizeRotationPolicy(5.0, Units.MB);
++ * </pre>
++ *
++ */
++public class FileSizeRotationPolicy implements FileRotationPolicy {
++ private static final Logger LOG = LoggerFactory.getLogger(FileSizeRotationPolicy.class);
++
++ public static enum Units {
++
++ KB((long)Math.pow(2, 10)),
++ MB((long)Math.pow(2, 20)),
++ GB((long)Math.pow(2, 30)),
++ TB((long)Math.pow(2, 40));
++
++ private long byteCount;
++
++ private Units(long byteCount){
++ this.byteCount = byteCount;
++ }
++
++ public long getByteCount(){
++ return byteCount;
++ }
++ }
++
++ private long maxBytes;
++
++ private long lastOffset = 0;
++ private long currentBytesWritten = 0;
++
++ public FileSizeRotationPolicy(float count, Units units){
++ this.maxBytes = (long)(count * units.getByteCount());
++ }
++
++ @Override
++ public boolean mark(Tuple tuple, long offset) {
++ long diff = offset - this.lastOffset;
++ this.currentBytesWritten += diff;
++ this.lastOffset = offset;
++ return this.currentBytesWritten >= this.maxBytes;
++ }
++
++ @Override
++ public void reset() {
++ this.currentBytesWritten = 0;
++ this.lastOffset = 0;
++ }
++
++}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2881ac9c/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/rotation/NoRotationPolicy.java
----------------------------------------------------------------------
diff --cc external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/rotation/NoRotationPolicy.java
index 0000000,0000000..884872e
new file mode 100644
--- /dev/null
+++ b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/rotation/NoRotationPolicy.java
@@@ -1,0 -1,0 +1,35 @@@
++/**
++ * Licensed to the Apache Software Foundation (ASF) under one
++ * or more contributor license agreements. See the NOTICE file
++ * distributed with this work for additional information
++ * regarding copyright ownership. The ASF licenses this file
++ * to you under the Apache License, Version 2.0 (the
++ * "License"); you may not use this file except in compliance
++ * with the License. You may obtain a copy of the License at
++ *
++ * http://www.apache.org/licenses/LICENSE-2.0
++ *
++ * Unless required by applicable law or agreed to in writing, software
++ * distributed under the License is distributed on an "AS IS" BASIS,
++ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
++ * See the License for the specific language governing permissions and
++ * limitations under the License.
++ */
++package org.apache.storm.hdfs.bolt.rotation;
++
++import backtype.storm.tuple.Tuple;
++
++/**
++ * File rotation policy that will never rotate...
++ * Just one big file. Intended for testing purposes.
++ */
++public class NoRotationPolicy implements FileRotationPolicy {
++ @Override
++ public boolean mark(Tuple tuple, long offset) {
++ return false;
++ }
++
++ @Override
++ public void reset() {
++ }
++}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2881ac9c/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/sync/CountSyncPolicy.java
----------------------------------------------------------------------
diff --cc external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/sync/CountSyncPolicy.java
index 0000000,0000000..2a69e4f
new file mode 100644
--- /dev/null
+++ b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/sync/CountSyncPolicy.java
@@@ -1,0 -1,0 +1,46 @@@
++/**
++ * Licensed to the Apache Software Foundation (ASF) under one
++ * or more contributor license agreements. See the NOTICE file
++ * distributed with this work for additional information
++ * regarding copyright ownership. The ASF licenses this file
++ * to you under the Apache License, Version 2.0 (the
++ * "License"); you may not use this file except in compliance
++ * with the License. You may obtain a copy of the License at
++ *
++ * http://www.apache.org/licenses/LICENSE-2.0
++ *
++ * Unless required by applicable law or agreed to in writing, software
++ * distributed under the License is distributed on an "AS IS" BASIS,
++ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
++ * See the License for the specific language governing permissions and
++ * limitations under the License.
++ */
++package org.apache.storm.hdfs.bolt.sync;
++
++
++import backtype.storm.tuple.Tuple;
++
++/**
++ * SyncPolicy implementation that will trigger a
++ * file system sync after a certain number of tuples
++ * have been processed.
++ */
++public class CountSyncPolicy implements SyncPolicy {
++ private int count;
++ private int executeCount = 0;
++
++ public CountSyncPolicy(int count){
++ this.count = count;
++ }
++
++ @Override
++ public boolean mark(Tuple tuple, long offset) {
++ this.executeCount++;
++ return this.executeCount >= this.count;
++ }
++
++ @Override
++ public void reset() {
++ this.executeCount = 0;
++ }
++}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2881ac9c/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/sync/SyncPolicy.java
----------------------------------------------------------------------
diff --cc external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/sync/SyncPolicy.java
index 0000000,0000000..205f13f
new file mode 100644
--- /dev/null
+++ b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/bolt/sync/SyncPolicy.java
@@@ -1,0 -1,0 +1,46 @@@
++/**
++ * Licensed to the Apache Software Foundation (ASF) under one
++ * or more contributor license agreements. See the NOTICE file
++ * distributed with this work for additional information
++ * regarding copyright ownership. The ASF licenses this file
++ * to you under the Apache License, Version 2.0 (the
++ * "License"); you may not use this file except in compliance
++ * with the License. You may obtain a copy of the License at
++ *
++ * http://www.apache.org/licenses/LICENSE-2.0
++ *
++ * Unless required by applicable law or agreed to in writing, software
++ * distributed under the License is distributed on an "AS IS" BASIS,
++ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
++ * See the License for the specific language governing permissions and
++ * limitations under the License.
++ */
++package org.apache.storm.hdfs.bolt.sync;
++
++import backtype.storm.tuple.Tuple;
++
++import java.io.Serializable;
++
++/**
++ * Interface for controlling when the HdfsBolt
++ * syncs and flushes the filesystem.
++ *
++ */
++public interface SyncPolicy extends Serializable {
++ /**
++ * Called for every tuple the HdfsBolt executes.
++ *
++ * @param tuple The tuple executed.
++ * @param offset current offset for the file being written
++ * @return true if a sync should be performed
++ */
++ boolean mark(Tuple tuple, long offset);
++
++
++ /**
++ * Called after the HdfsBolt performs a sync.
++ *
++ */
++ void reset();
++
++}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2881ac9c/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/common/rotation/MoveFileAction.java
----------------------------------------------------------------------
diff --cc external/storm-hdfs/src/main/java/org/apache/storm/hdfs/common/rotation/MoveFileAction.java
index 0000000,0000000..af6518b
new file mode 100644
--- /dev/null
+++ b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/common/rotation/MoveFileAction.java
@@@ -1,0 -1,0 +1,27 @@@
++package org.apache.storm.hdfs.common.rotation;
++
++import org.apache.hadoop.fs.FileSystem;
++import org.apache.hadoop.fs.Path;
++import org.slf4j.Logger;
++import org.slf4j.LoggerFactory;
++
++import java.io.IOException;
++
++public class MoveFileAction implements RotationAction {
++ private static final Logger LOG = LoggerFactory.getLogger(MoveFileAction.class);
++
++ private String destination;
++
++ public MoveFileAction toDestination(String destDir){
++ destination = destDir;
++ return this;
++ }
++
++ @Override
++ public void execute(FileSystem fileSystem, Path filePath) throws IOException {
++ Path destPath = new Path(destination, filePath.getName());
++ LOG.info("Moving file {} to {}", filePath, destPath);
++ boolean success = fileSystem.rename(filePath, destPath);
++ return;
++ }
++}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2881ac9c/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/common/rotation/RotationAction.java
----------------------------------------------------------------------
diff --cc external/storm-hdfs/src/main/java/org/apache/storm/hdfs/common/rotation/RotationAction.java
index 0000000,0000000..4469917
new file mode 100644
--- /dev/null
+++ b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/common/rotation/RotationAction.java
@@@ -1,0 -1,0 +1,12 @@@
++package org.apache.storm.hdfs.common.rotation;
++
++
++import org.apache.hadoop.fs.FileSystem;
++import org.apache.hadoop.fs.Path;
++
++import java.io.IOException;
++import java.io.Serializable;
++
++public interface RotationAction extends Serializable {
++ void execute(FileSystem fileSystem, Path filePath) throws IOException;
++}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2881ac9c/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/common/security/HdfsSecurityUtil.java
----------------------------------------------------------------------
diff --cc external/storm-hdfs/src/main/java/org/apache/storm/hdfs/common/security/HdfsSecurityUtil.java
index 0000000,0000000..adfaaf5
new file mode 100644
--- /dev/null
+++ b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/common/security/HdfsSecurityUtil.java
@@@ -1,0 -1,0 +1,48 @@@
++/**
++ * Licensed to the Apache Software Foundation (ASF) under one
++ * or more contributor license agreements. See the NOTICE file
++ * distributed with this work for additional information
++ * regarding copyright ownership. The ASF licenses this file
++ * to you under the Apache License, Version 2.0 (the
++ * "License"); you may not use this file except in compliance
++ * with the License. You may obtain a copy of the License at
++ *
++ * http://www.apache.org/licenses/LICENSE-2.0
++ *
++ * Unless required by applicable law or agreed to in writing, software
++ * distributed under the License is distributed on an "AS IS" BASIS,
++ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
++ * See the License for the specific language governing permissions and
++ * limitations under the License.
++ */
++package org.apache.storm.hdfs.common.security;
++
++import org.apache.hadoop.conf.Configuration;
++import org.apache.hadoop.security.SecurityUtil;
++import org.apache.hadoop.security.UserGroupInformation;
++
++import java.io.IOException;
++import java.util.Map;
++
++/**
++ * This class provides util methods for storm-hdfs connector communicating
++ * with secured HDFS.
++ */
++public class HdfsSecurityUtil {
++ public static final String STORM_KEYTAB_FILE_KEY = "storm.keytab.file";
++ public static final String STORM_USER_NAME_KEY = "storm.kerberos.principal";
++
++ public static void login(Map conf, Configuration hdfsConfig) throws IOException {
++ if (UserGroupInformation.isSecurityEnabled()) {
++ String keytab = (String) conf.get(STORM_KEYTAB_FILE_KEY);
++ if (keytab != null) {
++ hdfsConfig.set(STORM_KEYTAB_FILE_KEY, keytab);
++ }
++ String userName = (String) conf.get(STORM_USER_NAME_KEY);
++ if (userName != null) {
++ hdfsConfig.set(STORM_USER_NAME_KEY, userName);
++ }
++ SecurityUtil.login(hdfsConfig, STORM_KEYTAB_FILE_KEY, STORM_USER_NAME_KEY);
++ }
++ }
++}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/2881ac9c/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/trident/HdfsState.java
----------------------------------------------------------------------
diff --cc external/storm-hdfs/src/main/java/org/apache/storm/hdfs/trident/HdfsState.java
index 0000000,0000000..f2378cd
new file mode 100644
--- /dev/null
+++ b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/trident/HdfsState.java
@@@ -1,0 -1,0 +1,295 @@@
++package org.apache.storm.hdfs.trident;
++
++import backtype.storm.task.IMetricsContext;
++import backtype.storm.topology.FailedException;
++import org.apache.hadoop.conf.Configuration;
++import org.apache.hadoop.fs.FSDataOutputStream;
++import org.apache.hadoop.fs.FileSystem;
++import org.apache.hadoop.fs.Path;
++import org.apache.hadoop.hdfs.client.HdfsDataOutputStream;
++import org.apache.hadoop.io.SequenceFile;
++import org.apache.hadoop.io.compress.CompressionCodecFactory;
++import org.apache.storm.hdfs.common.rotation.RotationAction;
++import org.apache.storm.hdfs.common.security.HdfsSecurityUtil;
++import org.apache.storm.hdfs.trident.format.FileNameFormat;
++import org.apache.storm.hdfs.trident.format.RecordFormat;
++import org.apache.storm.hdfs.trident.format.SequenceFormat;
++import org.apache.storm.hdfs.trident.rotation.FileRotationPolicy;
++
++import org.slf4j.Logger;
++import org.slf4j.LoggerFactory;
++import storm.trident.operation.TridentCollector;
++import storm.trident.state.State;
++import storm.trident.tuple.TridentTuple;
++
++import java.io.IOException;
++import java.io.Serializable;
++import java.net.URI;
++import java.util.ArrayList;
++import java.util.EnumSet;
++import java.util.List;
++import java.util.Map;
++
++public class HdfsState implements State {
++
++ public static abstract class Options implements Serializable {
++
++ protected String fsUrl;
++ protected String configKey;
++ protected FileSystem fs;
++ private Path currentFile;
++ protected FileRotationPolicy rotationPolicy;
++ protected FileNameFormat fileNameFormat;
++ protected int rotation = 0;
++ protected Configuration hdfsConfig;
++ protected ArrayList<RotationAction> rotationActions = new ArrayList<RotationAction>();
++
++ abstract void closeOutputFile() throws IOException;
++
++ abstract Path createOutputFile() throws IOException;
++
++ abstract void execute(List<TridentTuple> tuples) throws IOException;
++
++ abstract void doPrepare(Map conf, int partitionIndex, int numPartitions) throws IOException;
++
++ protected void rotateOutputFile() throws IOException {
++ LOG.info("Rotating output file...");
++ long start = System.currentTimeMillis();
++ closeOutputFile();
++ this.rotation++;
++
++ Path newFile = createOutputFile();
++ LOG.info("Performing {} file rotation actions.", this.rotationActions.size());
++ for(RotationAction action : this.rotationActions){
++ action.execute(this.fs, this.currentFile);
++ }
++ this.currentFile = newFile;
++ long time = System.currentTimeMillis() - start;
++ LOG.info("File rotation took {} ms.", time);
++
++
++ }
++
++ void prepare(Map conf, int partitionIndex, int numPartitions){
++ if (this.rotationPolicy == null) throw new IllegalStateException("RotationPolicy must be specified.");
++ if (this.fsUrl == null) {
++ throw new IllegalStateException("File system URL must be specified.");
++ }
++ this.fileNameFormat.prepare(conf, partitionIndex, numPartitions);
++ this.hdfsConfig = new Configuration();
++ Map<String, Object> map = (Map<String, Object>)conf.get(this.configKey);
++ if(map != null){
++ for(String key : map.keySet()){
++ this.hdfsConfig.set(key, String.valueOf(map.get(key)));
++ }
++ }
++ try{
++ HdfsSecurityUtil.login(conf, hdfsConfig);
++ doPrepare(conf, partitionIndex, numPartitions);
++ this.currentFile = createOutputFile();
++
++ } catch (Exception e){
++ throw new RuntimeException("Error preparing HdfsState: " + e.getMessage(), e);
++ }
++ }
++
++ }
++
++ public static class HdfsFileOptions extends Options {
++
++ private FSDataOutputStream out;
++ protected RecordFormat format;
++ private long offset = 0;
++
++ public HdfsFileOptions withFsUrl(String fsUrl){
++ this.fsUrl = fsUrl;
++ return this;
++ }
++
++ public HdfsFileOptions withConfigKey(String configKey){
++ this.configKey = configKey;
++ return this;
++ }
++
++ public HdfsFileOptions withFileNameFormat(FileNameFormat fileNameFormat){
++ this.fileNameFormat = fileNameFormat;
++ return this;
++ }
++
++ public HdfsFileOptions withRecordFormat(RecordFormat format){
++ this.format = format;
++ return this;
++ }
++
++ public HdfsFileOptions withRotationPolicy(FileRotationPolicy rotationPolicy){
++ this.rotationPolicy = rotationPolicy;
++ return this;
++ }
++
++ public HdfsFileOptions addRotationAction(RotationAction action){
++ this.rotationActions.add(action);
++ return this;
++ }
++
++ @Override
++ void doPrepare(Map conf, int partitionIndex, int numPartitions) throws IOException {
++ LOG.info("Preparing HDFS Bolt...");
++ this.fs = FileSystem.get(URI.create(this.fsUrl), hdfsConfig);
++ }
++
++ @Override
++ void closeOutputFile() throws IOException {
++ this.out.hsync();
++ this.out.close();
++ }
++
++ @Override
++ Path createOutputFile() throws IOException {
++ Path path = new Path(this.fileNameFormat.getPath(), this.fileNameFormat.getName(this.rotation, System.currentTimeMillis()));
++ this.out = this.fs.create(path);
++ return path;
++ }
++
++ @Override
++ public void execute(List<TridentTuple> tuples) throws IOException {
++ boolean rotated = false;
++ for(TridentTuple tuple : tuples){
++ byte[] bytes = this.format.format(tuple);
++ out.write(bytes);
++ this.offset += bytes.length;
++
++ if(this.rotationPolicy.mark(tuple, this.offset)){
++ rotateOutputFile();
++ this.offset = 0;
++ this.rotationPolicy.reset();
++ rotated = true;
++ }
++ }
++ if(!rotated){
++ if(this.out instanceof HdfsDataOutputStream){
++ ((HdfsDataOutputStream)this.out).hsync(EnumSet.of(HdfsDataOutputStream.SyncFlag.UPDATE_LENGTH));
++ } else {
++ this.out.hsync();
++ }
++ }
++ }
++ }
++
++ public static class SequenceFileOptions extends Options {
++ private SequenceFormat format;
++ private SequenceFile.CompressionType compressionType = SequenceFile.CompressionType.RECORD;
++ private SequenceFile.Writer writer;
++ private String compressionCodec = "default";
++ private transient CompressionCodecFactory codecFactory;
++
++ public SequenceFileOptions withCompressionCodec(String codec){
++ this.compressionCodec = codec;
++ return this;
++ }
++
++ public SequenceFileOptions withFsUrl(String fsUrl) {
++ this.fsUrl = fsUrl;
++ return this;
++ }
++
++ public SequenceFileOptions withConfigKey(String configKey){
++ this.configKey = configKey;
++ return this;
++ }
++
++ public SequenceFileOptions withFileNameFormat(FileNameFormat fileNameFormat) {
++ this.fileNameFormat = fileNameFormat;
++ return this;
++ }
++
++ public SequenceFileOptions withSequenceFormat(SequenceFormat format) {
++ this.format = format;
++ return this;
++ }
++
++ public SequenceFileOptions withRotationPolicy(FileRotationPolicy rotationPolicy) {
++ this.rotationPolicy = rotationPolicy;
++ return this;
++ }
++
++ public SequenceFileOptions withCompressionType(SequenceFile.CompressionType compressionType){
++ this.compressionType = compressionType;
++ return this;
++ }
++
++ public SequenceFileOptions addRotationAction(RotationAction action){
++ this.rotationActions.add(action);
++ return this;
++ }
++
++ @Override
++ void doPrepare(Map conf, int partitionIndex, int numPartitions) throws IOException {
++ LOG.info("Preparing Sequence File State...");
++ if (this.format == null) throw new IllegalStateException("SequenceFormat must be specified.");
++
++ this.fs = FileSystem.get(URI.create(this.fsUrl), hdfsConfig);
++ this.codecFactory = new CompressionCodecFactory(hdfsConfig);
++ }
++
++ @Override
++ Path createOutputFile() throws IOException {
++ Path p = new Path(this.fsUrl + this.fileNameFormat.getPath(), this.fileNameFormat.getName(this.rotation, System.currentTimeMillis()));
++ this.writer = SequenceFile.createWriter(
++ this.hdfsConfig,
++ SequenceFile.Writer.file(p),
++ SequenceFile.Writer.keyClass(this.format.keyClass()),
++ SequenceFile.Writer.valueClass(this.format.valueClass()),
++ SequenceFile.Writer.compression(this.compressionType, this.codecFactory.getCodecByName(this.compressionCodec))
++ );
++ return p;
++ }
++
++ @Override
++ void closeOutputFile() throws IOException {
++ this.writer.hsync();
++ this.writer.close();
++ }
++
++ @Override
++ public void execute(List<TridentTuple> tuples) throws IOException {
++ for(TridentTuple tuple : tuples) {
++ this.writer.append(this.format.key(tuple), this.format.value(tuple));
++ long offset = this.writer.getLength();
++
++ if (this.rotationPolicy.mark(tuple, offset)) {
++ rotateOutputFile();
++ this.rotationPolicy.reset();
++ }
++ }
++ }
++
++ }
++
++ public static final Logger LOG = LoggerFactory.getLogger(HdfsState.class);
++ private Options options;
++
++ HdfsState(Options options){
++ this.options = options;
++ }
++
++ void prepare(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions){
++ this.options.prepare(conf, partitionIndex, numPartitions);
++ }
++
++ @Override
++ public void beginCommit(Long txId) {
++ }
++
++ @Override
++ public void commit(Long txId) {
++ }
++
++ public void updateState(List<TridentTuple> tuples, TridentCollector tridentCollector){
++ try{
++ this.options.execute(tuples);
++ } catch (IOException e){
++ LOG.warn("Failing batch due to IOException.", e);
++ throw new FailedException(e);
++ }
++ }
++}