You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ka...@apache.org on 2016/11/24 15:32:51 UTC
[1/3] storm git commit: STORM-2208: HDFS State Throws
FileNotFoundException in Azure Data Lake Store file system (adl://) - Close
OutputStream before file gets deleted
Repository: storm
Updated Branches:
refs/heads/1.0.x-branch 8e7fc73ad -> 872f6e2c9
STORM-2208: HDFS State Throws FileNotFoundException in Azure Data Lake Store file system (adl://)
- Close OutputStream before file gets deleted
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/43b62fe8
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/43b62fe8
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/43b62fe8
Branch: refs/heads/1.0.x-branch
Commit: 43b62fe8215b5c882b24a7429b6b6a8d69daf873
Parents: 8e7fc73
Author: Hugo Louro <hm...@gmail.com>
Authored: Sat Nov 19 08:58:33 2016 -0800
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Fri Nov 25 00:32:21 2016 +0900
----------------------------------------------------------------------
.../apache/storm/hdfs/trident/HdfsState.java | 31 ++++++++------------
1 file changed, 12 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/43b62fe8/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/trident/HdfsState.java
----------------------------------------------------------------------
diff --git a/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/trident/HdfsState.java b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/trident/HdfsState.java
index 6c4acca..43993c9 100644
--- a/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/trident/HdfsState.java
+++ b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/trident/HdfsState.java
@@ -17,9 +17,6 @@
*/
package org.apache.storm.hdfs.trident;
-import org.apache.storm.Config;
-import org.apache.storm.task.IMetricsContext;
-import org.apache.storm.topology.FailedException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
@@ -29,6 +26,7 @@ import org.apache.hadoop.hdfs.client.HdfsDataOutputStream;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
+import org.apache.storm.Config;
import org.apache.storm.hdfs.common.rotation.RotationAction;
import org.apache.storm.hdfs.common.security.HdfsSecurityUtil;
import org.apache.storm.hdfs.trident.format.FileNameFormat;
@@ -37,11 +35,13 @@ import org.apache.storm.hdfs.trident.format.SequenceFormat;
import org.apache.storm.hdfs.trident.rotation.FileRotationPolicy;
import org.apache.storm.hdfs.trident.rotation.FileSizeRotationPolicy;
import org.apache.storm.hdfs.trident.rotation.TimedRotationPolicy;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.storm.task.IMetricsContext;
+import org.apache.storm.topology.FailedException;
import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.state.State;
import org.apache.storm.trident.tuple.TridentTuple;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.BufferedReader;
import java.io.BufferedWriter;
@@ -490,16 +490,18 @@ public class HdfsState implements State {
}
private void updateIndex(long txId) {
- FSDataOutputStream out = null;
LOG.debug("Starting index update.");
- try {
- Path tmpPath = tmpFilePath(indexFilePath.toString());
- out = this.options.fs.create(tmpPath, true);
- BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(out));
+ final Path tmpPath = tmpFilePath(indexFilePath.toString());
+
+ try (FSDataOutputStream out = this.options.fs.create(tmpPath, true);
+ BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(out))) {
TxnRecord txnRecord = new TxnRecord(txId, options.currentFile.toString(), this.options.getCurrentOffset());
bw.write(txnRecord.toString());
bw.newLine();
bw.flush();
+ out.close(); /* In non error scenarios, for the Azure Data Lake Store File System (adl://),
+ the output stream must be closed before the file associated with it is deleted.
+ For ADLFS deleting the file also removes any handles to the file, hence out.close() will fail. */
/*
* Delete the current index file and rename the tmp file to atomically
* replace the index file. Orphan .tmp files are handled in getTxnRecord.
@@ -511,15 +513,6 @@ public class HdfsState implements State {
} catch (IOException e) {
LOG.warn("Begin commit failed due to IOException. Failing batch", e);
throw new FailedException(e);
- } finally {
- if (out != null) {
- try {
- out.close();
- } catch (IOException e) {
- LOG.warn("Begin commit failed due to IOException. Failing batch", e);
- throw new FailedException(e);
- }
- }
}
}
[2/3] storm git commit: Merge branch 'STORM-2208-1.0.x-merge' into
1.0.x-branch
Posted by ka...@apache.org.
Merge branch 'STORM-2208-1.0.x-merge' into 1.0.x-branch
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/4f81f689
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/4f81f689
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/4f81f689
Branch: refs/heads/1.0.x-branch
Commit: 4f81f68904d82e838aac947e5f4f0bda90548196
Parents: 8e7fc73 43b62fe
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Fri Nov 25 00:32:26 2016 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Fri Nov 25 00:32:26 2016 +0900
----------------------------------------------------------------------
.../apache/storm/hdfs/trident/HdfsState.java | 31 ++++++++------------
1 file changed, 12 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
[3/3] storm git commit: add STORM-2208 to CHANGELOG
Posted by ka...@apache.org.
add STORM-2208 to CHANGELOG
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/872f6e2c
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/872f6e2c
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/872f6e2c
Branch: refs/heads/1.0.x-branch
Commit: 872f6e2c99bd93e0b0d030d0a90871eac339ba92
Parents: 4f81f68
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Fri Nov 25 00:32:39 2016 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Fri Nov 25 00:32:39 2016 +0900
----------------------------------------------------------------------
CHANGELOG.md | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/872f6e2c/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index d0bb3a4..42dde1d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,5 @@
## 1.0.3
+ * STORM-2208: HDFS State Throws FileNotFoundException in Azure Data Lake Store file system (adl://)
* STORM-2213: ShellSpout has race condition when ShellSpout is being inactive longer than heartbeat timeout
* STORM-2210: remove array shuffle from ShuffleGrouping
* STORM-2052: Kafka Spout - New Client API - Performance Improvements