HADOOP-11959. WASB should configure client side socket timeout in storage client blob request options. Contributed by Ivan Mitic.

(cherry picked from commit 94e7d49a6d)

Conflicts:
	hadoop-project/pom.xml

hadoop-common-project/hadoop-common/CHANGES.txt

@@ -312,6 +312,9 @@ Release 2.8.0 - UNRELEASED
     HADOOP-11930. test-patch in offline mode should tell maven to be in
     offline mode (Sean Busbey via aw)
 
+    HADOOP-11959. WASB should configure client side socket timeout in storage
+    client blob request options. (Ivan Mitic via cnauroth)
+
 Release 2.7.1 - UNRELEASED
 
   INCOMPATIBLE CHANGES

hadoop-project/pom.xml

@@ -962,7 +962,7 @@
       <dependency>
         <groupId>com.microsoft.azure</groupId>
         <artifactId>azure-storage</artifactId>
-        <version>2.0.0</version>
+        <version>2.2.0</version>
     </dependency>
       
       <dependency>

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/AzureNativeFileSystemStore.java

@@ -2434,15 +2434,6 @@ public class AzureNativeFileSystemStore implements NativeFileSystemStore {
       //
       CloudBlobWrapper dstBlob = getBlobReference(dstKey);
 
-      // TODO: Remove at the time when we move to Azure Java SDK 1.2+.
-      // This is the workaround provided by Azure Java SDK team to
-      // mitigate the issue with un-encoded x-ms-copy-source HTTP
-      // request header. Azure sdk version before 1.2+ does not encode this
-      // header what causes all URIs that have special (category "other")
-      // characters in the URI not to work with startCopyFromBlob when
-      // specified as source (requests fail with HTTP 403).
-      URI srcUri = new URI(srcBlob.getUri().toASCIIString());
-
       // Rename the source blob to the destination blob by copying it to
       // the destination blob then deleting it.
       //
@@ -2451,7 +2442,7 @@ public class AzureNativeFileSystemStore implements NativeFileSystemStore {
       // a more intensive exponential retry policy when the cluster is getting 
       // throttled.
       try {
-        dstBlob.startCopyFromBlob(srcUri, null, getInstrumentedContext());
+        dstBlob.startCopyFromBlob(srcBlob, null, getInstrumentedContext());
       } catch (StorageException se) {
         if (se.getErrorCode().equals(
 		  StorageErrorCode.SERVER_BUSY.toString())) {
@@ -2475,7 +2466,7 @@ public class AzureNativeFileSystemStore implements NativeFileSystemStore {
           options.setRetryPolicyFactory(new RetryExponentialRetry(
             copyBlobMinBackoff, copyBlobDeltaBackoff, copyBlobMaxBackoff, 
 			copyBlobMaxRetries));
-          dstBlob.startCopyFromBlob(srcUri, options, getInstrumentedContext());
+          dstBlob.startCopyFromBlob(srcBlob, options, getInstrumentedContext());
         } else {
           throw se;
         }

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/StorageInterface.java

@@ -381,8 +381,8 @@ abstract class StorageInterface {
      * Copies an existing blob's contents, properties, and metadata to this instance of the <code>CloudBlob</code>
      * class, using the specified operation context.
      *
-     * @param source
+     * @param sourceBlob
-     *            A <code>java.net.URI</code> The URI of a source blob.
+     *            A <code>CloudBlob</code> object that represents the source blob to copy.
      * @param options
      *            A {@link BlobRequestOptions} object that specifies any additional options for the request. Specifying
      *            <code>null</code> will use the default request options from the associated service client (
@@ -397,7 +397,7 @@ abstract class StorageInterface {
      * @throws URISyntaxException
      *
      */
-    public abstract void startCopyFromBlob(URI source,
+    public abstract void startCopyFromBlob(CloudBlobWrapper sourceBlob,
         BlobRequestOptions options, OperationContext opContext)
         throws StorageException, URISyntaxException;
     

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/StorageInterfaceImpl.java

@@ -393,10 +393,10 @@ class StorageInterfaceImpl extends StorageInterface {
     }
 
     @Override
-    public void startCopyFromBlob(URI source, BlobRequestOptions options,
+    public void startCopyFromBlob(CloudBlobWrapper sourceBlob, BlobRequestOptions options,
         OperationContext opContext)
             throws StorageException, URISyntaxException {
-      getBlob().startCopyFromBlob(source,
+      getBlob().startCopyFromBlob(((CloudBlobWrapperImpl)sourceBlob).blob,
           null, null, options, opContext);
     }
 

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/MockStorageInterface.java

@@ -429,9 +429,9 @@ public class MockStorageInterface extends StorageInterface {
     }
 
     @Override
-    public void startCopyFromBlob(URI source, BlobRequestOptions options,
+    public void startCopyFromBlob(CloudBlobWrapper sourceBlob, BlobRequestOptions options,
         OperationContext opContext) throws StorageException, URISyntaxException {
-      backingStore.copy(convertUriToDecodedString(source), convertUriToDecodedString(uri));
+      backingStore.copy(convertUriToDecodedString(sourceBlob.getUri()), convertUriToDecodedString(uri));
       //TODO: set the backingStore.properties.CopyState and
       //      update azureNativeFileSystemStore.waitForCopyToComplete
     }

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestAzureFileSystemErrorConditions.java

@@ -205,6 +205,7 @@ public class TestAzureFileSystemErrorConditions {
         @Override
         public boolean isTargetConnection(HttpURLConnection connection) {
           return connection.getRequestMethod().equals("PUT")
+              && connection.getURL().getQuery() != null
               && connection.getURL().getQuery().contains("blocklist");
         }
       });

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestBlobDataValidation.java

@@ -191,6 +191,7 @@ public class TestBlobDataValidation {
 
     private static boolean isPutBlock(HttpURLConnection connection) {
       return connection.getRequestMethod().equals("PUT")
+          && connection.getURL().getQuery() != null
           && connection.getURL().getQuery().contains("blockid");
     }