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 dr...@apache.org on 2016/08/01 15:55:11 UTC
[20/50] [abbrv] hadoop git commit: HADOOP-13354. Update WASB driver
to use the latest version (4.2.0) of SDK for Microsoft Azure Storage Clients.
Contributed by Sivaguru Sankaridurg.
HADOOP-13354. Update WASB driver to use the latest version (4.2.0) of SDK for Microsoft Azure Storage Clients. Contributed by Sivaguru Sankaridurg.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/b43de800
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/b43de800
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/b43de800
Branch: refs/heads/HADOOP-12756
Commit: b43de80031d1272e8a08ea5bd31027efe45e9d70
Parents: eb7ff0c
Author: Chris Nauroth <cn...@apache.org>
Authored: Wed Jul 27 15:50:28 2016 -0700
Committer: Chris Nauroth <cn...@apache.org>
Committed: Wed Jul 27 15:50:38 2016 -0700
----------------------------------------------------------------------
hadoop-project/pom.xml | 2 +-
.../hadoop/fs/azure/BlockBlobAppendStream.java | 99 ++++++++++++++++----
.../hadoop/fs/azure/SendRequestIntercept.java | 2 +-
.../hadoop/fs/azure/StorageInterfaceImpl.java | 2 +-
4 files changed, 85 insertions(+), 20 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b43de800/hadoop-project/pom.xml
----------------------------------------------------------------------
diff --git a/hadoop-project/pom.xml b/hadoop-project/pom.xml
index 318573a..dee79f7 100644
--- a/hadoop-project/pom.xml
+++ b/hadoop-project/pom.xml
@@ -996,7 +996,7 @@
<dependency>
<groupId>com.microsoft.azure</groupId>
<artifactId>azure-storage</artifactId>
- <version>2.2.0</version>
+ <version>4.2.0</version>
</dependency>
<dependency>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b43de800/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/BlockBlobAppendStream.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/BlockBlobAppendStream.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/BlockBlobAppendStream.java
index d1ec8df..e419a3b 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/BlockBlobAppendStream.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/BlockBlobAppendStream.java
@@ -28,6 +28,7 @@ import java.util.Calendar;
import java.util.HashMap;
import java.util.Locale;
import java.util.List;
+import java.util.UUID;
import java.util.Random;
import java.util.TimeZone;
import java.util.concurrent.LinkedBlockingQueue;
@@ -98,6 +99,12 @@ public class BlockBlobAppendStream extends OutputStream {
*/
private long nextBlockCount = UNSET_BLOCKS_COUNT;
+ /**
+ * Variable to hold the block id prefix to be used for azure
+ * storage blocks from azure-storage-java sdk version 4.2.0 onwards
+ */
+ private String blockIdPrefix = null;
+
private final Random sequenceGenerator = new Random();
/**
@@ -180,7 +187,8 @@ public class BlockBlobAppendStream extends OutputStream {
this.key = aKey;
this.bufferSize = bufferSize;
this.threadSequenceNumber = new AtomicInteger(0);
- setBlocksCount();
+ this.blockIdPrefix = null;
+ setBlocksCountAndBlockIdPrefix();
this.outBuffer = new ByteArrayOutputStream(bufferSize);
this.uncommittedBlockEntries = new ArrayList<BlockEntry>();
@@ -433,22 +441,41 @@ public class BlockBlobAppendStream extends OutputStream {
* Helper method used to generate the blockIDs. The algorithm used is similar to the Azure
* storage SDK.
*/
- private void setBlocksCount() throws IOException {
- try {
+ private void setBlocksCountAndBlockIdPrefix() throws IOException {
- if (nextBlockCount == UNSET_BLOCKS_COUNT) {
+ try {
- nextBlockCount = (long) (sequenceGenerator.nextInt(Integer.MAX_VALUE))
- + sequenceGenerator.nextInt(Integer.MAX_VALUE - MAX_BLOCK_COUNT);
+ if (nextBlockCount == UNSET_BLOCKS_COUNT && blockIdPrefix==null) {
List<BlockEntry> blockEntries =
blob.downloadBlockList(BlockListingFilter.COMMITTED, new BlobRequestOptions(), opContext);
- nextBlockCount += blockEntries.size();
+ String blockZeroBlockId = (blockEntries.size() > 0) ? blockEntries.get(0).getId() : "";
+ String prefix = UUID.randomUUID().toString() + "-";
+ String sampleNewerVersionBlockId = generateNewerVersionBlockId(prefix, 0);
+
+ if (blockEntries.size() > 0 && blockZeroBlockId.length() < sampleNewerVersionBlockId.length()) {
+
+ // If blob has already been created with 2.2.0, append subsequent blocks with older version (2.2.0) blockId
+ // compute nextBlockCount, the way it was done before; and don't use blockIdPrefix
+ this.blockIdPrefix = "";
+ nextBlockCount = (long) (sequenceGenerator.nextInt(Integer.MAX_VALUE))
+ + sequenceGenerator.nextInt(Integer.MAX_VALUE - MAX_BLOCK_COUNT);
+ nextBlockCount += blockEntries.size();
+
+ } else {
+
+ // If there are no existing blocks, create the first block with newer version (4.2.0) blockId
+ // If blob has already been created with 4.2.0, append subsequent blocks with newer version (4.2.0) blockId
+ this.blockIdPrefix = prefix;
+ nextBlockCount = blockEntries.size();
+
+ }
}
+
} catch (StorageException ex) {
- LOG.debug("Encountered storage exception during setting next Block Count."
+ LOG.debug("Encountered storage exception during setting next Block Count and BlockId prefix."
+ " StorageException : {} ErrorCode : {}", ex, ex.getErrorCode());
throw new IOException(ex);
}
@@ -465,7 +492,40 @@ public class BlockBlobAppendStream extends OutputStream {
throw new IOException("Append Stream in invalid state. nextBlockCount not set correctly");
}
- byte[] blockIdInBytes = getBytesFromLong(nextBlockCount);
+ if (this.blockIdPrefix == null) {
+ throw new IOException("Append Stream in invalid state. blockIdPrefix not set correctly");
+ }
+
+ if (!this.blockIdPrefix.equals("")) {
+
+ return generateNewerVersionBlockId(this.blockIdPrefix, nextBlockCount++);
+
+ } else {
+
+ return generateOlderVersionBlockId(nextBlockCount++);
+
+ }
+
+ }
+
+ /**
+ * Helper method that generates an older (2.2.0) version blockId
+ * @return String representing the block ID generated.
+ */
+ private String generateOlderVersionBlockId(long id) {
+
+ byte[] blockIdInBytes = getBytesFromLong(id);
+ return new String(Base64.encodeBase64(blockIdInBytes), StandardCharsets.UTF_8);
+ }
+
+ /**
+ * Helper method that generates an newer (4.2.0) version blockId
+ * @return String representing the block ID generated.
+ */
+ private String generateNewerVersionBlockId(String prefix, long id) {
+
+ String blockIdSuffix = String.format("%06d", id);
+ byte[] blockIdInBytes = (prefix + blockIdSuffix).getBytes(StandardCharsets.UTF_8);
return new String(Base64.encodeBase64(blockIdInBytes), StandardCharsets.UTF_8);
}
@@ -481,28 +541,33 @@ public class BlockBlobAppendStream extends OutputStream {
* @return A byte array that represents the data of the specified <code>long</code> value.
*/
private static byte[] getBytesFromLong(final long value) {
- final byte[] tempArray = new byte[8];
- for (int m = 0; m < 8; m++) {
- tempArray[7 - m] = (byte) ((value >> (8 * m)) & 0xFF);
- }
+ final byte[] tempArray = new byte[8];
- return tempArray;
+ for (int m = 0; m < 8; m++) {
+ tempArray[7 - m] = (byte) ((value >> (8 * m)) & 0xFF);
+ }
+
+ return tempArray;
}
+
/**
* Helper method that creates a thread to upload a block to azure storage.
* @param payload
* @throws IOException
*/
- private synchronized void uploadBlockToStorage(byte[] payload) throws IOException {
+ private synchronized void uploadBlockToStorage(byte[] payload)
+ throws IOException {
// upload payload to azure storage
- nextBlockCount++;
String blockId = generateBlockId();
+
// Since uploads of the Azure storage are done in parallel threads, we go ahead
// add the blockId in the uncommitted list. If the upload of the block fails
// we don't commit the blockIds.
- uncommittedBlockEntries.add(new BlockEntry(blockId));
+ BlockEntry blockEntry = new BlockEntry(blockId);
+ blockEntry.setSize(payload.length);
+ uncommittedBlockEntries.add(blockEntry);
ioThreadPool.execute(new WriteRequest(payload, blockId));
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b43de800/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/SendRequestIntercept.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/SendRequestIntercept.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/SendRequestIntercept.java
index 4d564d5..f86f392 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/SendRequestIntercept.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/SendRequestIntercept.java
@@ -147,7 +147,7 @@ public final class SendRequestIntercept extends StorageEvent<SendingRequestEvent
try {
// Sign the request. GET's have no payload so the content length is
// zero.
- StorageCredentialsHelper.signBlobAndQueueRequest(getCredentials(),
+ StorageCredentialsHelper.signBlobQueueAndFileRequest(getCredentials(),
urlConnection, -1L, getOperationContext());
} catch (InvalidKeyException e) {
// Log invalid key exception to track signing error before the send
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b43de800/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/StorageInterfaceImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/StorageInterfaceImpl.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/StorageInterfaceImpl.java
index 298f3aa..367cd04 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/StorageInterfaceImpl.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/StorageInterfaceImpl.java
@@ -404,7 +404,7 @@ class StorageInterfaceImpl extends StorageInterface {
public void startCopyFromBlob(CloudBlobWrapper sourceBlob, BlobRequestOptions options,
OperationContext opContext)
throws StorageException, URISyntaxException {
- getBlob().startCopyFromBlob(((CloudBlobWrapperImpl) sourceBlob).blob,
+ getBlob().startCopy(sourceBlob.getBlob().getQualifiedUri(),
null, null, options, opContext);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org