You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by kk...@apache.org on 2019/02/04 09:03:27 UTC
[flink] branch master updated (2ef631f -> 892ff1d)
This is an automated email from the ASF dual-hosted git repository.
kkloudas pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 2ef631f [FLINK-10395][build] Remove lefover 'codebase' references
new 81e361c [FLINK-11419][filesystem] Wait until lease is revoked before truncating file in Hadoop.
new 892ff1d [FLINK-11419][filesystem] Wait for lease to be revoked when truncating file in Hadoop.
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../hdfs/HadoopRecoverableFsDataOutputStream.java | 46 ++++++++++++++--------
1 file changed, 30 insertions(+), 16 deletions(-)
[flink] 02/02: [FLINK-11419][filesystem] Wait for lease to be
revoked when truncating file in Hadoop.
Posted by kk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kkloudas pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 892ff1dff01f7ec10df7e8e41ed990b92eca0b34
Author: Kostas Kloudas <kk...@gmail.com>
AuthorDate: Thu Jan 31 13:36:09 2019 +0100
[FLINK-11419][filesystem] Wait for lease to be revoked when truncating file in Hadoop.
This closes #7588.
---
.../hdfs/HadoopRecoverableFsDataOutputStream.java | 52 ++++++++++++----------
1 file changed, 29 insertions(+), 23 deletions(-)
diff --git a/flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableFsDataOutputStream.java b/flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableFsDataOutputStream.java
index a716029..be0d134 100644
--- a/flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableFsDataOutputStream.java
+++ b/flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableFsDataOutputStream.java
@@ -86,20 +86,7 @@ class HadoopRecoverableFsDataOutputStream extends RecoverableFsDataOutputStream
this.targetFile = checkNotNull(recoverable.targetFile());
this.tempFile = checkNotNull(recoverable.tempFile());
- waitUntilLeaseIsRevoked(tempFile);
-
- // truncate back and append
- boolean truncated;
- try {
- truncated = truncate(fs, tempFile, recoverable.offset());
- } catch (Exception e) {
- throw new IOException("Missing data in tmp file: " + tempFile, e);
- }
-
- if (!truncated) {
- // Truncate did not complete immediately, we must wait for the operation to complete and release the lease
- waitUntilLeaseIsRevoked(tempFile);
- }
+ safelyTruncateFile(fs, tempFile, recoverable);
out = fs.append(tempFile);
@@ -162,6 +149,30 @@ class HadoopRecoverableFsDataOutputStream extends RecoverableFsDataOutputStream
// Hadoop 2.7, which have no truncation calls for HDFS.
// ------------------------------------------------------------------------
+ private static void safelyTruncateFile(
+ final FileSystem fileSystem,
+ final Path path,
+ final HadoopFsRecoverable recoverable) throws IOException {
+
+ ensureTruncateInitialized();
+
+ waitUntilLeaseIsRevoked(fileSystem, path);
+
+ // truncate back and append
+ boolean truncated;
+ try {
+ truncated = truncate(fileSystem, path, recoverable.offset());
+ } catch (Exception e) {
+ throw new IOException("Problem while truncating file: " + path, e);
+ }
+
+ if (!truncated) {
+ // Truncate did not complete immediately, we must wait for
+ // the operation to complete and release the lease.
+ waitUntilLeaseIsRevoked(fileSystem, path);
+ }
+ }
+
private static void ensureTruncateInitialized() throws FlinkRuntimeException {
if (truncateHandle == null) {
Method truncateMethod;
@@ -180,7 +191,7 @@ class HadoopRecoverableFsDataOutputStream extends RecoverableFsDataOutputStream
}
}
- static boolean truncate(FileSystem hadoopFs, Path file, long length) throws IOException {
+ private static boolean truncate(final FileSystem hadoopFs, final Path file, final long length) throws IOException {
if (truncateHandle != null) {
try {
return (Boolean) truncateHandle.invoke(hadoopFs, file, length);
@@ -197,7 +208,7 @@ class HadoopRecoverableFsDataOutputStream extends RecoverableFsDataOutputStream
else {
throw new IllegalStateException("Truncation handle has not been initialized");
}
- return true;
+ return false;
}
// ------------------------------------------------------------------------
@@ -268,12 +279,7 @@ class HadoopRecoverableFsDataOutputStream extends RecoverableFsDataOutputStream
if (srcStatus.getLen() > expectedLength) {
// can happen if we go from persist to recovering for commit directly
// truncate the trailing junk away
- try {
- truncate(fs, src, expectedLength);
- } catch (Exception e) {
- // this can happen if the file is smaller than expected
- throw new IOException("Problem while truncating file: " + src, e);
- }
+ safelyTruncateFile(fs, src, recoverable);
}
// rename to final location (if it exists, overwrite it)
@@ -312,7 +318,7 @@ class HadoopRecoverableFsDataOutputStream extends RecoverableFsDataOutputStream
*
* @param path The path to the file we want to resume writing to.
*/
- private boolean waitUntilLeaseIsRevoked(final Path path) throws IOException {
+ private static boolean waitUntilLeaseIsRevoked(final FileSystem fs, final Path path) throws IOException {
Preconditions.checkState(fs instanceof DistributedFileSystem);
final DistributedFileSystem dfs = (DistributedFileSystem) fs;
[flink] 01/02: [FLINK-11419][filesystem] Wait until lease is
revoked before truncating file in Hadoop.
Posted by kk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kkloudas pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 81e361ce3ae58ce84d57cce106dd89a2815937d2
Author: EAlexRojas <al...@gmail.com>
AuthorDate: Mon Jan 28 19:01:48 2019 +0100
[FLINK-11419][filesystem] Wait until lease is revoked before truncating file in Hadoop.
---
.../fs/hdfs/HadoopRecoverableFsDataOutputStream.java | 16 ++++++++++++----
1 file changed, 12 insertions(+), 4 deletions(-)
diff --git a/flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableFsDataOutputStream.java b/flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableFsDataOutputStream.java
index c688b32..a716029 100644
--- a/flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableFsDataOutputStream.java
+++ b/flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableFsDataOutputStream.java
@@ -86,14 +86,21 @@ class HadoopRecoverableFsDataOutputStream extends RecoverableFsDataOutputStream
this.targetFile = checkNotNull(recoverable.targetFile());
this.tempFile = checkNotNull(recoverable.tempFile());
+ waitUntilLeaseIsRevoked(tempFile);
+
// truncate back and append
+ boolean truncated;
try {
- truncate(fs, tempFile, recoverable.offset());
+ truncated = truncate(fs, tempFile, recoverable.offset());
} catch (Exception e) {
throw new IOException("Missing data in tmp file: " + tempFile, e);
}
- waitUntilLeaseIsRevoked(tempFile);
+ if (!truncated) {
+ // Truncate did not complete immediately, we must wait for the operation to complete and release the lease
+ waitUntilLeaseIsRevoked(tempFile);
+ }
+
out = fs.append(tempFile);
// sanity check
@@ -173,10 +180,10 @@ class HadoopRecoverableFsDataOutputStream extends RecoverableFsDataOutputStream
}
}
- static void truncate(FileSystem hadoopFs, Path file, long length) throws IOException {
+ static boolean truncate(FileSystem hadoopFs, Path file, long length) throws IOException {
if (truncateHandle != null) {
try {
- truncateHandle.invoke(hadoopFs, file, length);
+ return (Boolean) truncateHandle.invoke(hadoopFs, file, length);
}
catch (InvocationTargetException e) {
ExceptionUtils.rethrowIOException(e.getTargetException());
@@ -190,6 +197,7 @@ class HadoopRecoverableFsDataOutputStream extends RecoverableFsDataOutputStream
else {
throw new IllegalStateException("Truncation handle has not been initialized");
}
+ return true;
}
// ------------------------------------------------------------------------