User Agent expect

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

@@ -807,7 +807,7 @@ public void initialize(URI uri, Configuration conf, AzureFileSystemInstrumentati
     LOG.debug("Page blob directories:  {}", setToString(pageBlobDirs));
 
     // User-agent
-    userAgentId = "wasbdriverV2.4";
+    userAgentId = "wasbdriverV2.5";
 
     // Extract the directories that should contain block blobs with compaction
     blockBlobWithCompationDirs = getDirectorySet(

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

@@ -862,7 +862,7 @@ public int getAnalysisPeriod() {
   }
 
   public String getCustomUserAgentPrefix() {
-    return "abfsdriverV2.4";
+    return "abfsdriverV2.5";
   }
 
   public String getClusterName() {

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

@@ -35,6 +35,7 @@ public enum Mode {
   private final boolean isAppendBlob;
   private final String leaseId;
   private boolean isExpectHeaderEnabled;
+  private boolean isRetryDueToExpect;
 
   public AppendRequestParameters(final long position,
       final int offset,
@@ -50,6 +51,7 @@ public AppendRequestParameters(final long position,
     this.isAppendBlob = isAppendBlob;
     this.leaseId = leaseId;
     this.isExpectHeaderEnabled = isExpectHeaderEnabled;
+    this.isRetryDueToExpect = false;
   }
 
   public long getPosition() {
@@ -80,7 +82,15 @@ public boolean isExpectHeaderEnabled() {
     return isExpectHeaderEnabled;
   }
 
+  public boolean isRetryDueToExpect() {
+    return isRetryDueToExpect;
+  }
+
   public void setExpectHeaderEnabled(boolean expectHeaderEnabled) {
     isExpectHeaderEnabled = expectHeaderEnabled;
   }
+
+  public void setRetryDueToExpect(boolean retryDueToExpect) {
+    isRetryDueToExpect = retryDueToExpect;
+  }
 }

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

@@ -87,6 +87,7 @@
  */
 public class AbfsClient implements Closeable {
   public static final Logger LOG = LoggerFactory.getLogger(AbfsClient.class);
+  public static final String HUNDRED_CONTINUE_USER_AGENT = SINGLE_WHITE_SPACE + HUNDRED_CONTINUE + SEMICOLON;
 
   private final URL baseUrl;
   private final SharedKeyCredentials sharedKeyCredentials;
@@ -783,6 +784,12 @@ public AbfsRestOperation append(final String path, final byte[] buffer,
         abfsUriQueryBuilder.addQuery(QUERY_PARAM_CLOSE, TRUE);
       }
     }
+    if (reqParams.isRetryDueToExpect()) {
+      String userAgentRetry = userAgent;
+      userAgentRetry = userAgentRetry.replace(HUNDRED_CONTINUE_USER_AGENT, EMPTY_STRING);
+      requestHeaders.removeIf(header -> header.getName().equalsIgnoreCase(USER_AGENT));
+      requestHeaders.add(new AbfsHttpHeader(USER_AGENT, userAgentRetry));
+    }
 
     // AbfsInputStream/AbfsOutputStream reuse SAS tokens for better performance
     String sasTokenForReuse = appendSASTokenToQuery(path, SASTokenProvider.WRITE_OPERATION,
@@ -816,6 +823,7 @@ public AbfsRestOperation append(final String path, final byte[] buffer,
       if (checkUserError(responseStatusCode) && reqParams.isExpectHeaderEnabled()) {
         LOG.debug("User error, retrying without 100 continue enabled for the given path {}", path);
         reqParams.setExpectHeaderEnabled(false);
+        reqParams.setRetryDueToExpect(true);
         return this.append(path, buffer, reqParams, cachedSasToken,
             tracingContext);
       }
@@ -873,6 +881,13 @@ public AbfsRestOperation append(final String blockId, final String path, final b
     requestHeaders.add(new AbfsHttpHeader(CONTENT_LENGTH, String.valueOf(buffer.length)));
     requestHeaders.add(new AbfsHttpHeader(IF_MATCH, eTag));
 
+    if (reqParams.isRetryDueToExpect()) {
+      String userAgentRetry = userAgent;
+      userAgentRetry = userAgentRetry.replace(HUNDRED_CONTINUE_USER_AGENT, EMPTY_STRING);
+      requestHeaders.removeIf(header -> header.getName().equalsIgnoreCase(USER_AGENT));
+      requestHeaders.add(new AbfsHttpHeader(USER_AGENT, userAgentRetry));
+    }
+
     // AbfsInputStream/AbfsOutputStream reuse SAS tokens for better performance
     String sasTokenForReuse = appendSASTokenToQuery(path, SASTokenProvider.WRITE_OPERATION,
             abfsUriQueryBuilder, cachedSasToken);
@@ -896,6 +911,7 @@ public AbfsRestOperation append(final String blockId, final String path, final b
       if (checkUserErrorBlob(responseStatusCode) && reqParams.isExpectHeaderEnabled()) {
         LOG.debug("User error, retrying without 100 continue enabled for the given path {}", path);
         reqParams.setExpectHeaderEnabled(false);
+        reqParams.setRetryDueToExpect(true);
         return this.append(blockId, path, buffer, reqParams, cachedSasToken,
                 tracingContext, eTag);
       }
@@ -1955,6 +1971,12 @@ String initializeUserAgent(final AbfsConfiguration abfsConfiguration,
     appendIfNotEmpty(sb,
         ExtensionHelper.getUserAgentSuffix(tokenProvider, EMPTY_STRING), true);
 
+    if (abfsConfiguration.isExpectHeaderEnabled()) {
+      sb.append(SINGLE_WHITE_SPACE);
+      sb.append(HUNDRED_CONTINUE);
+      sb.append(SEMICOLON);
+    }
+
     sb.append(SINGLE_WHITE_SPACE);
     sb.append(abfsConfiguration.getClusterName());
     sb.append(FORWARD_SLASH);

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

@@ -87,6 +87,7 @@ public final class ITestAbfsClient extends AbstractAbfsIntegrationTest {
 
   private static final String ACCOUNT_NAME = "bogusAccountName.dfs.core.windows.net";
   private static final String FS_AZURE_USER_AGENT_PREFIX = "Partner Service";
+  private static final String HUNDRED_CONTINUE_USER_AGENT = SINGLE_WHITE_SPACE + HUNDRED_CONTINUE + SEMICOLON;
   private static final String TEST_PATH = "/testfile";
   public static final int REDUCED_RETRY_COUNT = 2;
   public static final int REDUCED_BACKOFF_INTERVAL = 100;
@@ -198,6 +199,33 @@ public void verifyUserAgentPrefix()
       .doesNotContain(FS_AZURE_USER_AGENT_PREFIX);
   }
 
+  @Test
+  public void verifyUserAgentExpectHeader()
+          throws IOException, IllegalAccessException {
+    final Configuration configuration = new Configuration();
+    configuration.addResource(TEST_CONFIGURATION_FILE_NAME);
+    configuration.set(ConfigurationKeys.FS_AZURE_USER_AGENT_PREFIX_KEY, FS_AZURE_USER_AGENT_PREFIX);
+    configuration.setBoolean(ConfigurationKeys.FS_AZURE_ACCOUNT_IS_EXPECT_HEADER_ENABLED, true);
+    AbfsConfiguration abfsConfiguration = new AbfsConfiguration(configuration,
+            ACCOUNT_NAME);
+    String userAgentStr = getUserAgentString(abfsConfiguration, false);
+
+    verifybBasicInfo(userAgentStr);
+    Assertions.assertThat(userAgentStr)
+            .describedAs("User-Agent string should contain " + HUNDRED_CONTINUE_USER_AGENT)
+            .contains(HUNDRED_CONTINUE_USER_AGENT);
+
+    configuration.setBoolean(ConfigurationKeys.FS_AZURE_ACCOUNT_IS_EXPECT_HEADER_ENABLED, false);
+    abfsConfiguration = new AbfsConfiguration(configuration,
+            ACCOUNT_NAME);
+    userAgentStr = getUserAgentString(abfsConfiguration, false);
+
+    verifybBasicInfo(userAgentStr);
+    Assertions.assertThat(userAgentStr)
+            .describedAs("User-Agent string should not contain " + HUNDRED_CONTINUE_USER_AGENT)
+            .doesNotContain(HUNDRED_CONTINUE_USER_AGENT);
+  }
+
   @Test
   public void verifyUserAgentWithoutSSLProvider() throws Exception {
     final Configuration configuration = new Configuration();