HADOOP-18146: ABFS: Added changes for expect hundred continue header

hadoop-tools/hadoop-azure/src/config/checkstyle-suppressions.xml

@@ -51,4 +51,6 @@
     <!-- allow tests to use _ for ordering. -->
     <suppress checks="MethodName"
               files="org[\\/]apache[\\/]hadoop[\\/]fs[\\/]azurebfs[\\/]commit[\\/]ITestAbfsTerasort.java"/>
+     <suppress checks="ParameterNumber"
+       files="org[\\/]apache[\\/]hadoop[\\/]fs[\\/]azurebfs[\\/]services[\\/]TestAbfsOutputStream.java"/>
 </suppressions>

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

@@ -117,6 +117,11 @@ public class AbfsConfiguration{
       DefaultValue = DEFAULT_OPTIMIZE_FOOTER_READ)
   private boolean optimizeFooterRead;
 
+  @BooleanConfigurationValidatorAnnotation(
+          ConfigurationKey = FS_AZURE_ACCOUNT_IS_EXPECT_HEADER_ENABLED,
+          DefaultValue = DEFAULT_FS_AZURE_ACCOUNT_IS_EXPECT_HEADER_ENABLED)
+  private boolean isExpectHeaderEnabled;
+
   @IntegerConfigurationValidatorAnnotation(ConfigurationKey = AZURE_READ_BUFFER_SIZE,
       MinValue = MIN_BUFFER_SIZE,
       MaxValue = MAX_BUFFER_SIZE,
@@ -689,6 +694,10 @@ public String getAppendBlobDirs() {
     return this.azureAppendBlobDirs;
   }
 
+  public boolean isExpectHeaderEnabled() {
+    return this.isExpectHeaderEnabled;
+  }
+
   public String getAzureInfiniteLeaseDirs() {
     return this.azureInfiniteLeaseDirs;
   }

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

@@ -694,6 +694,7 @@ private AbfsOutputStreamContext populateAbfsOutputStreamContext(
     return new AbfsOutputStreamContext(abfsConfiguration.getSasTokenRenewPeriodForStreamsInSeconds())
             .withWriteBufferSize(bufferSize)
             .enableFlush(abfsConfiguration.isFlushEnabled())
+            .enableExpectHeader(abfsConfiguration.isExpectHeaderEnabled())
             .enableSmallWriteOptimization(abfsConfiguration.isSmallWriteOptimizationEnabled())
             .disableOutputStreamFlush(abfsConfiguration.isOutputStreamFlushDisabled())
             .withStreamStatistics(new AbfsOutputStreamStatisticsImpl())

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/AbfsHttpConstants.java

@@ -103,6 +103,7 @@ public final class AbfsHttpConstants {
   public static final String DEFAULT_SCOPE = "default:";
   public static final String PERMISSION_FORMAT = "%04d";
   public static final String SUPER_USER = "$superuser";
+  public static final String HUNDRED_CONTINUE = "100-continue";
 
   public static final char CHAR_FORWARD_SLASH = '/';
   public static final char CHAR_EXCLAMATION_POINT = '!';

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/ConfigurationKeys.java

@@ -35,6 +35,7 @@ public final class ConfigurationKeys {
    * path to determine HNS status.
    */
   public static final String FS_AZURE_ACCOUNT_IS_HNS_ENABLED = "fs.azure.account.hns.enabled";
+  public static final String FS_AZURE_ACCOUNT_IS_EXPECT_HEADER_ENABLED = "fs.azure.account.expect.header.enabled";
   public static final String FS_AZURE_ACCOUNT_KEY_PROPERTY_NAME = "fs.azure.account.key";
   public static final String FS_AZURE_ACCOUNT_KEY_PROPERTY_NAME_REGX = "fs\\.azure\\.account\\.key\\.(.*)";
   public static final String FS_AZURE_SECURE_MODE = "fs.azure.secure.mode";

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/FileSystemConfigurations.java

@@ -32,7 +32,7 @@
 public final class FileSystemConfigurations {
 
   public static final String DEFAULT_FS_AZURE_ACCOUNT_IS_HNS_ENABLED = "";
-
+  public static final boolean DEFAULT_FS_AZURE_ACCOUNT_IS_EXPECT_HEADER_ENABLED = true;
   public static final String USER_HOME_DIRECTORY_PREFIX = "/user";
 
   private static final int SIXTY_SECONDS = 60 * 1000;

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/HttpHeaderConfigurations.java

@@ -60,6 +60,7 @@ public final class HttpHeaderConfigurations {
   public static final String X_MS_UMASK = "x-ms-umask";
   public static final String X_MS_NAMESPACE_ENABLED = "x-ms-namespace-enabled";
   public static final String X_MS_ABFS_CLIENT_LATENCY = "x-ms-abfs-client-latency";
+  public static final String EXPECT = "Expect";
   public static final String X_MS_ENCRYPTION_KEY = "x-ms-encryption-key";
   public static final String X_MS_ENCRYPTION_KEY_SHA256 = "x-ms-encryption-key-sha256";
   public static final String X_MS_ENCRYPTION_ALGORITHM = "x-ms-encryption-algorithm";

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/contracts/exceptions/InvalidAbfsRestOperationException.java

@@ -30,14 +30,24 @@
 @InterfaceAudience.Public
 @InterfaceStability.Evolving
 public class InvalidAbfsRestOperationException extends AbfsRestOperationException {
-  public InvalidAbfsRestOperationException(
-      final Exception innerException) {
-    super(
-        AzureServiceErrorCode.UNKNOWN.getStatusCode(),
-        AzureServiceErrorCode.UNKNOWN.getErrorCode(),
-        innerException != null
-            ? innerException.toString()
-            : "InvalidAbfsRestOperationException",
-        innerException);
-  }
+    public InvalidAbfsRestOperationException(
+            final Exception innerException) {
+        super(
+                AzureServiceErrorCode.UNKNOWN.getStatusCode(),
+                AzureServiceErrorCode.UNKNOWN.getErrorCode(),
+                innerException != null
+                        ? innerException.toString()
+                        : "InvalidAbfsRestOperationException",
+                innerException);
+    }
+
+    public InvalidAbfsRestOperationException(final Exception innerException, int retryCount) {
+        super(
+                AzureServiceErrorCode.UNKNOWN.getStatusCode(),
+                AzureServiceErrorCode.UNKNOWN.getErrorCode(),
+                innerException != null
+                        ? innerException.toString()
+                        : "InvalidAbfsRestOperationException" + "RetryCount: " + String.valueOf(retryCount),
+                innerException);
+    }
 }

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/contracts/services/AppendRequestParameters.java

@@ -34,19 +34,21 @@ public enum Mode {
   private final Mode mode;
   private final boolean isAppendBlob;
   private final String leaseId;
+  private boolean isExpectHeaderEnabled;
 
   public AppendRequestParameters(final long position,
       final int offset,
       final int length,
       final Mode mode,
       final boolean isAppendBlob,
-      final String leaseId) {
+      final String leaseId, boolean isExpectHeaderEnabled) {
     this.position = position;
     this.offset = offset;
     this.length = length;
     this.mode = mode;
     this.isAppendBlob = isAppendBlob;
     this.leaseId = leaseId;
+    this.isExpectHeaderEnabled = isExpectHeaderEnabled;
   }
 
   public long getPosition() {
@@ -72,4 +74,12 @@ public boolean isAppendBlob() {
   public String getLeaseId() {
     return this.leaseId;
   }
+
+  public boolean getIsExpectHeaderEnabled() {
+    return isExpectHeaderEnabled;
+  }
+
+  public void setExpectHeaderEnabled(boolean expectHeaderEnabled) {
+    isExpectHeaderEnabled = expectHeaderEnabled;
+  }
 }

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsClient.java

@@ -38,6 +38,7 @@
 import java.util.concurrent.TimeUnit;
 
 import org.apache.hadoop.classification.VisibleForTesting;
+import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsRestOperationException;
 import org.apache.hadoop.fs.store.LogExactlyOnce;
 import org.apache.hadoop.util.Preconditions;
 import org.apache.hadoop.thirdparty.com.google.common.base.Strings;
@@ -652,6 +653,9 @@ public AbfsRestOperation append(final String path, final byte[] buffer,
     addCustomerProvidedKeyHeaders(requestHeaders);
     // JDK7 does not support PATCH, so to workaround the issue we will use
     // PUT and specify the real method in the X-Http-Method-Override header.
+    if (reqParams.getIsExpectHeaderEnabled()) {
+      requestHeaders.add(new AbfsHttpHeader(EXPECT, HUNDRED_CONTINUE));
+    }
     requestHeaders.add(new AbfsHttpHeader(X_HTTP_METHOD_OVERRIDE,
         HTTP_METHOD_PATCH));
     if (reqParams.getLeaseId() != null) {
@@ -688,6 +692,17 @@ public AbfsRestOperation append(final String path, final byte[] buffer,
     try {
       op.execute(tracingContext);
     } catch (AzureBlobFileSystemException e) {
+      /*
+        If the http response code indicates a user error we retry the same append request with expect header disabled.
+        When "100-continue" header is enabled but a non Http 100 response comes, JDK fails to provide all response headers.
+        This handling is to avoid breaking of backward compatibility if someone has taken dependency on the exception message,
+        which is created using the error string present in the response header.
+       */
+      if (checkUserError(e) && reqParams.getIsExpectHeaderEnabled()) {
+        reqParams.setExpectHeaderEnabled(false);
+        return this.append(path, buffer, reqParams, cachedSasToken,
+            tracingContext);
+      }
       // If we have no HTTP response, throw the original exception.
       if (!op.hasResult()) {
         throw e;
@@ -714,6 +729,18 @@ && appendSuccessCheckOp(op, path,
     return op;
   }
 
+  /**
+   * Returns true if the status code lies in the range of user error
+   * @param e Exception caught
+   * @return True or False
+   */
+  private boolean checkUserError(AzureBlobFileSystemException e) {
+    return ((AbfsRestOperationException) e).getStatusCode()
+        >= HttpURLConnection.HTTP_BAD_REQUEST
+        && ((AbfsRestOperationException) e).getStatusCode()
+        < HttpURLConnection.HTTP_INTERNAL_ERROR;
+  }
+
   // For AppendBlob its possible that the append succeeded in the backend but the request failed.
   // However a retry would fail with an InvalidQueryParameterValue
   // (as the current offset would be unacceptable).
@@ -1273,6 +1300,14 @@ protected AbfsCounters getAbfsCounters() {
     return abfsCounters;
   }
 
+  /**
+   * Getter for abfsConfiguration from AbfsClient.
+   * @return AbfsConfiguration instance
+   */
+  protected AbfsConfiguration getAbfsConfiguration() {
+    return abfsConfiguration;
+  }
+
   public int getNumLeaseThreads() {
     return abfsConfiguration.getNumLeaseThreads();
   }

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsHttpOperation.java

@@ -320,6 +320,8 @@ public void sendRequest(byte[] buffer, int offset, int length) throws IOExceptio
       // accompanying statusCode
       this.bytesSent = length;
       outputStream.write(buffer, offset, length);
+    } catch (IOException e) {
+      this.bytesSent = length;
     } finally {
       if (this.isTraceEnabled) {
         this.sendRequestTimeMs = elapsedTimeMs(startTime);

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsOutputStream.java

@@ -80,6 +80,7 @@ public class AbfsOutputStream extends OutputStream implements Syncable,
   private boolean disableOutputStreamFlush;
   private boolean enableSmallWriteOptimization;
   private boolean isAppendBlob;
+  private boolean isExpectHeaderEnabled;
   private volatile IOException lastError;
 
   private long lastFlushOffset;
@@ -133,6 +134,7 @@ public AbfsOutputStream(AbfsOutputStreamContext abfsOutputStreamContext)
     this.position = abfsOutputStreamContext.getPosition();
     this.closed = false;
     this.supportFlush = abfsOutputStreamContext.isEnableFlush();
+    this.isExpectHeaderEnabled = abfsOutputStreamContext.isExpectHeaderEnabled();
     this.disableOutputStreamFlush = abfsOutputStreamContext
             .isDisableOutputStreamFlush();
     this.enableSmallWriteOptimization
@@ -327,7 +329,7 @@ private void uploadBlockAsync(DataBlocks.DataBlock blockToUpload,
              * leaseId - The AbfsLeaseId for this request.
              */
             AppendRequestParameters reqParams = new AppendRequestParameters(
-                offset, 0, bytesLength, mode, false, leaseId);
+                offset, 0, bytesLength, mode, false, leaseId, isExpectHeaderEnabled);
             AbfsRestOperation op =
                 client.append(path, blockUploadData.toByteArray(), reqParams,
                     cachedSasToken.get(), new TracingContext(tracingContext));
@@ -573,7 +575,7 @@ private void writeAppendBlobCurrentBufferToService() throws IOException {
     try (AbfsPerfInfo perfInfo = new AbfsPerfInfo(tracker,
         "writeCurrentBufferToService", "append")) {
       AppendRequestParameters reqParams = new AppendRequestParameters(offset, 0,
-          bytesLength, APPEND_MODE, true, leaseId);
+          bytesLength, APPEND_MODE, true, leaseId, isExpectHeaderEnabled);
       AbfsRestOperation op = client.append(path, uploadData.toByteArray(), reqParams,
           cachedSasToken.get(), new TracingContext(tracingContext));
       cachedSasToken.update(op.getSasToken());

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsOutputStreamContext.java

@@ -33,6 +33,8 @@ public class AbfsOutputStreamContext extends AbfsStreamContext {
 
   private boolean enableFlush;
 
+  private boolean enableExpectHeader;
+
   private boolean enableSmallWriteOptimization;
 
   private boolean disableOutputStreamFlush;
@@ -78,6 +80,11 @@ public AbfsOutputStreamContext enableFlush(final boolean enableFlush) {
     return this;
   }
 
+  public AbfsOutputStreamContext enableExpectHeader(final boolean enableExpectHeader) {
+    this.enableExpectHeader = enableExpectHeader;
+    return this;
+  }
+
   public AbfsOutputStreamContext enableSmallWriteOptimization(final boolean enableSmallWriteOptimization) {
     this.enableSmallWriteOptimization = enableSmallWriteOptimization;
     return this;
@@ -184,6 +191,10 @@ public boolean isEnableFlush() {
     return enableFlush;
   }
 
+  public boolean isExpectHeaderEnabled() {
+    return enableExpectHeader;
+  }
+
   public boolean isDisableOutputStreamFlush() {
     return disableOutputStreamFlush;
   }

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsRestOperation.java

@@ -303,7 +303,7 @@ private boolean executeHttpOperation(final int retryCount,
       LOG.warn("Unknown host name: {}. Retrying to resolve the host name...",
           hostname);
       if (!client.getRetryPolicy().shouldRetry(retryCount, -1)) {
-        throw new InvalidAbfsRestOperationException(ex);
+        throw new InvalidAbfsRestOperationException(ex, retryCount);
       }
       return false;
     } catch (IOException ex) {
@@ -312,7 +312,7 @@ private boolean executeHttpOperation(final int retryCount,
       }
 
       if (!client.getRetryPolicy().shouldRetry(retryCount, -1)) {
-        throw new InvalidAbfsRestOperationException(ex);
+        throw new InvalidAbfsRestOperationException(ex, retryCount);
       }
 
       return false;

hadoop-tools/hadoop-azure/src/site/markdown/abfs.md

@@ -767,6 +767,12 @@ Hflush() being the only documented API that can provide persistent data
 transfer, Flush() also attempting to persist buffered data will lead to
 performance issues.
 
+### <a name="100continueconfigoptions"></a> Hundred Continue Options
+
+`fs.azure.account.expect.header.enabled`: This configuration parameter is used
+to specify whether you wish to send a expect 100 continue header with each
+append request or not. It is configured to true by default.
+
 ### <a name="hnscheckconfigoptions"></a> HNS Check Options
 Config `fs.azure.account.hns.enabled` provides an option to specify whether
  the storage account is HNS enabled or not. In case the config is not provided,

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

@@ -203,7 +203,7 @@ public void testAppendWithCPK() throws Exception {
     //  Trying to append with correct CPK headers
     AppendRequestParameters appendRequestParameters =
         new AppendRequestParameters(
-        0, 0, 5, Mode.APPEND_MODE, false, null);
+        0, 0, 5, Mode.APPEND_MODE, false, null, true);
     byte[] buffer = getRandomBytesArray(5);
     AbfsClient abfsClient = fs.getAbfsClient();
     AbfsRestOperation abfsRestOperation = abfsClient
@@ -248,7 +248,7 @@ public void testAppendWithoutCPK() throws Exception {
     //  Trying to append without CPK headers
     AppendRequestParameters appendRequestParameters =
         new AppendRequestParameters(
-        0, 0, 5, Mode.APPEND_MODE, false, null);
+        0, 0, 5, Mode.APPEND_MODE, false, null, true);
     byte[] buffer = getRandomBytesArray(5);
     AbfsClient abfsClient = fs.getAbfsClient();
     AbfsRestOperation abfsRestOperation = abfsClient

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsClient.java

@@ -306,7 +306,7 @@ public static AbfsClient getMockAbfsClient(AbfsClient baseAbfsClientInstance,
     when(client.getAccessToken()).thenCallRealMethod();
     when(client.getSharedKeyCredentials()).thenCallRealMethod();
     when(client.createDefaultHeaders()).thenCallRealMethod();
-
+    when(client.getAbfsConfiguration()).thenReturn(abfsConfig);
     // override baseurl
     client = TestAbfsClient.setAbfsClientField(client, "abfsConfiguration",
         abfsConfig);