You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by bo...@apache.org on 2018/08/02 17:17:33 UTC
[10/50] [abbrv] hadoop git commit: HADOOP-15607. AliyunOSS: fix
duplicated partNumber issue in AliyunOSSBlockOutputStream. Contributed by
Jinhu Wu.
HADOOP-15607. AliyunOSS: fix duplicated partNumber issue in AliyunOSSBlockOutputStream. Contributed by Jinhu Wu.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/0857f116
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/0857f116
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/0857f116
Branch: refs/heads/YARN-7402
Commit: 0857f116b754d83d3c540cd6f989087af24fef27
Parents: 007e6f5
Author: Sammi Chen <sa...@intel.com>
Authored: Mon Jul 30 10:53:44 2018 +0800
Committer: Sammi Chen <sa...@intel.com>
Committed: Mon Jul 30 10:53:44 2018 +0800
----------------------------------------------------------------------
.../aliyun/oss/AliyunOSSBlockOutputStream.java | 59 ++++++++++++--------
.../fs/aliyun/oss/AliyunOSSFileSystemStore.java | 2 +
.../oss/TestAliyunOSSBlockOutputStream.java | 12 +++-
3 files changed, 49 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0857f116/hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/AliyunOSSBlockOutputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/AliyunOSSBlockOutputStream.java b/hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/AliyunOSSBlockOutputStream.java
index 12d551b..0a833b2 100644
--- a/hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/AliyunOSSBlockOutputStream.java
+++ b/hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/AliyunOSSBlockOutputStream.java
@@ -33,7 +33,9 @@ import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
@@ -50,7 +52,7 @@ public class AliyunOSSBlockOutputStream extends OutputStream {
private boolean closed;
private String key;
private File blockFile;
- private List<File> blockFiles = new ArrayList<>();
+ private Map<Integer, File> blockFiles = new HashMap<>();
private long blockSize;
private int blockId = 0;
private long blockWritten = 0L;
@@ -94,8 +96,9 @@ public class AliyunOSSBlockOutputStream extends OutputStream {
blockStream.flush();
blockStream.close();
- if (!blockFiles.contains(blockFile)) {
- blockFiles.add(blockFile);
+ if (!blockFiles.values().contains(blockFile)) {
+ blockId++;
+ blockFiles.put(blockId, blockFile);
}
try {
@@ -107,7 +110,7 @@ public class AliyunOSSBlockOutputStream extends OutputStream {
ListenableFuture<PartETag> partETagFuture =
executorService.submit(() -> {
PartETag partETag = store.uploadPart(blockFile, key, uploadId,
- blockId + 1);
+ blockId);
return partETag;
});
partETagsFutures.add(partETagFuture);
@@ -120,11 +123,7 @@ public class AliyunOSSBlockOutputStream extends OutputStream {
store.completeMultipartUpload(key, uploadId, partETags);
}
} finally {
- for (File tFile: blockFiles) {
- if (tFile.exists() && !tFile.delete()) {
- LOG.warn("Failed to delete temporary file {}", tFile);
- }
- }
+ removePartFiles();
closed = true;
}
}
@@ -141,38 +140,52 @@ public class AliyunOSSBlockOutputStream extends OutputStream {
if (closed) {
throw new IOException("Stream closed.");
}
- try {
- blockStream.write(b, off, len);
- blockWritten += len;
- if (blockWritten >= blockSize) {
- uploadCurrentPart();
- blockWritten = 0L;
+ blockStream.write(b, off, len);
+ blockWritten += len;
+ if (blockWritten >= blockSize) {
+ uploadCurrentPart();
+ blockWritten = 0L;
+ }
+ }
+
+ private void removePartFiles() throws IOException {
+ for (ListenableFuture<PartETag> partETagFuture : partETagsFutures) {
+ if (!partETagFuture.isDone()) {
+ continue;
}
- } finally {
- for (File tFile: blockFiles) {
- if (tFile.exists() && !tFile.delete()) {
- LOG.warn("Failed to delete temporary file {}", tFile);
+
+ try {
+ File blockFile = blockFiles.get(partETagFuture.get().getPartNumber());
+ if (blockFile != null && blockFile.exists() && !blockFile.delete()) {
+ LOG.warn("Failed to delete temporary file {}", blockFile);
}
+ } catch (InterruptedException | ExecutionException e) {
+ throw new IOException(e);
}
}
}
private void uploadCurrentPart() throws IOException {
- blockFiles.add(blockFile);
blockStream.flush();
blockStream.close();
if (blockId == 0) {
uploadId = store.getUploadId(key);
}
+
+ blockId++;
+ blockFiles.put(blockId, blockFile);
+
+ File currentFile = blockFile;
+ int currentBlockId = blockId;
ListenableFuture<PartETag> partETagFuture =
executorService.submit(() -> {
- PartETag partETag = store.uploadPart(blockFile, key, uploadId,
- blockId + 1);
+ PartETag partETag = store.uploadPart(currentFile, key, uploadId,
+ currentBlockId);
return partETag;
});
partETagsFutures.add(partETagFuture);
+ removePartFiles();
blockFile = newBlockFile();
- blockId++;
blockStream = new BufferedOutputStream(new FileOutputStream(blockFile));
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0857f116/hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/AliyunOSSFileSystemStore.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/AliyunOSSFileSystemStore.java b/hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/AliyunOSSFileSystemStore.java
index 5e21759..dc5f99ee 100644
--- a/hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/AliyunOSSFileSystemStore.java
+++ b/hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/AliyunOSSFileSystemStore.java
@@ -450,6 +450,8 @@ public class AliyunOSSFileSystemStore {
request.setRange(byteStart, byteEnd);
return ossClient.getObject(request).getObjectContent();
} catch (OSSException | ClientException e) {
+ LOG.error("Exception thrown when store retrieves key: "
+ + key + ", exception: " + e);
return null;
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0857f116/hadoop-tools/hadoop-aliyun/src/test/java/org/apache/hadoop/fs/aliyun/oss/TestAliyunOSSBlockOutputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aliyun/src/test/java/org/apache/hadoop/fs/aliyun/oss/TestAliyunOSSBlockOutputStream.java b/hadoop-tools/hadoop-aliyun/src/test/java/org/apache/hadoop/fs/aliyun/oss/TestAliyunOSSBlockOutputStream.java
index 365d931..6fe6f03 100644
--- a/hadoop-tools/hadoop-aliyun/src/test/java/org/apache/hadoop/fs/aliyun/oss/TestAliyunOSSBlockOutputStream.java
+++ b/hadoop-tools/hadoop-aliyun/src/test/java/org/apache/hadoop/fs/aliyun/oss/TestAliyunOSSBlockOutputStream.java
@@ -31,6 +31,7 @@ import org.junit.rules.Timeout;
import java.io.IOException;
import static org.apache.hadoop.fs.aliyun.oss.Constants.MULTIPART_UPLOAD_PART_SIZE_DEFAULT;
+import static org.apache.hadoop.fs.contract.ContractTestUtils.IO_CHUNK_BUFFER_SIZE;
/**
* Tests regular and multi-part upload functionality for
@@ -48,7 +49,10 @@ public class TestAliyunOSSBlockOutputStream {
public void setUp() throws Exception {
Configuration conf = new Configuration();
conf.setLong(Constants.MIN_MULTIPART_UPLOAD_THRESHOLD_KEY, 5 * 1024 * 1024);
- conf.setInt(Constants.MULTIPART_UPLOAD_PART_SIZE_KEY, 5 * 1024 * 1024);
+ conf.setInt(Constants.MULTIPART_UPLOAD_PART_SIZE_KEY, 1024 * 1024);
+ conf.setInt(IO_CHUNK_BUFFER_SIZE,
+ conf.getInt(Constants.MULTIPART_UPLOAD_PART_SIZE_KEY, 0));
+ conf.setInt(Constants.UPLOAD_ACTIVE_BLOCKS_KEY, 20);
fs = AliyunOSSTestUtils.createTestFileSystem(conf);
}
@@ -85,6 +89,12 @@ public class TestAliyunOSSBlockOutputStream {
}
@Test
+ public void testMultiPartUploadConcurrent() throws IOException {
+ ContractTestUtils.createAndVerifyFile(fs, getTestPath(),
+ 50 * 1024 * 1024 - 1);
+ }
+
+ @Test
public void testHugeUpload() throws IOException {
ContractTestUtils.createAndVerifyFile(fs, getTestPath(),
MULTIPART_UPLOAD_PART_SIZE_DEFAULT - 1);
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org