currentWeakRef = lookup(key);
+ // resolve it, after which if not null, we have a strong reference
+ V strongVal = resolve(currentWeakRef);
+ if (strongVal != null) {
+ // all good.
+ return strongVal;
+ }
+
+ // here, either currentWeakRef was null, or its reference was GC'd.
+ if (currentWeakRef != null) {
+ // garbage collection removed the reference.
+
+ // explicitly remove the weak ref from the map if it has not
+ // been updated by this point
+ // this is here just for completeness.
+ map.remove(key, currentWeakRef);
+
+ // log/report the loss.
+ noteLost(key);
+ }
+
+ // create a new value and add it to the map
+ return create(key);
+ }
+
+ /**
+ * Create a new instance under a key.
+ *
+ * The instance is created, added to the map and then the
+ * map value retrieved.
+ * This ensures that the reference returned is that in the map,
+ * even if there is more than one entry being created at the same time.
+ * If that race does occur, it will be logged the first time it happens
+ * for this specific map instance.
+ *
+ * HADOOP-18456 highlighted the risk of a concurrent GC resulting a null
+ * value being retrieved and so returned.
+ * To prevent this:
+ *
+ * - A strong reference is retained to the newly created instance
+ * in a local variable.
+ * - That variable is used after the resolution process, to ensure
+ * the JVM doesn't consider it "unreachable" and so eligible for GC.
+ * - A check is made for the resolved reference being null, and if so,
+ * the put() is repeated
+ *
+ * @param key key
+ * @return the created value
+ */
+ public V create(K key) {
+ entriesCreatedCount.incrementAndGet();
+ /*
+ Get a strong ref so even if a GC happens in this method the reference is not lost.
+ It is NOT enough to have a reference in a field, it MUST be used
+ so as to ensure the reference isn't optimized away prematurely.
+ "A reachable object is any object that can be accessed in any potential continuing
+ computation from any live thread."
+ */
+
+ final V strongRef = requireNonNull(factory.apply(key),
+ "factory returned a null instance");
+ V resolvedStrongRef;
+ do {
+ WeakReference newWeakRef = new WeakReference<>(strongRef);
+
+ // put it in the map
+ map.put(key, newWeakRef);
+
+ // get it back from the map
+ WeakReference retrievedWeakRef = map.get(key);
+ // resolve that reference, handling the situation where somehow it was removed from the map
+ // between the put() and the get()
+ resolvedStrongRef = resolve(retrievedWeakRef);
+ if (resolvedStrongRef == null) {
+ referenceLostDuringCreation.warn("reference to %s lost during creation", key);
+ noteLost(key);
+ }
+ } while (resolvedStrongRef == null);
+
+ // note if there was any change in the reference.
+ // as this forces strongRef to be kept in scope
+ if (strongRef != resolvedStrongRef) {
+ LOG.debug("Created instance for key {}: {} overwritten by {}",
+ key, strongRef, resolvedStrongRef);
+ }
+
+ return resolvedStrongRef;
+ }
+
+ /**
+ * Put a value under the key.
+ * A null value can be put, though on a get() call
+ * a new entry is generated
+ *
+ * @param key key
+ * @param value value
+ * @return any old non-null reference.
+ */
+ public V put(K key, V value) {
+ return resolve(map.put(key, new WeakReference<>(value)));
+ }
+
+ /**
+ * Remove any value under the key.
+ * @param key key
+ * @return any old non-null reference.
+ */
+ public V remove(K key) {
+ return resolve(map.remove(key));
+ }
+
+ /**
+ * Does the map have a valid reference for this object?
+ * no-side effects: there's no attempt to notify or cleanup
+ * if the reference is null.
+ * @param key key to look up
+ * @return true if there is a valid reference.
+ */
+ public boolean containsKey(K key) {
+ final WeakReference current = lookup(key);
+ return resolve(current) != null;
+ }
+
+ /**
+ * Given a possibly null weak reference, resolve
+ * its value.
+ * @param r reference to resolve
+ * @return the value or null
+ */
+ protected V resolve(WeakReference r) {
+ return r == null ? null : r.get();
+ }
+
+ /**
+ * Prune all null weak references, calling the referenceLost
+ * callback for each one.
+ *
+ * non-atomic and non-blocking.
+ * @return the number of entries pruned.
+ */
+ public int prune() {
+ int count = 0;
+ final Iterator>> it = map.entrySet().iterator();
+ while (it.hasNext()) {
+ final Map.Entry> next = it.next();
+ if (next.getValue().get() == null) {
+ it.remove();
+ count++;
+ noteLost(next.getKey());
+ }
+ }
+ return count;
+ }
+
+ /**
+ * Notify the reference lost callback.
+ * @param key key of lost reference
+ */
+ private void noteLost(final K key) {
+ // increment local counter
+ referenceLostCount.incrementAndGet();
+
+ // and call any notification function supplied in the constructor
+ if (referenceLost != null) {
+ referenceLost.accept(key);
+ }
+ }
+
+ /**
+ * Get count of references lost as detected
+ * during prune() or get() calls.
+ * @return count of references lost
+ */
+ public final long getReferenceLostCount() {
+ return referenceLostCount.get();
+ }
+
+ /**
+ * Get count of entries created on demand.
+ * @return count of entries created
+ */
+ public final long getEntriesCreatedCount() {
+ return entriesCreatedCount.get();
+ }
+}
+
diff --git a/hadoop-tools/hadoop-azure/src/site/markdown/abfs.md b/hadoop-tools/hadoop-azure/src/site/markdown/abfs.md
index dfb7f3f42a5cf9..ac77765f9e0b52 100644
--- a/hadoop-tools/hadoop-azure/src/site/markdown/abfs.md
+++ b/hadoop-tools/hadoop-azure/src/site/markdown/abfs.md
@@ -769,6 +769,26 @@ Hflush() being the only documented API that can provide persistent data
transfer, Flush() also attempting to persist buffered data will lead to
performance issues.
+<<<<<<< HEAD
+=======
+### 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. This flag configures
+the client to check with the Azure store before uploading a block of data from
+an output stream. This allows the client to throttle back gracefully -before
+actually attempting to upload the block. In experiments this provides
+significant throughput improvements under heavy load. For more information :
+- https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Expect
+
+
+### Account level throttling Options
+
+`fs.azure.account.operation.idle.timeout`: This value specifies the time after which the timer for the analyzer (read or
+write) should be paused until no new request is made again. The default value for the same is 60 seconds.
+
+>>>>>>> c88011c6046... HADOOP-18146: ABFS: Added changes for expect hundred continue header (#4039)
### 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,
@@ -874,6 +894,9 @@ when there are too many writes from the same process.
time. Effectively this will be the threadpool size within the
AbfsOutputStream instance. Set the value in between 1 to 8 both inclusive.
+`fs.azure.analysis.period`: The time after which sleep duration is recomputed after analyzing metrics. The default value
+for the same is 10 seconds.
+
`fs.azure.write.max.requests.to.queue`: To set the maximum write requests
that can be queued. Memory consumption of AbfsOutputStream instance can be
tuned with this config considering each queued request holds a buffer. Set
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/AbstractAbfsIntegrationTest.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/AbstractAbfsIntegrationTest.java
index 932b7638a48481..cca2a94add2bc5 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/AbstractAbfsIntegrationTest.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/AbstractAbfsIntegrationTest.java
@@ -25,6 +25,7 @@
import java.util.UUID;
import java.util.concurrent.Callable;
+import org.apache.hadoop.fs.azurebfs.oauth2.AccessTokenProvider;
import org.apache.hadoop.fs.azurebfs.services.AbfsClient;
import org.apache.hadoop.fs.azurebfs.services.PrefixMode;
import org.junit.After;
@@ -42,6 +43,7 @@
import org.apache.hadoop.fs.azurebfs.security.AbfsDelegationTokenManager;
import org.apache.hadoop.fs.azurebfs.services.AbfsOutputStream;
import org.apache.hadoop.fs.azurebfs.services.AuthType;
+import org.apache.hadoop.fs.azurebfs.services.ITestAbfsClient;
import org.apache.hadoop.fs.azure.AzureNativeFileSystemStore;
import org.apache.hadoop.fs.azure.NativeAzureFileSystem;
import org.apache.hadoop.fs.azure.metrics.AzureFileSystemInstrumentation;
@@ -73,7 +75,7 @@ public abstract class AbstractAbfsIntegrationTest extends
AbstractAbfsTestWithTimeout {
private static final Logger LOG =
- LoggerFactory.getLogger(AbstractAbfsIntegrationTest.class);
+ LoggerFactory.getLogger(AbstractAbfsIntegrationTest.class);
private boolean isIPAddress;
private NativeAzureFileSystem wasb;
@@ -110,7 +112,7 @@ protected AbstractAbfsIntegrationTest() throws Exception {
if (authType == AuthType.SharedKey) {
assumeTrue("Not set: " + FS_AZURE_ACCOUNT_KEY,
- abfsConfig.get(FS_AZURE_ACCOUNT_KEY) != null);
+ abfsConfig.get(FS_AZURE_ACCOUNT_KEY) != null);
// Update credentials
} else {
assumeTrue("Not set: " + FS_AZURE_ACCOUNT_TOKEN_PROVIDER_TYPE_PROPERTY_NAME,
@@ -243,6 +245,9 @@ public Hashtable call() throws Exception {
}
}
+ public AccessTokenProvider getAccessTokenProvider(final AzureBlobFileSystem fs) {
+ return ITestAbfsClient.getAccessTokenProvider(fs.getAbfsStore().getClient());
+ }
public void loadConfiguredFileSystem() throws Exception {
// disable auto-creation of filesystem
@@ -506,4 +511,30 @@ protected long assertAbfsStatistics(AbfsStatistic statistic,
(long) metricMap.get(statistic.getStatName()));
return expectedValue;
}
+
+ /**
+ * For creating directory with implicit parents. Doesn't change already explicit
+ * parents.
+ */
+ void createAzCopyDirectory(Path path) throws Exception {
+ AzcopyHelper azcopyHelper = new AzcopyHelper(
+ getAccountName(), getFileSystemName(), getFileSystem().getAbfsStore()
+ .getAbfsConfiguration()
+ .getRawConfiguration(), getFileSystem().getAbfsStore().getPrefixMode());
+ azcopyHelper.createFolderUsingAzcopy(
+ getFileSystem().makeQualified(path).toUri().getPath().substring(1));
+ }
+
+ /**
+ * For creating files with implicit parents. Doesn't change already explicit
+ * parents.
+ */
+ void createAzCopyFile(Path path) throws Exception {
+ AzcopyHelper azcopyHelper = new AzcopyHelper(getAccountName(),
+ getFileSystemName(), getFileSystem().getAbfsStore()
+ .getAbfsConfiguration()
+ .getRawConfiguration(), getFileSystem().getAbfsStore().getPrefixMode());
+ azcopyHelper.createFileUsingAzcopy(
+ getFileSystem().makeQualified(path).toUri().getPath().substring(1));
+ }
}
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/BlobDirectoryStateHelper.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/BlobDirectoryStateHelper.java
index e4fa38e365ec32..dad4551f65dbdb 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/BlobDirectoryStateHelper.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/BlobDirectoryStateHelper.java
@@ -18,7 +18,6 @@
package org.apache.hadoop.fs.azurebfs;
-import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.HttpURLConnection;
import java.util.List;
@@ -29,7 +28,6 @@
import org.mockito.Mockito;
-import org.apache.hadoop.fs.azurebfs.AzureBlobFileSystem;
import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsRestOperationException;
import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AzureBlobFileSystemException;
import org.apache.hadoop.fs.azurebfs.services.BlobProperty;
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAbfsRestOperationException.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAbfsRestOperationException.java
index 98ce66d4cbba6a..b45940f1549ca3 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAbfsRestOperationException.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAbfsRestOperationException.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.fs.azurebfs;
+import java.io.FileNotFoundException;
import java.io.IOException;
import org.assertj.core.api.Assertions;
@@ -31,6 +32,8 @@
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.azurebfs.services.PrefixMode;
+
import org.mockito.Mockito;
import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.DOT;
@@ -63,7 +66,12 @@ public void testAbfsRestOperationExceptionFormat() throws IOException {
Assert.assertEquals(4, errorFields.length);
// Check status message, status code, HTTP Request Type and URL.
- Assert.assertEquals("Operation failed: \"The specified path does not exist.\"", errorFields[0].trim());
+ if (fs.getAbfsStore().getPrefixMode() == PrefixMode.BLOB) {
+ Assert.assertEquals("Operation failed: \"The specified blob does not exist.\"", errorFields[0].trim());
+ }
+ else {
+ Assert.assertEquals("Operation failed: \"The specified path does not exist.\"", errorFields[0].trim());
+ }
Assert.assertEquals("404", errorFields[1].trim());
Assert.assertEquals("HEAD", errorFields[2].trim());
Assert.assertTrue(errorFields[3].trim().startsWith("http"));
@@ -79,7 +87,12 @@ public void testAbfsRestOperationExceptionFormat() throws IOException {
if (!getAbfsStore(fs).getAbfsConfiguration().enableAbfsListIterator()) {
Assert.assertEquals(6, errorFields.length);
// Check status message, status code, HTTP Request Type and URL.
- Assert.assertEquals("Operation failed: \"The specified path does not exist.\"", errorFields[0].trim());
+ if (fs.getAbfsStore().getPrefixMode() == PrefixMode.BLOB) {
+ Assert.assertEquals("Operation failed: \"The specified blob does not exist.\"", errorFields[0].trim());
+ }
+ else {
+ Assert.assertEquals("Operation failed: \"The specified path does not exist.\"", errorFields[0].trim());
+ }
Assert.assertEquals("404", errorFields[1].trim());
Assert.assertEquals("GET", errorFields[2].trim());
Assert.assertTrue(errorFields[3].trim().startsWith("http"));
@@ -90,7 +103,12 @@ public void testAbfsRestOperationExceptionFormat() throws IOException {
} else {
Assert.assertEquals(4, errorFields.length);
// Check status message, status code, HTTP Request Type and URL.
- Assert.assertEquals("Operation failed: \"The specified path does not exist.\"", errorFields[0].trim());
+ if (fs.getAbfsStore().getPrefixMode() == PrefixMode.BLOB) {
+ Assert.assertEquals("Operation failed: \"The specified blob does not exist.\"", errorFields[0].trim());
+ }
+ else {
+ Assert.assertEquals("Operation failed: \"The specified path does not exist.\"", errorFields[0].trim());
+ }
Assert.assertEquals("404", errorFields[1].trim());
Assert.assertEquals("HEAD", errorFields[2].trim());
Assert.assertTrue(errorFields[3].trim().startsWith("http"));
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemAppend.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemAppend.java
index fd4f8d4622e075..a6202f54378d4f 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemAppend.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemAppend.java
@@ -33,12 +33,13 @@
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsRestOperationException;
import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AzureBlobFileSystemException;
import org.apache.hadoop.fs.azurebfs.services.AbfsClient;
import org.apache.hadoop.fs.azurebfs.services.AbfsOutputStream;
import org.apache.hadoop.fs.azurebfs.services.OperativeEndpoint;
import org.apache.hadoop.fs.azurebfs.services.PrefixMode;
-import org.apache.hadoop.fs.azurebfs.services.TestAbfsClient;
+import org.apache.hadoop.fs.azurebfs.services.ITestAbfsClient;
import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
import org.junit.Assume;
import org.junit.Test;
@@ -242,6 +243,19 @@ public void testCreateEmptyBlob() throws IOException {
Mockito.any(TracingContext.class));
}
+ @Test
+ public void testAppendImplicitDirectoryAzcopy() throws Exception {
+ Assume.assumeTrue(getFileSystem().getAbfsStore().getPrefixMode() == PrefixMode.BLOB);
+ AzureBlobFileSystem fs = getFileSystem();
+ createAzCopyDirectory(new Path("/src"));
+ createAzCopyFile(new Path("/src/file"));
+ intercept(AbfsRestOperationException.class, () -> {
+ fs.getAbfsStore().getBlobProperty(new Path("/src"), Mockito.mock(
+ TracingContext.class));
+ });
+ intercept(FileNotFoundException.class, () -> fs.append(new Path("/src")));
+ }
+
/**
* Verify that no calls to getBlockList were made.
*/
@@ -252,7 +266,7 @@ public void testCreateNonEmptyBlob() throws IOException {
AzureBlobFileSystemStore store = Mockito.spy(fs.getAbfsStore());
Mockito.doReturn(store).when(fs).getAbfsStore();
AbfsClient client = store.getClient();
- AbfsClient testClient = Mockito.spy(TestAbfsClient.createTestClientFromCurrentContext(
+ AbfsClient testClient = Mockito.spy(ITestAbfsClient.createTestClientFromCurrentContext(
client,
fs.getAbfsStore().getAbfsConfiguration()));
store.setClient(testClient);
@@ -277,7 +291,7 @@ public void testValidateGetBlockList() throws Exception {
AzureBlobFileSystemStore store = Mockito.spy(fs.getAbfsStore());
Mockito.doReturn(store).when(fs).getAbfsStore();
AbfsClient client = store.getClient();
- AbfsClient testClient = Mockito.spy(TestAbfsClient.createTestClientFromCurrentContext(
+ AbfsClient testClient = Mockito.spy(ITestAbfsClient.createTestClientFromCurrentContext(
client,
fs.getAbfsStore().getAbfsConfiguration()));
store.setClient(testClient);
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemAttributes.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemAttributes.java
index beb7d0ebaaa8ea..e1eb315bc73bed 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemAttributes.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemAttributes.java
@@ -19,6 +19,8 @@
package org.apache.hadoop.fs.azurebfs;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
import java.util.EnumSet;
import org.junit.Assume;
@@ -26,8 +28,7 @@
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.XAttrSetFlag;
-import org.apache.hadoop.fs.azurebfs.constants.FSOperationType;
-import org.apache.hadoop.fs.azurebfs.utils.TracingHeaderValidator;
+import org.apache.hadoop.fs.azurebfs.services.PrefixMode;
import static org.apache.hadoop.test.LambdaTestUtils.intercept;
@@ -42,70 +43,218 @@ public ITestAzureBlobFileSystemAttributes() throws Exception {
super();
}
+ /**
+ * Test GetXAttr() and SetXAttr() with Unicode Attribute Values.
+ * DFS does not support Unicode characters in user-defined metadata properties.
+ * Blob Endpoint supports Unicode encoded in UTF_8 character encoding.
+ * @throws Exception
+ */
@Test
- public void testSetGetXAttr() throws Exception {
+ public void testGetSetXAttr() throws Exception {
AzureBlobFileSystem fs = getFileSystem();
- AbfsConfiguration conf = fs.getAbfsStore().getAbfsConfiguration();
- Assume.assumeTrue(getIsNamespaceEnabled(fs));
-
- byte[] attributeValue1 = fs.getAbfsStore().encodeAttribute("hi");
- byte[] attributeValue2 = fs.getAbfsStore().encodeAttribute("ä½ å¥½");
- String attributeName1 = "user.asciiAttribute";
- String attributeName2 = "user.unicodeAttribute";
- Path testFile = path("setGetXAttr");
-
- // after creating a file, the xAttr should not be present
- touch(testFile);
- assertNull(fs.getXAttr(testFile, attributeName1));
-
- // after setting the xAttr on the file, the value should be retrievable
- fs.registerListener(
- new TracingHeaderValidator(conf.getClientCorrelationId(),
- fs.getFileSystemId(), FSOperationType.SET_ATTR, true, 0));
- fs.setXAttr(testFile, attributeName1, attributeValue1);
- fs.setListenerOperation(FSOperationType.GET_ATTR);
- assertArrayEquals(attributeValue1, fs.getXAttr(testFile, attributeName1));
- fs.registerListener(null);
-
- // after setting a second xAttr on the file, the first xAttr values should not be overwritten
- fs.setXAttr(testFile, attributeName2, attributeValue2);
- assertArrayEquals(attributeValue1, fs.getXAttr(testFile, attributeName1));
- assertArrayEquals(attributeValue2, fs.getXAttr(testFile, attributeName2));
+ final Path path = new Path("a/b");
+ fs.create(path);
+
+ String attributeName1 = "user.attribute1";
+ String attributeName2 = "user.attribute2";
+ String decodedAttributeValue1;
+ String decodedAttributeValue2;
+ byte[] attributeValue1;
+ byte[] attributeValue2;
+
+ if(fs.getAbfsStore().getPrefixMode() == PrefixMode.BLOB) {
+ Assume.assumeTrue(!getIsNamespaceEnabled(fs)); // Blob endpoint Currently Supports FNS only
+ decodedAttributeValue1 = "hi";
+ decodedAttributeValue2 = "hello"; //Блюз //ä½ å¥½
+ // TODO: Modify them to unicode characters when support is added
+ attributeValue1 = fs.getAbfsStore().encodeAttribute(decodedAttributeValue1);
+ attributeValue2 = fs.getAbfsStore().encodeAttribute(decodedAttributeValue2);
+ }
+ else {
+ decodedAttributeValue1 = "hi";
+ decodedAttributeValue2 = "hello"; // DFS Endpoint only Supports ASCII
+ attributeValue1 = fs.getAbfsStore().encodeAttribute(decodedAttributeValue1);
+ attributeValue2 = fs.getAbfsStore().encodeAttribute(decodedAttributeValue2);
+ }
+
+ // Attribute not present initially
+ assertNull(fs.getXAttr(path, attributeName1));
+ assertNull(fs.getXAttr(path, attributeName2));
+
+ // Set the Attributes
+ fs.setXAttr(path, attributeName1, attributeValue1);
+
+ // Check if the attribute is retrievable
+ byte[] rv = fs.getXAttr(path, attributeName1);
+ assertTrue(Arrays.equals(rv, attributeValue1));
+ assertEquals(new String(rv, StandardCharsets.UTF_8), decodedAttributeValue1);
+
+ // Set the second Attribute
+ fs.setXAttr(path, attributeName2, attributeValue2);
+
+ // Check all the attributes present and previous Attribute not overridden
+ rv = fs.getXAttr(path, attributeName1);
+ assertTrue(Arrays.equals(rv, attributeValue1));
+ assertEquals(new String(rv, StandardCharsets.UTF_8), decodedAttributeValue1);
+ rv = fs.getXAttr(path, attributeName2);
+ assertTrue(Arrays.equals(rv, attributeValue2));
+ assertEquals(new String(rv, StandardCharsets.UTF_8), decodedAttributeValue2);
+ }
+
+ @Test
+ public void testGetXAttrOnImplicitPath() throws Exception {
+ final AzureBlobFileSystem fs = getFileSystem();
+ AzcopyHelper azcopyHelper = new AzcopyHelper(
+ getAccountName(),
+ getFileSystemName(),
+ getRawConfiguration(),
+ fs.getAbfsStore().getPrefixMode()
+ );
+
+ Path testPath = new Path("a/b");
+ azcopyHelper.createFolderUsingAzcopy(fs.makeQualified(testPath).toUri().getPath().substring(1));
+
+ assertTrue("Path is implicit.",
+ BlobDirectoryStateHelper.isImplicitDirectory(testPath, fs));
+
+ String attributeName1 = "user.attribute1";
+ assertNull(fs.getXAttr(testPath, attributeName1));
+ }
+
+ /**
+ * Trying to set same attribute multiple times should result in no failure
+ * @throws Exception
+ */
+ @Test
+ public void testSetXAttrMultipleOperations() throws Exception {
+ AzureBlobFileSystem fs = getFileSystem();
+ final Path path = new Path("a/b");
+ fs.create(path);
+
+ String attributeName1 = "user.attribute1";
+ byte[] attributeValue1;
+ String decodedAttributeValue1 = "hi";
+
+ if(fs.getAbfsStore().getPrefixMode() == PrefixMode.BLOB) {
+ Assume.assumeTrue(!getIsNamespaceEnabled(fs));
+ // TODO: Modify them to unicode characters when support is added
+ attributeValue1 = fs.getAbfsStore().encodeAttribute("hi");
+ }
+ else {
+ attributeValue1 = fs.getAbfsStore().encodeAttribute("hi");
+ }
+
+ // Attribute not present initially
+ assertNull(fs.getXAttr(path, attributeName1));
+
+ // Set the Attributes Multiple times
+ // Filesystem internally adds create and replace flags
+ fs.setXAttr(path, attributeName1, attributeValue1);
+ fs.setXAttr(path, attributeName1, attributeValue1);
+
+ // Check if the attribute is retrievable
+ byte[] rv = fs.getXAttr(path, attributeName1);
+ assertTrue(Arrays.equals(rv, attributeValue1));
+ assertEquals(new String(rv, StandardCharsets.UTF_8), decodedAttributeValue1);
+ }
+
+ /**
+ * Test that setting metadata over marker blob do not override
+ * x-ms-meta-hdi_IsFolder
+ * TODO: Confirm Expected Behavior
+ * @throws Exception
+ */
+ @Test
+ public void testSetXAttrOverMarkerBlob() throws Exception {
+ AzureBlobFileSystem fs = getFileSystem();
+ final Path path = new Path("a/b");
+ fs.mkdirs(path);
+
+ assertTrue(BlobDirectoryStateHelper.isExplicitDirectory(path, fs));
+
+ String attributeName1 = "user.attribute1";
+ byte[] attributeValue1;
+
+ if(fs.getAbfsStore().getPrefixMode() == PrefixMode.BLOB) {
+ Assume.assumeTrue(!getIsNamespaceEnabled(fs));
+ // TODO: Modify them to unicode characters when support is added
+ attributeValue1 = fs.getAbfsStore().encodeAttribute("hi");
+ }
+ else {
+ attributeValue1 = fs.getAbfsStore().encodeAttribute("hi");
+ }
+
+ // Attribute not present initially
+ assertNull(fs.getXAttr(path, attributeName1));
+
+ // Set the Attribute on marker blob
+ fs.setXAttr(path, attributeName1, attributeValue1);
+
+ // Check if the attribute is retrievable
+ byte[] rv = fs.getXAttr(path, attributeName1);
+ assertTrue(Arrays.equals(rv, attributeValue1));
+
+ // Check if Marker blob still exists as marker.
+ assertTrue(BlobDirectoryStateHelper.isExplicitDirectory(path, fs));
}
@Test
public void testSetGetXAttrCreateReplace() throws Exception {
AzureBlobFileSystem fs = getFileSystem();
- Assume.assumeTrue(getIsNamespaceEnabled(fs));
- byte[] attributeValue = fs.getAbfsStore().encodeAttribute("one");
- String attributeName = "user.someAttribute";
- Path testFile = path("createReplaceXAttr");
+ final Path testFile = new Path("a/b");
+
+ String attributeName = "user.attribute1";
+ String decodedAttributeValue1;
+ byte[] attributeValue;
+
+ if(fs.getAbfsStore().getPrefixMode() == PrefixMode.BLOB) {
+ Assume.assumeTrue(!getIsNamespaceEnabled(fs)); // Blob endpoint Currently Supports FNS only
+ decodedAttributeValue1 = "hi";
+ attributeValue = decodedAttributeValue1.getBytes(StandardCharsets.UTF_8);
+ }
+ else {
+ decodedAttributeValue1 = "hi";
+ attributeValue = fs.getAbfsStore().encodeAttribute(decodedAttributeValue1);
+ }
// after creating a file, it must be possible to create a new xAttr
- touch(testFile);
+ fs.create(testFile);
fs.setXAttr(testFile, attributeName, attributeValue, CREATE_FLAG);
assertArrayEquals(attributeValue, fs.getXAttr(testFile, attributeName));
- // however after the xAttr is created, creating it again must fail
+ // however, after the xAttr is created, creating it again must fail
intercept(IOException.class, () -> fs.setXAttr(testFile, attributeName, attributeValue, CREATE_FLAG));
}
@Test
public void testSetGetXAttrReplace() throws Exception {
AzureBlobFileSystem fs = getFileSystem();
- Assume.assumeTrue(getIsNamespaceEnabled(fs));
- byte[] attributeValue1 = fs.getAbfsStore().encodeAttribute("one");
- byte[] attributeValue2 = fs.getAbfsStore().encodeAttribute("two");
- String attributeName = "user.someAttribute";
- Path testFile = path("replaceXAttr");
+ final Path testFile = new Path("a/b");
+
+ String attributeName = "user.attribute1";
+ String decodedAttributeValue1 = "one";
+ String decodedAttributeValue2 = "two";
+
+ byte[] attributeValue1;
+ byte[] attributeValue2;
+
+ if(fs.getAbfsStore().getPrefixMode() == PrefixMode.BLOB) {
+ Assume.assumeTrue(!getIsNamespaceEnabled(fs)); // Blob endpoint Currently Supports FNS only
+ attributeValue1 = decodedAttributeValue1.getBytes(StandardCharsets.UTF_8);
+ attributeValue2 = decodedAttributeValue2.getBytes(StandardCharsets.UTF_8);
+ }
+ else {
+ attributeValue1 = fs.getAbfsStore().encodeAttribute(decodedAttributeValue1);
+ attributeValue2 = fs.getAbfsStore().encodeAttribute(decodedAttributeValue2);
+ }
// after creating a file, it must not be possible to replace an xAttr
intercept(IOException.class, () -> {
- touch(testFile);
+ fs.create(testFile);
fs.setXAttr(testFile, attributeName, attributeValue1, REPLACE_FLAG);
});
- // however after the xAttr is created, replacing it must succeed
+ // however, after the xAttr is created, replacing it must succeed
fs.setXAttr(testFile, attributeName, attributeValue1, CREATE_FLAG);
fs.setXAttr(testFile, attributeName, attributeValue2, REPLACE_FLAG);
assertArrayEquals(attributeValue2, fs.getXAttr(testFile, attributeName));
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemCreate.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemCreate.java
index eaf2f94269ed60..122b9ab36d81db 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemCreate.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemCreate.java
@@ -61,7 +61,7 @@
import org.apache.hadoop.fs.azurebfs.services.AbfsClient;
import org.apache.hadoop.fs.azurebfs.services.AbfsRestOperation;
import org.apache.hadoop.fs.azurebfs.services.AbfsHttpOperation;
-import org.apache.hadoop.fs.azurebfs.services.TestAbfsClient;
+import org.apache.hadoop.fs.azurebfs.services.ITestAbfsClient;
import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
import org.apache.hadoop.fs.azurebfs.utils.TracingHeaderValidator;
import org.mockito.Mockito;
@@ -75,6 +75,7 @@
import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_CLIENT_PROVIDED_ENCRYPTION_KEY;
import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.TRUE;
import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ENABLE_BLOB_MKDIR_OVERWRITE;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_LEASE_CREATE_NON_RECURSIVE;
import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_MKDIRS_FALLBACK_TO_DFS;
import static org.apache.hadoop.fs.azurebfs.constants.FileSystemUriSchemes.ABFS_DNS_PREFIX;
import static org.apache.hadoop.fs.azurebfs.constants.FileSystemUriSchemes.WASB_DNS_PREFIX;
@@ -954,7 +955,9 @@ public void testActiveCreateNonRecursiveDenyParallelReadOnAtomicDir() throws Exc
Assume.assumeTrue(
getFileSystem().getAbfsStore().getAbfsConfiguration().getPrefixMode()
== PrefixMode.BLOB);
- AzureBlobFileSystem fileSystem = (AzureBlobFileSystem) FileSystem.newInstance(getRawConfiguration());
+ Configuration configuration = Mockito.spy(getRawConfiguration());
+ configuration.set(FS_AZURE_LEASE_CREATE_NON_RECURSIVE, "true");
+ AzureBlobFileSystem fileSystem = (AzureBlobFileSystem) FileSystem.newInstance(configuration);
AbfsClient client = Mockito.spy(fileSystem.getAbfsClient());
fileSystem.getAbfsStore().setClient(client);
fileSystem.setWorkingDirectory(new Path("/"));
@@ -994,6 +997,53 @@ public void testActiveCreateNonRecursiveDenyParallelReadOnAtomicDir() throws Exc
Assert.assertTrue(fileSystem.exists(new Path("/hbase/dir/file")));
}
+ @Test
+ public void testActiveCreateNonRecursiveNotDenyParallelReadOnAtomicDirIfLeaseConfigDisabled() throws Exception {
+ Assume.assumeTrue(
+ getFileSystem().getAbfsStore().getAbfsConfiguration().getPrefixMode()
+ == PrefixMode.BLOB);
+ Configuration configuration = Mockito.spy(getRawConfiguration());
+ AzureBlobFileSystem fileSystem = (AzureBlobFileSystem) FileSystem.newInstance(configuration);
+ AbfsClient client = Mockito.spy(fileSystem.getAbfsClient());
+ fileSystem.getAbfsStore().setClient(client);
+ fileSystem.setWorkingDirectory(new Path("/"));
+ fileSystem.mkdirs(new Path("/hbase/dir"));
+ fileSystem.create(new Path("/hbase/dir/file"));
+ AtomicBoolean createCalled = new AtomicBoolean(false);
+ AtomicBoolean parallelRenameDone = new AtomicBoolean(false);
+ AtomicBoolean exceptionCaught = new AtomicBoolean(false);
+
+ Mockito.doAnswer(answer -> {
+ AbfsRestOperation op = (AbfsRestOperation) answer.callRealMethod();
+ createCalled.set(true);
+ while(!parallelRenameDone.get());
+ return op;
+ }).when(client).createPathBlob(Mockito.anyString(), Mockito.anyBoolean(),
+ Mockito.anyBoolean(), Mockito.nullable(HashMap.class), Mockito.nullable(String.class), Mockito.nullable(TracingContext.class));
+
+ new Thread(() -> {
+ try {
+ while(!createCalled.get());
+ getFileSystem().rename(new Path("/hbase/dir/"), new Path("/hbase/dir2"));
+ } catch (Exception e) {
+ exceptionCaught.set(true);
+ } finally {
+ parallelRenameDone.set(true);
+ }
+ }).start();
+
+ fileSystem.createFile(new Path("/hbase/dir/file1"))
+ .overwrite(false)
+ .replication((short) 1)
+ .bufferSize(1024)
+ .blockSize(1024)
+ .build();
+
+ Assert.assertFalse(exceptionCaught.get());
+ Assert.assertFalse(fileSystem.exists(new Path("/hbase/dir/file")));
+ Assert.assertTrue(fileSystem.exists(new Path("/hbase/dir2/file")));
+ }
+
/**
* Attempts to use to the ABFS stream after it is closed.
*/
@@ -1123,9 +1173,15 @@ public void testCreateFileOverwrite(boolean enableConditionalCreateOverwrite)
// One request to server to create path should be issued
// two calls added for -
- // 1. getFileStatus
- // 2. actual create call
+ // 1. getFileStatus : 1
+ // 2. actual create call: 1
createRequestCount+=2;
+
+ // In case of blob endpoint getFileStatus makes additional call to check if path is implicit
+ if (fs.getAbfsStore().getPrefixMode() == PrefixMode.BLOB) {
+ createRequestCount++;
+ }
+
createRequestCount+=ifBlobCheckIfPathDir;
assertAbfsStatistics(
@@ -1226,7 +1282,7 @@ public void testNegativeScenariosForCreateOverwriteDisabled()
// Get mock AbfsClient with current config
AbfsClient
mockClient
- = TestAbfsClient.getMockAbfsClient(
+ = ITestAbfsClient.getMockAbfsClient(
fs.getAbfsStore().getClient(),
fs.getAbfsStore().getAbfsConfiguration());
@@ -1286,6 +1342,13 @@ public void testNegativeScenariosForCreateOverwriteDisabled()
.when(mockClient)
.getPathStatus(any(String.class), eq(false), any(TracingContext.class));
+ doThrow(fileNotFoundResponseEx) // Scn1: GFS fails with Http404
+ .doThrow(serverErrorResponseEx) // Scn2: GFS fails with Http500
+ .doReturn(successOp) // Scn3: create overwrite=true fails with Http412
+ .doReturn(successOp) // Scn4: create overwrite=true fails with Http500
+ .when(mockClient)
+ .getBlobProperty(any(Path.class), any(TracingContext.class));
+
// mock for overwrite=true
doThrow(
preConditionResponseEx) // Scn3: create overwrite=true fails with Http412
@@ -1432,7 +1495,12 @@ private String extractFileEtag(String fileName) throws IOException {
final AzureBlobFileSystem fs = getFileSystem();
final AbfsClient client = fs.getAbfsClient();
final TracingContext testTracingContext = getTestTracingContext(fs, false);
- AbfsRestOperation op = client.getPathStatus(fileName, true, testTracingContext);
+ AbfsRestOperation op;
+ if (fs.getAbfsStore().getPrefixMode() == PrefixMode.BLOB) {
+ op = client.getBlobProperty(new Path(fileName), testTracingContext);
+ } else {
+ op = client.getPathStatus(fileName, true, testTracingContext);
+ }
return AzureBlobFileSystemStore.extractEtagHeader(op.getResult());
}
}
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemDelete.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemDelete.java
index bc18c85a9e711c..a2549aa17b5113 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemDelete.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemDelete.java
@@ -38,7 +38,7 @@
import org.apache.hadoop.fs.azurebfs.services.AbfsClient;
import org.apache.hadoop.fs.azurebfs.services.AbfsHttpOperation;
import org.apache.hadoop.fs.azurebfs.services.AbfsRestOperation;
-import org.apache.hadoop.fs.azurebfs.services.TestAbfsClient;
+import org.apache.hadoop.fs.azurebfs.services.ITestAbfsClient;
import org.apache.hadoop.fs.azurebfs.services.TestAbfsPerfTracker;
import org.apache.hadoop.fs.azurebfs.utils.TestMockHelpers;
import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
@@ -195,7 +195,7 @@ public void testDeleteIdempotency() throws Exception {
final AzureBlobFileSystem fs = getFileSystem();
AbfsClient abfsClient = fs.getAbfsStore().getClient();
- AbfsClient testClient = TestAbfsClient.createTestClientFromCurrentContext(
+ AbfsClient testClient = ITestAbfsClient.createTestClientFromCurrentContext(
abfsClient,
abfsConfig);
@@ -242,7 +242,7 @@ public void testDeleteIdempotency() throws Exception {
public void testDeleteIdempotencyTriggerHttp404() throws Exception {
final AzureBlobFileSystem fs = getFileSystem();
- AbfsClient client = TestAbfsClient.createTestClientFromCurrentContext(
+ AbfsClient client = ITestAbfsClient.createTestClientFromCurrentContext(
fs.getAbfsStore().getClient(),
this.getConfiguration());
@@ -261,7 +261,7 @@ public void testDeleteIdempotencyTriggerHttp404() throws Exception {
getTestTracingContext(fs, true)));
// mock idempotency check to mimic retried case
- AbfsClient mockClient = TestAbfsClient.getMockAbfsClient(
+ AbfsClient mockClient = ITestAbfsClient.getMockAbfsClient(
fs.getAbfsStore().getClient(),
this.getConfiguration());
AzureBlobFileSystemStore mockStore = mock(AzureBlobFileSystemStore.class);
@@ -276,10 +276,10 @@ public void testDeleteIdempotencyTriggerHttp404() throws Exception {
// Case 2: Mimic retried case
// Idempotency check on Delete always returns success
- AbfsRestOperation idempotencyRetOp = TestAbfsClient.getRestOp(
+ AbfsRestOperation idempotencyRetOp = ITestAbfsClient.getRestOp(
DeletePath, mockClient, HTTP_METHOD_DELETE,
- TestAbfsClient.getTestUrl(mockClient, "/NonExistingPath"),
- TestAbfsClient.getTestRequestHeaders(mockClient));
+ ITestAbfsClient.getTestUrl(mockClient, "/NonExistingPath"),
+ ITestAbfsClient.getTestRequestHeaders(mockClient));
idempotencyRetOp.hardSetResult(HTTP_OK);
doReturn(idempotencyRetOp).when(mockClient).deleteIdempotencyCheckOp(any());
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemExplictImplicitRename.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemExplictImplicitRename.java
index 8ffbbadf13a18d..44f2f1b9c94754 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemExplictImplicitRename.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemExplictImplicitRename.java
@@ -49,32 +49,6 @@ public void setup() throws Exception {
== PrefixMode.BLOB);
}
- /**
- * For creating directory with implicit parents. Doesn't change already explicit
- * parents.
- */
- void createAzCopyDirectory(Path path) throws Exception {
- AzcopyHelper azcopyHelper = new AzcopyHelper(
- getAccountName(), getFileSystemName(), getFileSystem().getAbfsStore()
- .getAbfsConfiguration()
- .getRawConfiguration(), getFileSystem().getAbfsStore().getPrefixMode());
- azcopyHelper.createFolderUsingAzcopy(
- getFileSystem().makeQualified(path).toUri().getPath().substring(1));
- }
-
- /**
- * For creating files with implicit parents. Doesn't change already explicit
- * parents.
- */
- void createAzCopyFile(Path path) throws Exception {
- AzcopyHelper azcopyHelper = new AzcopyHelper(getAccountName(),
- getFileSystemName(), getFileSystem().getAbfsStore()
- .getAbfsConfiguration()
- .getRawConfiguration(), getFileSystem().getAbfsStore().getPrefixMode());
- azcopyHelper.createFileUsingAzcopy(
- getFileSystem().makeQualified(path).toUri().getPath().substring(1));
- }
-
@Test
public void testRenameSrcFileInImplicitParentDirectory() throws Exception {
AzureBlobFileSystem fs = getFileSystem();
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemFileStatus.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemFileStatus.java
index b9498be89a1ffe..4136da519ec913 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemFileStatus.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemFileStatus.java
@@ -27,6 +27,8 @@
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
+import static org.apache.hadoop.test.LambdaTestUtils.intercept;
+
/**
* Test FileStatus.
*/
@@ -140,4 +142,158 @@ public void testLastModifiedTime() throws IOException {
assertTrue("lastModifiedTime should be before createEndTime",
createEndTime > lastModifiedTime);
}
+
+ @Test
+ public void testFileStatusOnFileWithImplicitParent() throws Exception {
+ final AzureBlobFileSystem fs = getFileSystem();
+ AzcopyHelper azcopyHelper = new AzcopyHelper(
+ getAccountName(),
+ getFileSystemName(),
+ getRawConfiguration(),
+ fs.getAbfsStore().getPrefixMode()
+ );
+
+ Path testPath = new Path("a/b.txt");
+ azcopyHelper.createFileUsingAzcopy(fs.makeQualified(testPath).toUri().getPath().substring(1));
+
+ assertTrue("Parent directory is implicit.",
+ BlobDirectoryStateHelper.isImplicitDirectory(testPath.getParent(), fs));
+
+ // Assert getFileStatus Succeed on path
+ FileStatus fileStatus = fs.getFileStatus(testPath);
+ assertNotNull(fileStatus.getPath());
+ assertFalse(fileStatus.isDirectory());
+ assertNotEquals(0L, fileStatus.getLen());
+ }
+
+ @Test
+ public void testFileStatusOnFileWithExplicitParent() throws Exception {
+ final AzureBlobFileSystem fs = getFileSystem();
+ Path testPath = new Path("a/b.txt");
+ fs.create(testPath);
+
+ assertTrue("Parent directory is explicit.",
+ BlobDirectoryStateHelper.isExplicitDirectory(testPath.getParent(), fs));
+
+ FileStatus fileStatus = fs.getFileStatus(testPath);
+ assertNotNull(fileStatus.getPath());
+ assertFalse(fileStatus.isDirectory());
+ }
+
+ @Test
+ public void testFileStatusOnImplicitDirWithImplicitParent() throws Exception {
+ final AzureBlobFileSystem fs = getFileSystem();
+ AzcopyHelper azcopyHelper = new AzcopyHelper(
+ getAccountName(),
+ getFileSystemName(),
+ getRawConfiguration(),
+ fs.getAbfsStore().getPrefixMode()
+ );
+
+ Path testPath = new Path("a/b");
+ azcopyHelper.createFolderUsingAzcopy(fs.makeQualified(testPath).toUri().getPath().substring(1));
+
+ assertTrue("Path is implicit.",
+ BlobDirectoryStateHelper.isImplicitDirectory(testPath, fs));
+ assertTrue("Parent directory is implicit.",
+ BlobDirectoryStateHelper.isImplicitDirectory(testPath.getParent(), fs));
+
+ // Assert that getFileStatus succeeds
+ FileStatus fileStatus = fs.getFileStatus(testPath);
+ assertNotNull(fileStatus.getPath());
+ assertTrue(fileStatus.isDirectory());
+ assertEquals(0L, fileStatus.getLen());
+ }
+
+ @Test
+ public void testFileStatusOnImplicitDirWithExplicitParent() throws Exception {
+ final AzureBlobFileSystem fs = getFileSystem();
+ AzcopyHelper azcopyHelper = new AzcopyHelper(
+ getAccountName(),
+ getFileSystemName(),
+ getRawConfiguration(),
+ fs.getAbfsStore().getPrefixMode()
+ );
+
+ Path testPath = new Path("a/b");
+ azcopyHelper.createFolderUsingAzcopy(fs.makeQualified(testPath).toUri().getPath().substring(1));
+ fs.mkdirs(testPath.getParent());
+
+ assertTrue("Path is implicit.",
+ BlobDirectoryStateHelper.isImplicitDirectory(testPath, fs));
+ assertTrue("Parent directory is explicit.",
+ BlobDirectoryStateHelper.isExplicitDirectory(testPath.getParent(), fs));
+
+ // Assert that getFileStatus succeeds
+ FileStatus fileStatus = fs.getFileStatus(testPath);
+ assertNotNull(fileStatus.getPath());
+ assertTrue(fileStatus.isDirectory());
+ assertEquals(0L, fileStatus.getLen());
+ }
+
+ @Test
+ public void testFileStatusOnExplicitDirWithExplicitParent() throws Exception {
+ final AzureBlobFileSystem fs = getFileSystem();
+ Path testPath = new Path("a/b");
+ fs.mkdirs(testPath);
+
+ assertTrue("Parent directory is explicit.",
+ BlobDirectoryStateHelper.isExplicitDirectory(testPath.getParent(), fs));
+ assertTrue("Path is explicit.",
+ BlobDirectoryStateHelper.isExplicitDirectory(testPath, fs));
+
+ // Assert that getFileStatus Succeeds
+ FileStatus fileStatus = fs.getFileStatus(testPath);
+ assertNotNull(fileStatus.getPath());
+ assertTrue(fileStatus.isDirectory());
+ assertEquals(0L, fileStatus.getLen());
+ }
+
+ @Test
+ public void testFileStatusOnNonExistingPathWithExplicitParent() throws Exception {
+ final AzureBlobFileSystem fs = getFileSystem();
+ Path testPath = new Path("a/b.txt");
+ fs.mkdirs(testPath.getParent());
+
+ assertTrue("Parent directory is explicit.",
+ BlobDirectoryStateHelper.isExplicitDirectory(testPath.getParent(), fs));
+
+ // assert that getFileStatus fails
+ intercept(IOException.class,
+ () -> fs.getFileStatus(testPath));
+ }
+
+ @Test
+ public void testFileStatusOnNonExistingPathWithImplicitParent() throws Exception {
+ final AzureBlobFileSystem fs = getFileSystem();
+ AzcopyHelper azcopyHelper = new AzcopyHelper(
+ getAccountName(),
+ getFileSystemName(),
+ getRawConfiguration(),
+ fs.getAbfsStore().getPrefixMode()
+ );
+
+ Path testPath = new Path("a/b.txt");
+ azcopyHelper.createFolderUsingAzcopy(fs.makeQualified(
+ testPath.getParent()).toUri().getPath().substring(1));
+
+ assertTrue("Parent directory is implicit.",
+ BlobDirectoryStateHelper.isImplicitDirectory(testPath.getParent(), fs));
+
+ // assert that getFileStatus Fails with IOException
+ intercept(IOException.class,
+ () -> fs.getFileStatus(testPath));
+ }
+
+ @Test
+ public void testFileStatusOnRoot() throws Exception {
+ final AzureBlobFileSystem fs = getFileSystem();
+ final Path path = new Path("/");
+ fs.setWorkingDirectory(new Path("/"));
+
+ // Assert that getFileSus on root path succeed.
+ FileStatus fileStatus = fs.getFileStatus(path);
+ assertTrue(fileStatus.isDirectory());
+ assertTrue(fileStatus.getLen() == 0L);
+ }
}
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemLease.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemLease.java
index 2ebdc7492a18d6..14e9df12f47038 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemLease.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemLease.java
@@ -20,6 +20,8 @@
import java.io.IOException;
import java.util.concurrent.RejectedExecutionException;
+import org.apache.hadoop.fs.azurebfs.services.AbfsBlobLease;
+import org.apache.hadoop.fs.azurebfs.services.AbfsDfsLease;
import org.apache.hadoop.fs.azurebfs.services.OperativeEndpoint;
import org.apache.hadoop.fs.azurebfs.services.PrefixMode;
import org.junit.Assert;
@@ -347,8 +349,14 @@ public void testAcquireRetry() throws Exception {
FSOperationType.TEST_OP, true, 0);
tracingContext.setListener(listener);
- AbfsLease lease = new AbfsLease(fs.getAbfsClient(),
- testFilePath.toUri().getPath(), tracingContext);
+ AbfsLease lease;
+ if(getPrefixMode(fs) == PrefixMode.BLOB) {
+ lease = new AbfsBlobLease(fs.getAbfsClient(),
+ testFilePath.toUri().getPath(), null, tracingContext);
+ } else {
+ lease = new AbfsDfsLease(fs.getAbfsClient(),
+ testFilePath.toUri().getPath(), null, tracingContext);
+ }
Assert.assertNotNull("Did not successfully lease file", lease.getLeaseID());
listener.setOperation(FSOperationType.RELEASE_LEASE);
lease.free();
@@ -362,7 +370,18 @@ public void testAcquireRetry() throws Exception {
.doCallRealMethod().when(mockClient)
.acquireLease(anyString(), anyInt(), any(TracingContext.class));
- lease = new AbfsLease(mockClient, testFilePath.toUri().getPath(), 5, 1, tracingContext);
+ doThrow(new AbfsLease.LeaseException("failed to acquire 1"))
+ .doThrow(new AbfsLease.LeaseException("failed to acquire 2"))
+ .doCallRealMethod().when(mockClient)
+ .acquireBlobLease(anyString(), anyInt(), any(TracingContext.class));
+
+ if(getPrefixMode(fs) == PrefixMode.BLOB) {
+ lease = new AbfsBlobLease(mockClient, testFilePath.toUri().getPath(), 5, 1, null,
+ tracingContext);
+ } else {
+ lease = new AbfsDfsLease(mockClient, testFilePath.toUri().getPath(), 5, 1, null,
+ tracingContext);
+ }
Assert.assertNotNull("Acquire lease should have retried", lease.getLeaseID());
lease.free();
Assert.assertEquals("Unexpected acquire retry count", 2, lease.getAcquireRetryCount());
@@ -370,9 +389,17 @@ public void testAcquireRetry() throws Exception {
doThrow(new AbfsLease.LeaseException("failed to acquire")).when(mockClient)
.acquireLease(anyString(), anyInt(), any(TracingContext.class));
+ doThrow(new AbfsLease.LeaseException("failed to acquire")).when(mockClient)
+ .acquireBlobLease(anyString(), anyInt(), any(TracingContext.class));
+
LambdaTestUtils.intercept(AzureBlobFileSystemException.class, () -> {
- new AbfsLease(mockClient, testFilePath.toUri().getPath(), 5, 1,
- tracingContext);
+ if(getPrefixMode(fs) == PrefixMode.BLOB) {
+ new AbfsBlobLease(mockClient, testFilePath.toUri().getPath(), 5, 1, null,
+ tracingContext);
+ } else {
+ new AbfsDfsLease(mockClient, testFilePath.toUri().getPath(), 5, 1, null,
+ tracingContext);
+ }
});
}
}
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemMkDir.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemMkDir.java
index 6bed7839935f7c..acdcf66942ab78 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemMkDir.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemMkDir.java
@@ -20,8 +20,12 @@
import java.util.UUID;
+import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsRestOperationException;
+import org.apache.hadoop.fs.azurebfs.services.AbfsClient;
import org.apache.hadoop.fs.azurebfs.services.OperativeEndpoint;
import org.apache.hadoop.fs.azurebfs.services.PrefixMode;
+import org.apache.hadoop.fs.azurebfs.services.ITestAbfsClient;
+import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
import org.junit.Assume;
import org.junit.Test;
@@ -29,6 +33,7 @@
import org.apache.hadoop.fs.FileAlreadyExistsException;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.mockito.Mockito;
import static org.apache.hadoop.fs.azurebfs.AbfsStatistic.CONNECTIONS_MADE;
import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ENABLE_MKDIR_OVERWRITE;
@@ -152,4 +157,29 @@ public void testCreateDirOverwrite(boolean enableConditionalCreateOverwrite)
totalConnectionMadeBeforeTest + mkdirRequestCount,
fs.getInstrumentationMap());
}
+
+ @Test
+ public void testVerifyGetBlobProperty() throws Exception {
+ Assume.assumeTrue(getFileSystem().getAbfsStore().getPrefixMode() == PrefixMode.BLOB);
+ AzureBlobFileSystem fs = Mockito.spy(getFileSystem());
+ AzureBlobFileSystemStore store = Mockito.spy(fs.getAbfsStore());
+ Mockito.doReturn(store).when(fs).getAbfsStore();
+ AbfsClient client = store.getClient();
+ AbfsClient testClient = Mockito.spy(ITestAbfsClient.createTestClientFromCurrentContext(
+ client,
+ fs.getAbfsStore().getAbfsConfiguration()));
+ store.setClient(testClient);
+
+ createAzCopyDirectory(new Path("/src"));
+ intercept(AbfsRestOperationException.class, () -> {
+ store.getBlobProperty(new Path("/src"), Mockito.mock(
+ TracingContext.class));
+ });
+ fs.mkdirs(new Path("/src/dir"));
+ Mockito.verify(testClient, Mockito.times(0)).getPathStatus(Mockito.any(String.class),
+ Mockito.anyBoolean(), Mockito.any(TracingContext.class));
+ Mockito.verify(testClient, Mockito.times(1)).getBlobProperty(Mockito.any(Path.class),
+ Mockito.any(TracingContext.class));
+
+ }
}
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemRandomRead.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemRandomRead.java
index 0a9713cffa94b5..b4a8582c57d78c 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemRandomRead.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemRandomRead.java
@@ -18,14 +18,20 @@
package org.apache.hadoop.fs.azurebfs;
import java.io.EOFException;
+import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.UUID;
+import org.apache.hadoop.fs.azurebfs.services.AbfsClient;
+import org.apache.hadoop.fs.azurebfs.services.PrefixMode;
+import org.apache.hadoop.fs.azurebfs.services.ITestAbfsClient;
+import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
import org.junit.Assume;
import org.junit.Ignore;
import org.junit.Test;
+import org.mockito.Mockito;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -574,6 +580,46 @@ public void testAlwaysReadBufferSizeConfig(boolean alwaysReadBufferSizeConfigVal
assertAbfsStatistics(BYTES_RECEIVED, dateSizeReadStatAtStart + newDataSizeRead, fs.getInstrumentationMap());
}
+ @Test
+ public void testReadBlob() throws IOException {
+ Assume.assumeTrue(PrefixMode.BLOB == getFileSystem().getAbfsStore().getPrefixMode());
+ AzureBlobFileSystem fs = Mockito.spy(getFileSystem());
+ AzureBlobFileSystemStore store = Mockito.spy(fs.getAbfsStore());
+ AbfsClient client = store.getClient();
+ AbfsClient mockClient = Mockito.spy(ITestAbfsClient.createTestClientFromCurrentContext(
+ client,
+ fs.getAbfsStore().getAbfsConfiguration()
+ ));
+ store.setClient(mockClient);
+ Mockito.doReturn(mockClient).when(store).getClient();
+ Mockito.doReturn(store).when(fs).getAbfsStore();
+
+ Path testPath = new Path("/testReadFile");
+ fs.create(testPath);
+ FSDataInputStream in = fs.open(testPath);
+ Mockito.verify(mockClient, Mockito.atLeast(1)).getBlobProperty(
+ Mockito.any(Path.class), Mockito.any(TracingContext.class));
+ Mockito.verify(mockClient, Mockito.times(0)).getPathStatus(
+ Mockito.any(String.class), Mockito.anyBoolean(), Mockito.any(TracingContext.class));
+ }
+
+ @Test
+ public void testInvalidImplicitDirRead() throws Exception {
+ AzureBlobFileSystem fs = (AzureBlobFileSystem) getFileSystem();
+ AzcopyHelper azcopyhelper = new AzcopyHelper(getAccountName(),
+ getFileSystemName(),
+ getRawConfiguration(),
+ fs.getAbfsStore().getPrefixMode());
+ String fullPath = "/implicitDirPath/testFile";
+ String path = "/implicitDirPath";
+ azcopyhelper.createFolderUsingAzcopy(
+ fs.makeQualified(new Path(fullPath)).toUri().getPath().substring(1)
+ );
+
+ intercept(FileNotFoundException.class, () ->
+ fs.open(new Path(path)));
+
+ }
private long sequentialRead(String version,
Path testPath,
FileSystem fs,
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemRename.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemRename.java
index 5057664ab2af17..a3eebf3e9dc9cc 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemRename.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemRename.java
@@ -61,6 +61,7 @@
import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
import org.apache.hadoop.fs.azurebfs.services.PrefixMode;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_LEASE_CREATE_NON_RECURSIVE;
import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_REDIRECT_RENAME;
import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.TRUE;
import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_INGRESS_FALLBACK_TO_DFS;
@@ -1735,8 +1736,10 @@ public void testParallelAppendToFileBeingCopiedInAtomicDirectory()
@Test
public void testParallelCreateNonRecursiveToFilePartOfAtomicDirectoryInRename()
throws Exception {
- FileSystem fsCreate = FileSystem.newInstance(getRawConfiguration());
- AzureBlobFileSystem fs = (AzureBlobFileSystem) FileSystem.newInstance(getRawConfiguration());
+ Configuration configuration = Mockito.spy(getRawConfiguration());
+ configuration.set(FS_AZURE_LEASE_CREATE_NON_RECURSIVE, "true");
+ FileSystem fsCreate = FileSystem.newInstance(configuration);
+ AzureBlobFileSystem fs = (AzureBlobFileSystem) FileSystem.newInstance(configuration);
assumeNonHnsAccountBlobEndpoint(fs);
fs.setWorkingDirectory(new Path("/"));
fs.mkdirs(new Path("/hbase/dir1"));
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestCustomerProvidedKey.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestCustomerProvidedKey.java
index 2c0bd31bf8eeb8..8ef5f1d4516703 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestCustomerProvidedKey.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestCustomerProvidedKey.java
@@ -202,7 +202,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
@@ -247,7 +247,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
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestGetNameSpaceEnabled.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestGetNameSpaceEnabled.java
index ccfdb72cdb0c98..fbfd6884bb363a 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestGetNameSpaceEnabled.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestGetNameSpaceEnabled.java
@@ -32,6 +32,7 @@
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.azurebfs.enums.Trilean;
+import org.apache.hadoop.fs.azurebfs.services.PrefixMode;
import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
import static org.apache.hadoop.fs.azurebfs.constants.FileSystemUriSchemes.ABFS_DNS_PREFIX;
@@ -142,11 +143,20 @@ public void testFailedRequestWhenFSNotExist() throws Exception {
+ testUri.substring(testUri.indexOf("@"));
AzureBlobFileSystem fs = this.getFileSystem(nonExistingFsUrl);
- intercept(FileNotFoundException.class,
- "\"The specified filesystem does not exist.\", 404",
- ()-> {
- fs.getFileStatus(new Path("/")); // Run a dummy FS call
- });
+ if (fs.getAbfsStore().getPrefixMode() == PrefixMode.BLOB) {
+ intercept(FileNotFoundException.class,
+ "\"The specified container does not exist.\", 404",
+ ()-> {
+ fs.getFileStatus(new Path("/")); // Run a dummy FS call
+ });
+ }
+ else {
+ intercept(FileNotFoundException.class,
+ "\"The specified filesystem does not exist.\", 404",
+ ()-> {
+ fs.getFileStatus(new Path("/")); // Run a dummy FS call
+ });
+ }
}
@Test
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestListBlobProducer.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestListBlobProducer.java
index 0093f699ce5110..541e17870931bf 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestListBlobProducer.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestListBlobProducer.java
@@ -19,6 +19,12 @@
package org.apache.hadoop.fs.azurebfs;
import java.net.HttpURLConnection;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -34,13 +40,14 @@
import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AzureBlobFileSystemException;
import org.apache.hadoop.fs.azurebfs.services.AbfsClient;
import org.apache.hadoop.fs.azurebfs.services.AbfsRestOperation;
+import org.apache.hadoop.fs.azurebfs.services.BlobProperty;
import org.apache.hadoop.fs.azurebfs.services.ListBlobConsumer;
import org.apache.hadoop.fs.azurebfs.services.ListBlobProducer;
import org.apache.hadoop.fs.azurebfs.services.ListBlobQueue;
import org.apache.hadoop.fs.azurebfs.services.PrefixMode;
import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
-import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_MAX_CONSUMER_LAG;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_PRODUCER_QUEUE_MAX_SIZE;
public class ITestListBlobProducer extends AbstractAbfsIntegrationTest {
@@ -59,7 +66,7 @@ public void setup() throws Exception {
@Test
public void testProducerWaitingForConsumerLagToGoDown() throws Exception {
Configuration configuration = Mockito.spy(getRawConfiguration());
- configuration.set(FS_AZURE_MAX_CONSUMER_LAG, "10");
+ configuration.set(FS_AZURE_PRODUCER_QUEUE_MAX_SIZE, "10");
AzureBlobFileSystem fs = (AzureBlobFileSystem) FileSystem.newInstance(
configuration);
AbfsClient client = fs.getAbfsClient();
@@ -67,57 +74,67 @@ public void testProducerWaitingForConsumerLagToGoDown() throws Exception {
fs.getAbfsStore().setClient(spiedClient);
fs.setWorkingDirectory(new Path("/"));
fs.mkdirs(new Path("/src"));
+ ExecutorService executor = Executors.newFixedThreadPool(5);
+ List futureList = new ArrayList<>();
for (int i = 0; i < 20; i++) {
- fs.create(new Path("/src/file" + i));
+ int iter = i;
+ futureList.add(executor.submit(() -> {
+ return fs.create(new Path("/src/file" + iter));
+ }));
+ }
+ for(Future future : futureList) {
+ future.get();
}
- AtomicBoolean produced = new AtomicBoolean(true);
AtomicInteger producedBlobs = new AtomicInteger(0);
AtomicInteger listBlobInvoked = new AtomicInteger(0);
+ final ITestListBlobProducer testObj = this;
+ final ListBlobQueue queue = new ListBlobQueue(
+ fs.getAbfsStore().getAbfsConfiguration().getProducerQueueMaxSize(),
+ 1);
+
Mockito.doAnswer(answer -> {
- listBlobInvoked.incrementAndGet();
- AbfsRestOperation op = client.getListBlobs(answer.getArgument(0),
- answer.getArgument(1), 1, answer.getArgument(3));
- producedBlobs.incrementAndGet();
- produced.set(true);
- return op;
+ synchronized (testObj) {
+ listBlobInvoked.incrementAndGet();
+ AbfsRestOperation op = client.getListBlobs(answer.getArgument(0),
+ answer.getArgument(1), 1, answer.getArgument(3));
+ producedBlobs.incrementAndGet();
+ if(producedBlobs.get() > 10) {
+ Assert.assertTrue(queue.availableSize() > 0);
+ }
+ return op;
+ }
})
.when(spiedClient)
.getListBlobs(Mockito.nullable(String.class),
Mockito.nullable(String.class), Mockito.nullable(Integer.class),
Mockito.nullable(TracingContext.class));
- ListBlobQueue queue = new ListBlobQueue(null);
+
ListBlobProducer producer = new ListBlobProducer("src/", spiedClient, queue,
null, Mockito.mock(
TracingContext.class));
ListBlobConsumer consumer = new ListBlobConsumer(queue);
while (producedBlobs.get() < 10) ;
- int producedBlobCount = producedBlobs.get();
-
int oldInvocation = listBlobInvoked.get();
- Thread.sleep(10_000L);
Assert.assertTrue(listBlobInvoked.get() == oldInvocation);
while (!consumer.isCompleted()) {
- produced.set(false);
- consumer.consume();
- while (!produced.get() && !queue.getIsCompleted()) ;
- if (!queue.getIsCompleted()) {
- Assert.assertEquals(producedBlobs.get() - 1, producedBlobCount);
+ synchronized (testObj) {
+ consumer.consume();
+ Assert.assertTrue(queue.availableSize() > 0);
}
- producedBlobCount = producedBlobs.get();
}
- Assert.assertTrue(producedBlobCount == 20);
+ Assert.assertTrue(producedBlobs.get() == 20);
}
@Test
public void testConsumerWhenProducerThrowException() throws Exception {
Configuration configuration = Mockito.spy(getRawConfiguration());
- configuration.set(FS_AZURE_MAX_CONSUMER_LAG, "10");
+ configuration.set(FS_AZURE_PRODUCER_QUEUE_MAX_SIZE, "10");
AzureBlobFileSystem fs = (AzureBlobFileSystem) FileSystem.newInstance(
configuration);
AbfsClient client = fs.getAbfsClient();
@@ -139,7 +156,8 @@ public void testConsumerWhenProducerThrowException() throws Exception {
Mockito.nullable(String.class), Mockito.nullable(Integer.class),
Mockito.nullable(TracingContext.class));
- ListBlobQueue queue = new ListBlobQueue(null);
+ ListBlobQueue queue = new ListBlobQueue(getConfiguration().getProducerQueueMaxSize(),
+ getConfiguration().getProducerQueueMaxSize());
ListBlobProducer producer = new ListBlobProducer("src/", spiedClient, queue,
null, Mockito.mock(
TracingContext.class));
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/constants/TestConfigurationKeys.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/constants/TestConfigurationKeys.java
index 565eb38c4f70a7..9e40f22d231b05 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/constants/TestConfigurationKeys.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/constants/TestConfigurationKeys.java
@@ -24,6 +24,9 @@
public final class TestConfigurationKeys {
public static final String FS_AZURE_ACCOUNT_NAME = "fs.azure.account.name";
public static final String FS_AZURE_ABFS_ACCOUNT_NAME = "fs.azure.abfs.account.name";
+ public static final String FS_AZURE_ABFS_ACCOUNT1_NAME = "fs.azure.abfs.account1.name";
+ public static final String FS_AZURE_ENABLE_AUTOTHROTTLING = "fs.azure.enable.autothrottling";
+ public static final String FS_AZURE_ANALYSIS_PERIOD = "fs.azure.analysis.period";
public static final String FS_AZURE_ACCOUNT_KEY = "fs.azure.account.key";
public static final String FS_AZURE_CONTRACT_TEST_URI = "fs.contract.test.fs.abfs";
public static final String FS_AZURE_TEST_NAMESPACE_ENABLED_ACCOUNT = "fs.azure.test.namespace.enabled";
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/AbfsRestOperationTestUtil.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/AbfsRestOperationTestUtil.java
index d1c661eea3b780..56556b1930566d 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/AbfsRestOperationTestUtil.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/AbfsRestOperationTestUtil.java
@@ -41,7 +41,7 @@ public static void addAbfsHttpOpProcessResponseMock(final AbfsRestOperation spie
spiedRestOp.getMethod(), spiedRestOp.getRequestHeaders());
AbfsHttpOperation spiedOp = Mockito.spy(op);
return functionRaisingIOE.apply(spiedOp, actualOp);
- }).when(spiedRestOp).createNewHttpOperation();
+ }).when(spiedRestOp).createHttpOperation();
}
public static void setResult(final AbfsRestOperation op, final AbfsHttpOperation result) {
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsClient.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/ITestAbfsClient.java
similarity index 62%
rename from hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsClient.java
rename to hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/ITestAbfsClient.java
index 8dfef876561f73..e798a4baa36ab0 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsClient.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/ITestAbfsClient.java
@@ -20,21 +20,44 @@
import java.io.IOException;
import java.lang.reflect.Field;
+import java.net.HttpURLConnection;
+import java.net.ProtocolException;
import java.net.URL;
import java.util.List;
+import java.util.Random;
import java.util.regex.Pattern;
import org.junit.Ignore;
+import org.assertj.core.api.Assertions;
import org.junit.Test;
+import org.mockito.Mockito;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.azurebfs.AbfsConfiguration;
+import org.apache.hadoop.fs.azurebfs.AbstractAbfsIntegrationTest;
+import org.apache.hadoop.fs.azurebfs.AzureBlobFileSystem;
+import org.apache.hadoop.fs.azurebfs.TestAbfsConfigurationFieldsValidation;
+import org.apache.hadoop.fs.azurebfs.constants.FSOperationType;
import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AzureBlobFileSystemException;
+import org.apache.hadoop.fs.azurebfs.contracts.services.AppendRequestParameters;
import org.apache.hadoop.fs.azurebfs.oauth2.AccessTokenProvider;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys;
+import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
+import org.apache.hadoop.fs.azurebfs.utils.TracingHeaderFormat;
import org.apache.hadoop.security.ssl.DelegatingSSLSocketFactory;
-import static org.assertj.core.api.Assertions.assertThat;
+import static java.net.HttpURLConnection.HTTP_NOT_FOUND;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.APPEND_ACTION;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.HTTP_METHOD_PATCH;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.HTTP_METHOD_PUT;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.HUNDRED_CONTINUE;
+import static org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.EXPECT;
+import static org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.X_HTTP_METHOD_OVERRIDE;
+import static org.apache.hadoop.fs.azurebfs.constants.HttpQueryParams.QUERY_PARAM_ACTION;
+import static org.apache.hadoop.fs.azurebfs.constants.HttpQueryParams.QUERY_PARAM_POSITION;
+import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.FS_AZURE_ABFS_ACCOUNT_NAME;
+import static org.apache.hadoop.test.LambdaTestUtils.intercept;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -60,14 +83,19 @@
* Test useragent of abfs client.
*
*/
-public final class TestAbfsClient {
+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 TEST_PATH = "/testfile";
+ public static final int REDUCED_RETRY_COUNT = 2;
+ public static final int REDUCED_BACKOFF_INTERVAL = 100;
+ public static final int BUFFER_LENGTH = 5;
+ public static final int BUFFER_OFFSET = 0;
private final Pattern userAgentStringPattern;
- public TestAbfsClient(){
+ public ITestAbfsClient() throws Exception {
StringBuilder regEx = new StringBuilder();
regEx.append("^");
regEx.append(APN_VERSION);
@@ -125,7 +153,7 @@ public void verifybBasicInfo() throws Exception {
}
private void verifybBasicInfo(String userAgentStr) {
- assertThat(userAgentStr)
+ Assertions.assertThat(userAgentStr)
.describedAs("User-Agent string [" + userAgentStr
+ "] should be of the pattern: " + this.userAgentStringPattern.pattern())
.matches(this.userAgentStringPattern)
@@ -155,7 +183,7 @@ public void verifyUserAgentPrefix()
String userAgentStr = getUserAgentString(abfsConfiguration, false);
verifybBasicInfo(userAgentStr);
- assertThat(userAgentStr)
+ Assertions.assertThat(userAgentStr)
.describedAs("User-Agent string should contain " + FS_AZURE_USER_AGENT_PREFIX)
.contains(FS_AZURE_USER_AGENT_PREFIX);
@@ -165,7 +193,7 @@ public void verifyUserAgentPrefix()
userAgentStr = getUserAgentString(abfsConfiguration, false);
verifybBasicInfo(userAgentStr);
- assertThat(userAgentStr)
+ Assertions.assertThat(userAgentStr)
.describedAs("User-Agent string should not contain " + FS_AZURE_USER_AGENT_PREFIX)
.doesNotContain(FS_AZURE_USER_AGENT_PREFIX);
}
@@ -181,14 +209,14 @@ public void verifyUserAgentWithoutSSLProvider() throws Exception {
String userAgentStr = getUserAgentString(abfsConfiguration, true);
verifybBasicInfo(userAgentStr);
- assertThat(userAgentStr)
+ Assertions.assertThat(userAgentStr)
.describedAs("User-Agent string should contain sslProvider")
.contains(DelegatingSSLSocketFactory.getDefaultFactory().getProviderName());
userAgentStr = getUserAgentString(abfsConfiguration, false);
verifybBasicInfo(userAgentStr);
- assertThat(userAgentStr)
+ Assertions.assertThat(userAgentStr)
.describedAs("User-Agent string should not contain sslProvider")
.doesNotContain(DelegatingSSLSocketFactory.getDefaultFactory().getProviderName());
}
@@ -204,7 +232,7 @@ public void verifyUserAgentClusterName() throws Exception {
String userAgentStr = getUserAgentString(abfsConfiguration, false);
verifybBasicInfo(userAgentStr);
- assertThat(userAgentStr)
+ Assertions.assertThat(userAgentStr)
.describedAs("User-Agent string should contain cluster name")
.contains(clusterName);
@@ -214,7 +242,7 @@ public void verifyUserAgentClusterName() throws Exception {
userAgentStr = getUserAgentString(abfsConfiguration, false);
verifybBasicInfo(userAgentStr);
- assertThat(userAgentStr)
+ Assertions.assertThat(userAgentStr)
.describedAs("User-Agent string should not contain cluster name")
.doesNotContain(clusterName)
.describedAs("User-Agent string should contain UNKNOWN as cluster name config is absent")
@@ -232,7 +260,7 @@ public void verifyUserAgentClusterType() throws Exception {
String userAgentStr = getUserAgentString(abfsConfiguration, false);
verifybBasicInfo(userAgentStr);
- assertThat(userAgentStr)
+ Assertions.assertThat(userAgentStr)
.describedAs("User-Agent string should contain cluster type")
.contains(clusterType);
@@ -242,7 +270,7 @@ public void verifyUserAgentClusterType() throws Exception {
userAgentStr = getUserAgentString(abfsConfiguration, false);
verifybBasicInfo(userAgentStr);
- assertThat(userAgentStr)
+ Assertions.assertThat(userAgentStr)
.describedAs("User-Agent string should not contain cluster type")
.doesNotContain(clusterType)
.describedAs("User-Agent string should contain UNKNOWN as cluster type config is absent")
@@ -308,24 +336,28 @@ public static AbfsClient getMockAbfsClient(AbfsClient baseAbfsClientInstance,
when(client.getAccessToken()).thenCallRealMethod();
when(client.getSharedKeyCredentials()).thenCallRealMethod();
when(client.createDefaultHeaders()).thenCallRealMethod();
-
+ when(client.getAbfsConfiguration()).thenReturn(abfsConfig);
+ when(client.getIntercept()).thenReturn(
+ AbfsThrottlingInterceptFactory.getInstance(
+ abfsConfig.getAccountName().substring(0,
+ abfsConfig.getAccountName().indexOf(DOT)), abfsConfig));
// override baseurl
- client = TestAbfsClient.setAbfsClientField(client, "abfsConfiguration",
+ client = ITestAbfsClient.setAbfsClientField(client, "abfsConfiguration",
abfsConfig);
// override baseurl
- client = TestAbfsClient.setAbfsClientField(client, "baseUrl",
+ client = ITestAbfsClient.setAbfsClientField(client, "baseUrl",
baseAbfsClientInstance.getBaseUrl());
// override auth provider
if (currentAuthType == AuthType.SharedKey) {
- client = TestAbfsClient.setAbfsClientField(client, "sharedKeyCredentials",
+ client = ITestAbfsClient.setAbfsClientField(client, "sharedKeyCredentials",
new SharedKeyCredentials(
abfsConfig.getAccountName().substring(0,
abfsConfig.getAccountName().indexOf(DOT)),
abfsConfig.getStorageAccountKey()));
} else {
- client = TestAbfsClient.setAbfsClientField(client, "tokenProvider",
+ client = ITestAbfsClient.setAbfsClientField(client, "tokenProvider",
abfsConfig.getTokenProvider());
}
@@ -333,7 +365,7 @@ public static AbfsClient getMockAbfsClient(AbfsClient baseAbfsClientInstance,
String userAgent = "APN/1.0 Azure Blob FS/3.4.0-SNAPSHOT (PrivateBuild "
+ "JavaJRE 1.8.0_252; Linux 5.3.0-59-generic/amd64; openssl-1.0; "
+ "UNKNOWN/UNKNOWN) MSFT";
- client = TestAbfsClient.setAbfsClientField(client, "userAgent", userAgent);
+ client = ITestAbfsClient.setAbfsClientField(client, "userAgent", userAgent);
return client;
}
@@ -397,4 +429,160 @@ public static AbfsRestOperation getRestOp(AbfsRestOperationType type,
url,
requestHeaders);
}
+
+ public static AccessTokenProvider getAccessTokenProvider(AbfsClient client) {
+ return client.getTokenProvider();
+ }
+
+ /**
+ * Test helper method to get random bytes array.
+ * @param length The length of byte buffer.
+ * @return byte buffer.
+ */
+ private byte[] getRandomBytesArray(int length) {
+ final byte[] b = new byte[length];
+ new Random().nextBytes(b);
+ return b;
+ }
+
+ /**
+ * Test to verify that client retries append request without
+ * expect header enabled if append with expect header enabled fails
+ * with 4xx kind of error.
+ * @throws Exception
+ */
+ @Test
+ public void testExpectHundredContinue() throws Exception {
+ // Get the filesystem.
+ final AzureBlobFileSystem fs = getFileSystem();
+
+ final Configuration configuration = new Configuration();
+ configuration.addResource(TEST_CONFIGURATION_FILE_NAME);
+ AbfsClient abfsClient = getClient(fs);
+
+ AbfsConfiguration abfsConfiguration = new AbfsConfiguration(configuration,
+ configuration.get(FS_AZURE_ABFS_ACCOUNT_NAME));
+
+ // Update the configuration with reduced retry count and reduced backoff interval.
+ AbfsConfiguration abfsConfig
+ = TestAbfsConfigurationFieldsValidation.updateRetryConfigs(
+ abfsConfiguration,
+ REDUCED_RETRY_COUNT, REDUCED_BACKOFF_INTERVAL);
+
+ // Gets the client.
+ AbfsClient testClient = Mockito.spy(
+ ITestAbfsClient.createTestClientFromCurrentContext(
+ abfsClient,
+ abfsConfig));
+
+ // Create the append request params with expect header enabled initially.
+ AppendRequestParameters appendRequestParameters
+ = new AppendRequestParameters(
+ BUFFER_OFFSET, BUFFER_OFFSET, BUFFER_LENGTH,
+ AppendRequestParameters.Mode.APPEND_MODE, false, null, true);
+
+ byte[] buffer = getRandomBytesArray(BUFFER_LENGTH);
+
+ // Create a test container to upload the data.
+ Path testPath = path(TEST_PATH);
+ fs.create(testPath);
+ String finalTestPath = testPath.toString()
+ .substring(testPath.toString().lastIndexOf("/"));
+
+ // Creates a list of request headers.
+ final List requestHeaders
+ = ITestAbfsClient.getTestRequestHeaders(testClient);
+ requestHeaders.add(
+ new AbfsHttpHeader(X_HTTP_METHOD_OVERRIDE, HTTP_METHOD_PATCH));
+ if (appendRequestParameters.isExpectHeaderEnabled()) {
+ requestHeaders.add(new AbfsHttpHeader(EXPECT, HUNDRED_CONTINUE));
+ }
+
+ // Updates the query parameters.
+ final AbfsUriQueryBuilder abfsUriQueryBuilder
+ = testClient.createDefaultUriQueryBuilder();
+ abfsUriQueryBuilder.addQuery(QUERY_PARAM_ACTION, APPEND_ACTION);
+ abfsUriQueryBuilder.addQuery(QUERY_PARAM_POSITION,
+ Long.toString(appendRequestParameters.getPosition()));
+
+ // Creates the url for the specified path.
+ URL url = testClient.createRequestUrl(finalTestPath, abfsUriQueryBuilder.toString());
+
+ // Create a mock of the AbfsRestOperation to set the urlConnection in the corresponding httpOperation.
+ AbfsRestOperation op = Mockito.spy(new AbfsRestOperation(
+ AbfsRestOperationType.Append,
+ testClient,
+ HTTP_METHOD_PUT,
+ url,
+ requestHeaders, buffer,
+ appendRequestParameters.getoffset(),
+ appendRequestParameters.getLength(), null));
+
+ AbfsHttpOperation abfsHttpOperation = Mockito.spy(new AbfsHttpOperation(url,
+ HTTP_METHOD_PUT, requestHeaders));
+
+ // Sets the expect request property if expect header is enabled.
+ if (appendRequestParameters.isExpectHeaderEnabled()) {
+ Mockito.doReturn(HUNDRED_CONTINUE).when(abfsHttpOperation)
+ .getConnProperty(EXPECT);
+ }
+
+ HttpURLConnection urlConnection = mock(HttpURLConnection.class);
+ Mockito.doNothing().when(urlConnection).setRequestProperty(Mockito
+ .any(), Mockito.any());
+ Mockito.doReturn(HTTP_METHOD_PUT).when(urlConnection).getRequestMethod();
+ Mockito.doReturn(url).when(urlConnection).getURL();
+ Mockito.doReturn(urlConnection).when(abfsHttpOperation).getConnection();
+
+ Mockito.doNothing().when(abfsHttpOperation).setRequestProperty(Mockito
+ .any(), Mockito.any());
+ Mockito.doReturn(url).when(abfsHttpOperation).getConnUrl();
+
+ // Give user error code 404 when processResponse is called.
+ Mockito.doReturn(HTTP_METHOD_PUT).when(abfsHttpOperation).getConnRequestMethod();
+ Mockito.doReturn(HTTP_NOT_FOUND).when(abfsHttpOperation).getConnResponseCode();
+ Mockito.doReturn("Resource Not Found")
+ .when(abfsHttpOperation)
+ .getConnResponseMessage();
+
+ // Make the getOutputStream throw IOException to see it returns from the sendRequest correctly.
+ Mockito.doThrow(new ProtocolException("Server rejected Operation"))
+ .when(abfsHttpOperation)
+ .getConnOutputStream();
+
+ // Sets the httpOperation for the rest operation.
+ Mockito.doReturn(abfsHttpOperation)
+ .when(op)
+ .createHttpOperation();
+
+ // Mock the restOperation for the client.
+ Mockito.doReturn(op)
+ .when(testClient)
+ .getAbfsRestOperationForAppend(Mockito.any(),
+ Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(),
+ Mockito.nullable(int.class), Mockito.nullable(int.class),
+ Mockito.any());
+
+ TracingContext tracingContext = Mockito.spy(new TracingContext("abcd",
+ "abcde", FSOperationType.APPEND,
+ TracingHeaderFormat.ALL_ID_FORMAT, null));
+
+ // Check that expect header is enabled before the append call.
+ Assertions.assertThat(appendRequestParameters.isExpectHeaderEnabled())
+ .describedAs("The expect header is not true before the append call")
+ .isTrue();
+
+ intercept(AzureBlobFileSystemException.class,
+ () -> testClient.append(finalTestPath, buffer, appendRequestParameters, null, tracingContext));
+
+ // Verify that the request was not exponentially retried because of user error.
+ Assertions.assertThat(tracingContext.getRetryCount())
+ .describedAs("The retry count is incorrect")
+ .isEqualTo(0);
+
+ // Verify that the same request was retried with expect header disabled.
+ Assertions.assertThat(appendRequestParameters.isExpectHeaderEnabled())
+ .describedAs("The expect header is not false")
+ .isFalse();
+ }
}
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/ITestAbfsRestOperation.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/ITestAbfsRestOperation.java
new file mode 100644
index 00000000000000..fe3c2a9892c4c6
--- /dev/null
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/ITestAbfsRestOperation.java
@@ -0,0 +1,358 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.services;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.ProtocolException;
+import java.net.URL;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Random;
+
+import org.assertj.core.api.Assertions;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.mockito.Mockito;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.azurebfs.AbfsConfiguration;
+import org.apache.hadoop.fs.azurebfs.AbstractAbfsIntegrationTest;
+import org.apache.hadoop.fs.azurebfs.AzureBlobFileSystem;
+import org.apache.hadoop.fs.azurebfs.TestAbfsConfigurationFieldsValidation;
+import org.apache.hadoop.fs.azurebfs.constants.FSOperationType;
+import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AzureBlobFileSystemException;
+import org.apache.hadoop.fs.azurebfs.contracts.services.AppendRequestParameters;
+import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
+import org.apache.hadoop.fs.azurebfs.utils.TracingHeaderFormat;
+
+import static java.net.HttpURLConnection.HTTP_NOT_FOUND;
+import static java.net.HttpURLConnection.HTTP_OK;
+import static java.net.HttpURLConnection.HTTP_UNAVAILABLE;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.APPEND_ACTION;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.HTTP_METHOD_PATCH;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.HTTP_METHOD_PUT;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.HUNDRED_CONTINUE;
+import static org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.EXPECT;
+import static org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.X_HTTP_METHOD_OVERRIDE;
+import static org.apache.hadoop.fs.azurebfs.constants.HttpQueryParams.QUERY_PARAM_ACTION;
+import static org.apache.hadoop.fs.azurebfs.constants.HttpQueryParams.QUERY_PARAM_POSITION;
+import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.FS_AZURE_ABFS_ACCOUNT_NAME;
+import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.TEST_CONFIGURATION_FILE_NAME;
+import static org.apache.hadoop.test.LambdaTestUtils.intercept;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+
+@RunWith(Parameterized.class)
+public class ITestAbfsRestOperation extends AbstractAbfsIntegrationTest {
+
+ // Specifies whether getOutputStream() or write() throws IOException.
+ public enum ErrorType {OUTPUTSTREAM, WRITE};
+
+ private static final int HTTP_EXPECTATION_FAILED = 417;
+ private static final int HTTP_ERROR = 0;
+ private static final int ZERO = 0;
+ private static final int REDUCED_RETRY_COUNT = 2;
+ private static final int REDUCED_BACKOFF_INTERVAL = 100;
+ private static final int BUFFER_LENGTH = 5;
+ private static final int BUFFER_OFFSET = 0;
+ private static final String TEST_PATH = "/testfile";
+
+ // Specifies whether the expect header is enabled or not.
+ @Parameterized.Parameter
+ public boolean expectHeaderEnabled;
+
+ // Gives the http response code.
+ @Parameterized.Parameter(1)
+ public int responseCode;
+
+ // Gives the http response message.
+ @Parameterized.Parameter(2)
+ public String responseMessage;
+
+ // Gives the errorType based on the enum.
+ @Parameterized.Parameter(3)
+ public ErrorType errorType;
+
+ // The intercept.
+ private AbfsThrottlingIntercept intercept;
+
+ /*
+ HTTP_OK = 200,
+ HTTP_UNAVAILABLE = 503,
+ HTTP_NOT_FOUND = 404,
+ HTTP_EXPECTATION_FAILED = 417,
+ HTTP_ERROR = 0.
+ */
+ @Parameterized.Parameters(name = "expect={0}-code={1}-ErrorType={3}")
+ public static Iterable params() {
+ return Arrays.asList(new Object[][]{
+ {true, HTTP_OK, "OK", ErrorType.WRITE},
+ {false, HTTP_OK, "OK", ErrorType.WRITE},
+ {true, HTTP_UNAVAILABLE, "ServerBusy", ErrorType.OUTPUTSTREAM},
+ {true, HTTP_NOT_FOUND, "Resource Not Found", ErrorType.OUTPUTSTREAM},
+ {true, HTTP_EXPECTATION_FAILED, "Expectation Failed", ErrorType.OUTPUTSTREAM},
+ {true, HTTP_ERROR, "Error", ErrorType.OUTPUTSTREAM}
+ });
+ }
+
+ public ITestAbfsRestOperation() throws Exception {
+ super();
+ }
+
+ /**
+ * Test helper method to get random bytes array.
+ * @param length The length of byte buffer
+ * @return byte buffer
+ */
+ private byte[] getRandomBytesArray(int length) {
+ final byte[] b = new byte[length];
+ new Random().nextBytes(b);
+ return b;
+ }
+
+ /**
+ * Gives the AbfsRestOperation.
+ * @return abfsRestOperation.
+ */
+ private AbfsRestOperation getRestOperation() throws Exception {
+ // Get the filesystem.
+ final AzureBlobFileSystem fs = getFileSystem();
+
+ final Configuration configuration = new Configuration();
+ configuration.addResource(TEST_CONFIGURATION_FILE_NAME);
+ AbfsClient abfsClient = getClient(fs);
+
+ AbfsConfiguration abfsConfiguration = new AbfsConfiguration(configuration,
+ configuration.get(FS_AZURE_ABFS_ACCOUNT_NAME));
+
+ // Update the configuration with reduced retry count and reduced backoff interval.
+ AbfsConfiguration abfsConfig
+ = TestAbfsConfigurationFieldsValidation.updateRetryConfigs(
+ abfsConfiguration,
+ REDUCED_RETRY_COUNT, REDUCED_BACKOFF_INTERVAL);
+
+ intercept = Mockito.mock(AbfsThrottlingIntercept.class);
+ Mockito.doNothing().when(intercept).updateMetrics(Mockito.any(), Mockito.any());
+
+ // Gets the client.
+ AbfsClient testClient = Mockito.spy(ITestAbfsClient.createTestClientFromCurrentContext(
+ abfsClient,
+ abfsConfig));
+
+ Mockito.doReturn(intercept).when(testClient).getIntercept();
+
+ // Expect header is enabled or not based on the parameter.
+ AppendRequestParameters appendRequestParameters
+ = new AppendRequestParameters(
+ BUFFER_OFFSET, BUFFER_OFFSET, BUFFER_LENGTH,
+ AppendRequestParameters.Mode.APPEND_MODE, false, null,
+ expectHeaderEnabled);
+
+ byte[] buffer = getRandomBytesArray(5);
+
+ // Create a test container to upload the data.
+ Path testPath = path(TEST_PATH);
+ fs.create(testPath);
+ String finalTestPath = testPath.toString().substring(testPath.toString().lastIndexOf("/"));
+
+ // Creates a list of request headers.
+ final List requestHeaders = ITestAbfsClient.getTestRequestHeaders(testClient);
+ requestHeaders.add(new AbfsHttpHeader(X_HTTP_METHOD_OVERRIDE, HTTP_METHOD_PATCH));
+ if (appendRequestParameters.isExpectHeaderEnabled()) {
+ requestHeaders.add(new AbfsHttpHeader(EXPECT, HUNDRED_CONTINUE));
+ }
+
+ // Updates the query parameters.
+ final AbfsUriQueryBuilder abfsUriQueryBuilder = testClient.createDefaultUriQueryBuilder();
+ abfsUriQueryBuilder.addQuery(QUERY_PARAM_ACTION, APPEND_ACTION);
+ abfsUriQueryBuilder.addQuery(QUERY_PARAM_POSITION, Long.toString(appendRequestParameters.getPosition()));
+
+ // Creates the url for the specified path.
+ URL url = testClient.createRequestUrl(finalTestPath, abfsUriQueryBuilder.toString());
+
+ // Create a mock of the AbfsRestOperation to set the urlConnection in the corresponding httpOperation.
+ AbfsRestOperation op = Mockito.spy(new AbfsRestOperation(
+ AbfsRestOperationType.Append,
+ testClient,
+ HTTP_METHOD_PUT,
+ url,
+ requestHeaders, buffer,
+ appendRequestParameters.getoffset(),
+ appendRequestParameters.getLength(), null));
+
+ AbfsHttpOperation abfsHttpOperation = Mockito.spy(new AbfsHttpOperation(url, HTTP_METHOD_PUT, requestHeaders));
+
+ // Sets the expect request property if expect header is enabled.
+ if (expectHeaderEnabled) {
+ Mockito.doReturn(HUNDRED_CONTINUE)
+ .when(abfsHttpOperation)
+ .getConnProperty(EXPECT);
+ }
+
+ HttpURLConnection urlConnection = mock(HttpURLConnection.class);
+ Mockito.doNothing().when(urlConnection).setRequestProperty(Mockito
+ .any(), Mockito.any());
+ Mockito.doReturn(HTTP_METHOD_PUT).when(urlConnection).getRequestMethod();
+ Mockito.doReturn(url).when(urlConnection).getURL();
+ Mockito.doReturn(urlConnection).when(abfsHttpOperation).getConnection();
+
+ Mockito.doNothing().when(abfsHttpOperation).setRequestProperty(Mockito
+ .any(), Mockito.any());
+ Mockito.doReturn(url).when(abfsHttpOperation).getConnUrl();
+ Mockito.doReturn(HTTP_METHOD_PUT).when(abfsHttpOperation).getConnRequestMethod();
+
+ switch (errorType) {
+ case OUTPUTSTREAM:
+ // If the getOutputStream() throws IOException and Expect Header is
+ // enabled, it returns back to processResponse and hence we have
+ // mocked the response code and the response message to check different
+ // behaviour based on response code.
+ Mockito.doReturn(responseCode).when(abfsHttpOperation).getConnResponseCode();
+ Mockito.doReturn(responseMessage)
+ .when(abfsHttpOperation)
+ .getConnResponseMessage();
+ Mockito.doThrow(new ProtocolException("Server rejected Operation"))
+ .when(abfsHttpOperation)
+ .getConnOutputStream();
+ break;
+ case WRITE:
+ // If write() throws IOException and Expect Header is
+ // enabled or not, it should throw back the exception.
+ OutputStream outputStream = Mockito.spy(new OutputStream() {
+ @Override
+ public void write(final int i) throws IOException {
+ }
+ });
+ Mockito.doReturn(outputStream).when(abfsHttpOperation).getConnOutputStream();
+ Mockito.doThrow(new IOException())
+ .when(outputStream)
+ .write(buffer, appendRequestParameters.getoffset(),
+ appendRequestParameters.getLength());
+ break;
+ default:
+ break;
+ }
+
+ // Sets the httpOperation for the rest operation.
+ Mockito.doReturn(abfsHttpOperation)
+ .when(op)
+ .createHttpOperation();
+ return op;
+ }
+
+ void assertTraceContextState(int retryCount, int assertRetryCount, int bytesSent, int assertBytesSent,
+ int expectedBytesSent, int assertExpectedBytesSent) {
+ // Assert that the request is retried or not.
+ Assertions.assertThat(retryCount)
+ .describedAs("The retry count is incorrect")
+ .isEqualTo(assertRetryCount);
+
+ // Assert that metrics will be updated correctly.
+ Assertions.assertThat(bytesSent)
+ .describedAs("The bytes sent is incorrect")
+ .isEqualTo(assertBytesSent);
+ Assertions.assertThat(expectedBytesSent)
+ .describedAs("The expected bytes sent is incorrect")
+ .isEqualTo(assertExpectedBytesSent);
+ }
+
+ /**
+ * Test the functionalities based on whether getOutputStream() or write()
+ * throws exception and what is the corresponding response code.
+ */
+ @Test
+ public void testExpectHundredContinue() throws Exception {
+ // Gets the AbfsRestOperation.
+ AbfsRestOperation op = getRestOperation();
+ AbfsHttpOperation httpOperation = op.createHttpOperation();
+
+ TracingContext tracingContext = Mockito.spy(new TracingContext("abcd",
+ "abcde", FSOperationType.APPEND,
+ TracingHeaderFormat.ALL_ID_FORMAT, null));
+
+ switch (errorType) {
+ case WRITE:
+ // If write() throws IOException and Expect Header is
+ // enabled or not, it should throw back the exception
+ // which is caught and exponential retry logic comes into place.
+ intercept(IOException.class,
+ () -> op.execute(tracingContext));
+
+ // Asserting update of metrics and retries.
+ assertTraceContextState(tracingContext.getRetryCount(), REDUCED_RETRY_COUNT, httpOperation.getBytesSent(), BUFFER_LENGTH,
+ 0, 0);
+ break;
+ case OUTPUTSTREAM:
+ switch (responseCode) {
+ case HTTP_UNAVAILABLE:
+ // In the case of 503 i.e. throttled case, we should retry.
+ intercept(IOException.class,
+ () -> op.execute(tracingContext));
+
+ // Asserting update of metrics and retries.
+ assertTraceContextState(tracingContext.getRetryCount(), REDUCED_RETRY_COUNT, httpOperation.getBytesSent(), ZERO,
+ httpOperation.getExpectedBytesToBeSent(), BUFFER_LENGTH);
+
+ // Verifies that update Metrics call is made for throttle case and for the first without retry +
+ // for the retried cases as well.
+ Mockito.verify(intercept, times(REDUCED_RETRY_COUNT + 1))
+ .updateMetrics(Mockito.any(), Mockito.any());
+ break;
+ case HTTP_ERROR:
+ // In the case of http status code 0 i.e. ErrorType case, we should retry.
+ intercept(IOException.class,
+ () -> op.execute(tracingContext));
+
+ // Asserting update of metrics and retries.
+ assertTraceContextState(tracingContext.getRetryCount(), REDUCED_RETRY_COUNT, httpOperation.getBytesSent(),
+ ZERO, 0, 0);
+
+ // Verifies that update Metrics call is made for ErrorType case and for the first without retry +
+ // for the retried cases as well.
+ Mockito.verify(intercept, times(REDUCED_RETRY_COUNT + 1))
+ .updateMetrics(Mockito.any(), Mockito.any());
+ break;
+ case HTTP_NOT_FOUND:
+ case HTTP_EXPECTATION_FAILED:
+ // In the case of 4xx ErrorType. i.e. user ErrorType, retry should not happen.
+ intercept(AzureBlobFileSystemException.class,
+ () -> op.execute(tracingContext));
+
+ // Asserting update of metrics and retries.
+ assertTraceContextState(tracingContext.getRetryCount(), ZERO, 0,
+ 0, 0, 0);
+
+ // Verifies that update Metrics call is not made for user ErrorType case.
+ Mockito.verify(intercept, never())
+ .updateMetrics(Mockito.any(), Mockito.any());
+ break;
+ default:
+ break;
+ }
+ break;
+ default:
+ break;
+ }
+ }
+}
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/ITestBlobOperation.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/ITestBlobOperation.java
index 5530fb0e1b9cd9..a0b116ab0593a2 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/ITestBlobOperation.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/ITestBlobOperation.java
@@ -92,7 +92,7 @@ public void testPutBlob() throws Exception {
configuration.get(FS_AZURE_ABFS_ACCOUNT_NAME));
// Gets the client.
- AbfsClient testClient = Mockito.spy(TestAbfsClient.createTestClientFromCurrentContext(
+ AbfsClient testClient = Mockito.spy(ITestAbfsClient.createTestClientFromCurrentContext(
abfsClient,
abfsConfiguration));
@@ -103,7 +103,7 @@ public void testPutBlob() throws Exception {
String finalTestPath = testPath.toString().substring(testPath.toString().lastIndexOf("/"));
// Creates a list of request headers.
- final List requestHeaders = TestAbfsClient.getTestRequestHeaders(testClient);
+ final List requestHeaders = ITestAbfsClient.getTestRequestHeaders(testClient);
requestHeaders.add(new AbfsHttpHeader(CONTENT_LENGTH, String.valueOf(buffer.length)));
requestHeaders.add(new AbfsHttpHeader(X_MS_BLOB_TYPE, BLOCK_BLOB_TYPE));
String ContentMD5 = computeMd5(buffer);
@@ -160,7 +160,7 @@ public void testPutBlockWithNullData() throws Exception {
AbfsConfiguration abfsConfiguration = new AbfsConfiguration(configuration,
configuration.get(FS_AZURE_ABFS_ACCOUNT_NAME));
- AbfsClient testClient = Mockito.spy(TestAbfsClient.createTestClientFromCurrentContext(
+ AbfsClient testClient = Mockito.spy(ITestAbfsClient.createTestClientFromCurrentContext(
abfsClient,
abfsConfiguration));
@@ -168,7 +168,7 @@ public void testPutBlockWithNullData() throws Exception {
byte[] data = null;
Path testPath = path(TEST_PATH);
String finalTestPath = testPath.toString().substring(testPath.toString().lastIndexOf("/"));
- final List requestHeaders = TestAbfsClient.getTestRequestHeaders(testClient);
+ final List requestHeaders = ITestAbfsClient.getTestRequestHeaders(testClient);
final AbfsUriQueryBuilder abfsUriQueryBuilder = testClient.createDefaultUriQueryBuilder();
abfsUriQueryBuilder.addQuery(QUERY_PARAM_COMP, BLOCK);
@@ -211,13 +211,13 @@ public void testPutBlockWithDifferentLengthBlockIds() throws Exception {
"world".getBytes(),
"!".getBytes()
));
- AbfsClient testClient = Mockito.spy(TestAbfsClient.createTestClientFromCurrentContext(
+ AbfsClient testClient = Mockito.spy(ITestAbfsClient.createTestClientFromCurrentContext(
abfsClient,
abfsConfiguration));
Path testPath = path(TEST_PATH);
String finalTestPath = testPath.toString().substring(testPath.toString().lastIndexOf("/"));
- final List requestHeaders = TestAbfsClient.getTestRequestHeaders(testClient);
+ final List requestHeaders = ITestAbfsClient.getTestRequestHeaders(testClient);
final AbfsUriQueryBuilder abfsUriQueryBuilder = testClient.createDefaultUriQueryBuilder();
abfsUriQueryBuilder.addQuery(QUERY_PARAM_COMP, BLOCK);
@@ -273,12 +273,12 @@ public void testGetCommittedBlockList() throws IOException, IllegalAccessExcepti
"world".getBytes(),
"!".getBytes()
));
- AbfsClient testClient = Mockito.spy(TestAbfsClient.createTestClientFromCurrentContext(
+ AbfsClient testClient = Mockito.spy(ITestAbfsClient.createTestClientFromCurrentContext(
abfsClient,
abfsConfiguration));
Path testPath = path(TEST_PATH);
String finalTestPath = testPath.toString().substring(testPath.toString().lastIndexOf("/"));
- final List requestHeaders = TestAbfsClient.getTestRequestHeaders(testClient);
+ final List requestHeaders = ITestAbfsClient.getTestRequestHeaders(testClient);
final AbfsUriQueryBuilder abfsUriQueryBuilder = testClient.createDefaultUriQueryBuilder();
abfsUriQueryBuilder.addQuery(QUERY_PARAM_COMP, BLOCK);
List encodedBlockIds = new ArrayList<>();
@@ -306,7 +306,7 @@ public void testGetCommittedBlockList() throws IOException, IllegalAccessExcepti
}
byte[] bufferString = generateBlockListXml(blockIds).getBytes(StandardCharsets.UTF_8);
final AbfsUriQueryBuilder abfsUriQueryBuilder1 = testClient.createDefaultUriQueryBuilder();
- final List requestHeaders1 = TestAbfsClient.getTestRequestHeaders(testClient);
+ final List requestHeaders1 = ITestAbfsClient.getTestRequestHeaders(testClient);
abfsUriQueryBuilder1.addQuery(QUERY_PARAM_COMP, BLOCKLIST);
requestHeaders1.add(new AbfsHttpHeader(CONTENT_LENGTH, String.valueOf(bufferString.length)));
requestHeaders1.add(new AbfsHttpHeader(CONTENT_TYPE, "application/xml"));
@@ -351,12 +351,12 @@ public void testPutBlockListForAdditionalBlockId() throws Exception {
"world".getBytes(),
"!".getBytes()
));
- AbfsClient testClient = Mockito.spy(TestAbfsClient.createTestClientFromCurrentContext(
+ AbfsClient testClient = Mockito.spy(ITestAbfsClient.createTestClientFromCurrentContext(
abfsClient,
abfsConfiguration));
Path testPath = path(TEST_PATH);
String finalTestPath = testPath.toString().substring(testPath.toString().lastIndexOf("/"));
- final List requestHeaders = TestAbfsClient.getTestRequestHeaders(testClient);
+ final List requestHeaders = ITestAbfsClient.getTestRequestHeaders(testClient);
final AbfsUriQueryBuilder abfsUriQueryBuilder = testClient.createDefaultUriQueryBuilder();
abfsUriQueryBuilder.addQuery(QUERY_PARAM_COMP, BLOCK);
@@ -383,7 +383,7 @@ public void testPutBlockListForAdditionalBlockId() throws Exception {
}
byte[] bufferString = generateBlockListXml(blockIds).getBytes(StandardCharsets.UTF_8);
final AbfsUriQueryBuilder abfsUriQueryBuilder1 = testClient.createDefaultUriQueryBuilder();
- final List requestHeaders1 = TestAbfsClient.getTestRequestHeaders(testClient);
+ final List requestHeaders1 = ITestAbfsClient.getTestRequestHeaders(testClient);
abfsUriQueryBuilder1.addQuery(QUERY_PARAM_COMP, BLOCKLIST);
requestHeaders1.add(new AbfsHttpHeader(CONTENT_LENGTH, String.valueOf(bufferString.length)));
requestHeaders1.add(new AbfsHttpHeader(CONTENT_TYPE, "application/xml"));
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsClientThrottlingAnalyzer.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsClientThrottlingAnalyzer.java
index 3f680e499300dd..bf3bd203a62ed9 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsClientThrottlingAnalyzer.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsClientThrottlingAnalyzer.java
@@ -18,9 +18,15 @@
package org.apache.hadoop.fs.azurebfs.services;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.azurebfs.AbfsConfiguration;
import org.apache.hadoop.fs.contract.ContractTestUtils;
import org.junit.Test;
+import java.io.IOException;
+
+import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.FS_AZURE_ANALYSIS_PERIOD;
+import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.TEST_CONFIGURATION_FILE_NAME;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -33,6 +39,15 @@ public class TestAbfsClientThrottlingAnalyzer {
+ ANALYSIS_PERIOD / 10;
private static final long MEGABYTE = 1024 * 1024;
private static final int MAX_ACCEPTABLE_PERCENT_DIFFERENCE = 20;
+ private AbfsConfiguration abfsConfiguration;
+
+ public TestAbfsClientThrottlingAnalyzer() throws IOException, IllegalAccessException {
+ final Configuration configuration = new Configuration();
+ configuration.addResource(TEST_CONFIGURATION_FILE_NAME);
+ configuration.setInt(FS_AZURE_ANALYSIS_PERIOD, 1000);
+ this.abfsConfiguration = new AbfsConfiguration(configuration,
+ "dummy");
+ }
private void sleep(long milliseconds) {
try {
@@ -82,8 +97,7 @@ private void validateLessThanOrEqual(long maxExpected, long actual) {
@Test
public void testNoMetricUpdatesThenNoWaiting() {
AbfsClientThrottlingAnalyzer analyzer = new AbfsClientThrottlingAnalyzer(
- "test",
- ANALYSIS_PERIOD);
+ "test", abfsConfiguration);
validate(0, analyzer.getSleepDuration());
sleep(ANALYSIS_PERIOD_PLUS_10_PERCENT);
validate(0, analyzer.getSleepDuration());
@@ -96,8 +110,7 @@ public void testNoMetricUpdatesThenNoWaiting() {
@Test
public void testOnlySuccessThenNoWaiting() {
AbfsClientThrottlingAnalyzer analyzer = new AbfsClientThrottlingAnalyzer(
- "test",
- ANALYSIS_PERIOD);
+ "test", abfsConfiguration);
analyzer.addBytesTransferred(8 * MEGABYTE, false);
validate(0, analyzer.getSleepDuration());
sleep(ANALYSIS_PERIOD_PLUS_10_PERCENT);
@@ -112,8 +125,7 @@ public void testOnlySuccessThenNoWaiting() {
@Test
public void testOnlyErrorsAndWaiting() {
AbfsClientThrottlingAnalyzer analyzer = new AbfsClientThrottlingAnalyzer(
- "test",
- ANALYSIS_PERIOD);
+ "test", abfsConfiguration);
validate(0, analyzer.getSleepDuration());
analyzer.addBytesTransferred(4 * MEGABYTE, true);
sleep(ANALYSIS_PERIOD_PLUS_10_PERCENT);
@@ -132,8 +144,7 @@ public void testOnlyErrorsAndWaiting() {
@Test
public void testSuccessAndErrorsAndWaiting() {
AbfsClientThrottlingAnalyzer analyzer = new AbfsClientThrottlingAnalyzer(
- "test",
- ANALYSIS_PERIOD);
+ "test", abfsConfiguration);
validate(0, analyzer.getSleepDuration());
analyzer.addBytesTransferred(8 * MEGABYTE, false);
analyzer.addBytesTransferred(2 * MEGABYTE, true);
@@ -157,8 +168,7 @@ public void testSuccessAndErrorsAndWaiting() {
@Test
public void testManySuccessAndErrorsAndWaiting() {
AbfsClientThrottlingAnalyzer analyzer = new AbfsClientThrottlingAnalyzer(
- "test",
- ANALYSIS_PERIOD);
+ "test", abfsConfiguration);
validate(0, analyzer.getSleepDuration());
final int numberOfRequests = 20;
for (int i = 0; i < numberOfRequests; i++) {
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsOutputStream.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsOutputStream.java
index 7d1a44b038642d..5c4f24362381b6 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsOutputStream.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsOutputStream.java
@@ -72,6 +72,7 @@ private AbfsOutputStreamContext populateAbfsOutputStreamContext(
boolean isFlushEnabled,
boolean disableOutputStreamFlush,
boolean isAppendBlob,
+ boolean isExpectHeaderEnabled,
AbfsClient client,
String path,
TracingContext tracingContext,
@@ -89,6 +90,7 @@ private AbfsOutputStreamContext populateAbfsOutputStreamContext(
return new AbfsOutputStreamContext(2)
.withWriteBufferSize(writeBufferSize)
+ .enableExpectHeader(isExpectHeaderEnabled)
.enableFlush(isFlushEnabled)
.disableOutputStreamFlush(disableOutputStreamFlush)
.withStreamStatistics(new AbfsOutputStreamStatisticsImpl())
@@ -137,6 +139,7 @@ public void verifyShortWriteRequest() throws Exception {
true,
false,
false,
+ true,
client,
PATH,
new TracingContext(abfsConf.getClientCorrelationId(), "test-fs-id",
@@ -157,9 +160,9 @@ public void verifyShortWriteRequest() throws Exception {
out.hsync();
AppendRequestParameters firstReqParameters = new AppendRequestParameters(
- 0, 0, WRITE_SIZE, APPEND_MODE, false, null);
+ 0, 0, WRITE_SIZE, APPEND_MODE, false, null, true);
AppendRequestParameters secondReqParameters = new AppendRequestParameters(
- WRITE_SIZE, 0, 2 * WRITE_SIZE, APPEND_MODE, false, null);
+ WRITE_SIZE, 0, 2 * WRITE_SIZE, APPEND_MODE, false, null, true);
verify(client, times(1)).append(
eq(PATH), any(byte[].class), refEq(firstReqParameters), any(),
@@ -196,6 +199,7 @@ public void verifyWriteRequest() throws Exception {
true,
false,
false,
+ true,
client,
PATH,
tracingContext,
@@ -209,9 +213,9 @@ public void verifyWriteRequest() throws Exception {
out.close();
AppendRequestParameters firstReqParameters = new AppendRequestParameters(
- 0, 0, BUFFER_SIZE, APPEND_MODE, false, null);
+ 0, 0, BUFFER_SIZE, APPEND_MODE, false, null, true);
AppendRequestParameters secondReqParameters = new AppendRequestParameters(
- BUFFER_SIZE, 0, 5*WRITE_SIZE-BUFFER_SIZE, APPEND_MODE, false, null);
+ BUFFER_SIZE, 0, 5*WRITE_SIZE-BUFFER_SIZE, APPEND_MODE, false, null, true);
verify(client, times(1)).append(
eq(PATH), any(byte[].class), refEq(firstReqParameters), any(),
@@ -269,6 +273,7 @@ public void verifyWriteRequestOfBufferSizeAndClose() throws Exception {
true,
false,
false,
+ true,
client,
PATH,
tracingContext,
@@ -282,9 +287,9 @@ public void verifyWriteRequestOfBufferSizeAndClose() throws Exception {
out.close();
AppendRequestParameters firstReqParameters = new AppendRequestParameters(
- 0, 0, BUFFER_SIZE, APPEND_MODE, false, null);
+ 0, 0, BUFFER_SIZE, APPEND_MODE, false, null, true);
AppendRequestParameters secondReqParameters = new AppendRequestParameters(
- BUFFER_SIZE, 0, BUFFER_SIZE, APPEND_MODE, false, null);
+ BUFFER_SIZE, 0, BUFFER_SIZE, APPEND_MODE, false, null, true);
verify(client, times(1)).append(
eq(PATH), any(byte[].class), refEq(firstReqParameters), any(), any(TracingContext.class));
@@ -339,6 +344,7 @@ public void verifyWriteRequestOfBufferSize() throws Exception {
true,
false,
false,
+ true,
client,
PATH,
new TracingContext(abfsConf.getClientCorrelationId(), "test-fs-id",
@@ -354,9 +360,9 @@ public void verifyWriteRequestOfBufferSize() throws Exception {
Thread.sleep(1000);
AppendRequestParameters firstReqParameters = new AppendRequestParameters(
- 0, 0, BUFFER_SIZE, APPEND_MODE, false, null);
+ 0, 0, BUFFER_SIZE, APPEND_MODE, false, null, true);
AppendRequestParameters secondReqParameters = new AppendRequestParameters(
- BUFFER_SIZE, 0, BUFFER_SIZE, APPEND_MODE, false, null);
+ BUFFER_SIZE, 0, BUFFER_SIZE, APPEND_MODE, false, null, true);
verify(client, times(1)).append(
eq(PATH), any(byte[].class), refEq(firstReqParameters), any(), any(TracingContext.class));
@@ -392,6 +398,7 @@ public void verifyWriteRequestOfBufferSizeWithAppendBlob() throws Exception {
true,
false,
true,
+ true,
client,
PATH,
new TracingContext(abfsConf.getClientCorrelationId(), "test-fs-id",
@@ -407,9 +414,9 @@ public void verifyWriteRequestOfBufferSizeWithAppendBlob() throws Exception {
Thread.sleep(1000);
AppendRequestParameters firstReqParameters = new AppendRequestParameters(
- 0, 0, BUFFER_SIZE, APPEND_MODE, true, null);
+ 0, 0, BUFFER_SIZE, APPEND_MODE, true, null, true);
AppendRequestParameters secondReqParameters = new AppendRequestParameters(
- BUFFER_SIZE, 0, BUFFER_SIZE, APPEND_MODE, true, null);
+ BUFFER_SIZE, 0, BUFFER_SIZE, APPEND_MODE, true, null, true);
verify(client, times(1)).append(
eq(PATH), any(byte[].class), refEq(firstReqParameters), any(), any(TracingContext.class));
@@ -453,6 +460,7 @@ public void verifyWriteRequestOfBufferSizeAndHFlush() throws Exception {
true,
false,
false,
+ true,
client,
PATH,
new TracingContext(abfsConf.getClientCorrelationId(), "test-fs-id",
@@ -468,9 +476,9 @@ public void verifyWriteRequestOfBufferSizeAndHFlush() throws Exception {
out.hflush();
AppendRequestParameters firstReqParameters = new AppendRequestParameters(
- 0, 0, BUFFER_SIZE, APPEND_MODE, false, null);
+ 0, 0, BUFFER_SIZE, APPEND_MODE, false, null, true);
AppendRequestParameters secondReqParameters = new AppendRequestParameters(
- BUFFER_SIZE, 0, BUFFER_SIZE, APPEND_MODE, false, null);
+ BUFFER_SIZE, 0, BUFFER_SIZE, APPEND_MODE, false, null, true);
verify(client, times(1)).append(
eq(PATH), any(byte[].class), refEq(firstReqParameters), any(), any(TracingContext.class));
@@ -521,6 +529,7 @@ public void verifyWriteRequestOfBufferSizeAndFlush() throws Exception {
true,
false,
false,
+ true,
client,
PATH,
new TracingContext(abfsConf.getClientCorrelationId(), "test-fs-id",
@@ -538,9 +547,9 @@ public void verifyWriteRequestOfBufferSizeAndFlush() throws Exception {
Thread.sleep(1000);
AppendRequestParameters firstReqParameters = new AppendRequestParameters(
- 0, 0, BUFFER_SIZE, APPEND_MODE, false, null);
+ 0, 0, BUFFER_SIZE, APPEND_MODE, false, null, true);
AppendRequestParameters secondReqParameters = new AppendRequestParameters(
- BUFFER_SIZE, 0, BUFFER_SIZE, APPEND_MODE, false, null);
+ BUFFER_SIZE, 0, BUFFER_SIZE, APPEND_MODE, false, null, true);
verify(client, times(1)).append(
eq(PATH), any(byte[].class), refEq(firstReqParameters), any(), any(TracingContext.class));
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsOutputStreamBlob.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsOutputStreamBlob.java
index 90e4aaa25bbd42..e6ea4e0d43b9fd 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsOutputStreamBlob.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsOutputStreamBlob.java
@@ -68,6 +68,7 @@ private AbfsOutputStreamContext populateAbfsOutputStreamContext(
boolean isFlushEnabled,
boolean disableOutputStreamFlush,
boolean isAppendBlob,
+ boolean isExpectHeaderEnabled,
AbfsClient client,
String path,
TracingContext tracingContext,
@@ -85,6 +86,7 @@ private AbfsOutputStreamContext populateAbfsOutputStreamContext(
return new AbfsOutputStreamContext(2)
.withWriteBufferSize(writeBufferSize)
+ .enableExpectHeader(isExpectHeaderEnabled)
.enableFlush(isFlushEnabled)
.disableOutputStreamFlush(disableOutputStreamFlush)
.withStreamStatistics(new AbfsOutputStreamStatisticsImpl())
@@ -134,18 +136,19 @@ public AbfsClient getClient() throws IOException, IllegalAccessException {
}
public AbfsOutputStream getOutputStream(AbfsClient client, AbfsConfiguration abfsConf) throws IOException, IllegalAccessException {
- AbfsOutputStream out = Mockito.spy(new AbfsOutputStream(
+ AbfsOutputStream out = new AbfsOutputStream(
populateAbfsOutputStreamContext(
BUFFER_SIZE,
true,
false,
false,
+ true,
client,
PATH,
new TracingContext(abfsConf.getClientCorrelationId(), "test-fs-id",
FSOperationType.WRITE, abfsConf.getTracingHeaderFormat(),
null),
- createExecutorService(abfsConf))));
+ createExecutorService(abfsConf)));
return out;
}
@@ -172,9 +175,9 @@ public void verifyShortWriteRequest() throws Exception {
out.hsync();
AppendRequestParameters firstReqParameters = new AppendRequestParameters(
- 0, 0, WRITE_SIZE, APPEND_MODE, false, null);
+ 0, 0, WRITE_SIZE, APPEND_MODE, false, null, true);
AppendRequestParameters secondReqParameters = new AppendRequestParameters(
- WRITE_SIZE, 0, 2 * WRITE_SIZE, APPEND_MODE, false, null);
+ WRITE_SIZE, 0, 2 * WRITE_SIZE, APPEND_MODE, false, null, true);
verify(client, times(1)).append(any(),
eq(PATH), any(byte[].class), refEq(firstReqParameters), any(),
@@ -204,9 +207,9 @@ public void verifyWriteRequest() throws Exception {
out.close();
AppendRequestParameters firstReqParameters = new AppendRequestParameters(
- 0, 0, BUFFER_SIZE, APPEND_MODE, false, null);
+ 0, 0, BUFFER_SIZE, APPEND_MODE, false, null, true);
AppendRequestParameters secondReqParameters = new AppendRequestParameters(
- BUFFER_SIZE, 0, 5 * WRITE_SIZE - BUFFER_SIZE, APPEND_MODE, false, null);
+ BUFFER_SIZE, 0, 5 * WRITE_SIZE - BUFFER_SIZE, APPEND_MODE, false, null, true);
verify(client, times(1)).append(any(),
eq(PATH), any(byte[].class), refEq(firstReqParameters), any(),
@@ -251,9 +254,9 @@ public void verifyWriteRequestOfBufferSizeAndClose() throws Exception {
out.close();
AppendRequestParameters firstReqParameters = new AppendRequestParameters(
- 0, 0, BUFFER_SIZE, APPEND_MODE, false, null);
+ 0, 0, BUFFER_SIZE, APPEND_MODE, false, null, true);
AppendRequestParameters secondReqParameters = new AppendRequestParameters(
- BUFFER_SIZE, 0, BUFFER_SIZE, APPEND_MODE, false, null);
+ BUFFER_SIZE, 0, BUFFER_SIZE, APPEND_MODE, false, null, true);
verify(client, times(1)).append(any(),
eq(PATH), any(byte[].class), refEq(firstReqParameters), any(), any(TracingContext.class), any());
@@ -294,9 +297,9 @@ public void verifyWriteRequestOfBufferSize() throws Exception {
Thread.sleep(1000);
AppendRequestParameters firstReqParameters = new AppendRequestParameters(
- 0, 0, BUFFER_SIZE, APPEND_MODE, false, null);
+ 0, 0, BUFFER_SIZE, APPEND_MODE, false, null, true);
AppendRequestParameters secondReqParameters = new AppendRequestParameters(
- BUFFER_SIZE, 0, BUFFER_SIZE, APPEND_MODE, false, null);
+ BUFFER_SIZE, 0, BUFFER_SIZE, APPEND_MODE, false, null, true);
verify(client, times(1)).append(any(),
eq(PATH), any(byte[].class), refEq(firstReqParameters), any(), any(TracingContext.class), any());
@@ -320,6 +323,7 @@ public void verifyWriteRequestOfBufferSizeWithAppendBlob() throws Exception {
true,
false,
true,
+ true,
client,
PATH,
new TracingContext(abfsConf.getClientCorrelationId(), "test-fs-id",
@@ -345,7 +349,7 @@ public void verifyWriteRequestOfBufferSizeWithAppendBlob() throws Exception {
@Test
public void verifyWriteRequestOfBufferSizeAndHFlush() throws Exception {
AbfsClient client = getClient();
- AbfsOutputStream out = Mockito.spy(getOutputStream(client, getConf()));
+ AbfsOutputStream out = getOutputStream(client, getConf());
final byte[] b = new byte[BUFFER_SIZE];
new Random().nextBytes(b);
@@ -356,9 +360,9 @@ public void verifyWriteRequestOfBufferSizeAndHFlush() throws Exception {
out.hflush();
AppendRequestParameters firstReqParameters = new AppendRequestParameters(
- 0, 0, BUFFER_SIZE, APPEND_MODE, false, null);
+ 0, 0, BUFFER_SIZE, APPEND_MODE, false, null, true);
AppendRequestParameters secondReqParameters = new AppendRequestParameters(
- BUFFER_SIZE, 0, BUFFER_SIZE, APPEND_MODE, false, null);
+ BUFFER_SIZE, 0, BUFFER_SIZE, APPEND_MODE, false, null, true);
verify(client, times(1)).append(any(),
eq(PATH), any(byte[].class), refEq(firstReqParameters), any(), any(TracingContext.class), any());
@@ -401,9 +405,9 @@ public void verifyWriteRequestOfBufferSizeAndFlush() throws Exception {
Thread.sleep(1000);
AppendRequestParameters firstReqParameters = new AppendRequestParameters(
- 0, 0, BUFFER_SIZE, APPEND_MODE, false, null);
+ 0, 0, BUFFER_SIZE, APPEND_MODE, false, null, true);
AppendRequestParameters secondReqParameters = new AppendRequestParameters(
- BUFFER_SIZE, 0, BUFFER_SIZE, APPEND_MODE, false, null);
+ BUFFER_SIZE, 0, BUFFER_SIZE, APPEND_MODE, false, null, true);
verify(client, times(1)).append(any(),
eq(PATH), any(byte[].class), refEq(firstReqParameters), any(), any(TracingContext.class), any());
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestExponentialRetryPolicy.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestExponentialRetryPolicy.java
index 0f8dc55aa14a4c..511e110c2887c8 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestExponentialRetryPolicy.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestExponentialRetryPolicy.java
@@ -18,20 +18,38 @@
package org.apache.hadoop.fs.azurebfs.services;
+import static java.net.HttpURLConnection.HTTP_INTERNAL_ERROR;
+import static org.apache.hadoop.fs.azure.integration.AzureTestConstants.TEST_CONFIGURATION_FILE_NAME;
import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_BACKOFF_INTERVAL;
import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_MAX_BACKOFF_INTERVAL;
import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_MAX_IO_RETRIES;
import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_MIN_BACKOFF_INTERVAL;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ACCOUNT_LEVEL_THROTTLING_ENABLED;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.MIN_BUFFER_SIZE;
+import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.FS_AZURE_ABFS_ACCOUNT1_NAME;
+import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.FS_AZURE_ACCOUNT_NAME;
+import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.FS_AZURE_ENABLE_AUTOTHROTTLING;
+import static org.junit.Assume.assumeTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import java.net.URI;
import java.util.Random;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.azurebfs.AzureBlobFileSystem;
+import org.assertj.core.api.Assertions;
import org.junit.Assert;
+import org.junit.Assume;
import org.junit.Test;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.azurebfs.AbfsConfiguration;
import org.apache.hadoop.fs.azurebfs.AbstractAbfsIntegrationTest;
+import org.mockito.Mockito;
/**
* Unit test TestExponentialRetryPolicy.
@@ -41,6 +59,9 @@ public class TestExponentialRetryPolicy extends AbstractAbfsIntegrationTest {
private final int noRetryCount = 0;
private final int retryCount = new Random().nextInt(maxRetryCount);
private final int retryCountBeyondMax = maxRetryCount + 1;
+ private static final String TEST_PATH = "/testfile";
+ private static final double MULTIPLYING_FACTOR = 1.5;
+ private static final int ANALYSIS_PERIOD = 10000;
public TestExponentialRetryPolicy() throws Exception {
@@ -67,6 +88,172 @@ public void testDefaultMaxIORetryCount() throws Exception {
testMaxIOConfig(abfsConfig);
}
+ @Test
+ public void testThrottlingIntercept() throws Exception {
+ AzureBlobFileSystem fs = getFileSystem();
+ final Configuration configuration = new Configuration();
+ configuration.addResource(TEST_CONFIGURATION_FILE_NAME);
+ configuration.setBoolean(FS_AZURE_ENABLE_AUTOTHROTTLING, false);
+
+ // On disabling throttling AbfsNoOpThrottlingIntercept object is returned
+ AbfsConfiguration abfsConfiguration = new AbfsConfiguration(configuration,
+ "dummy.dfs.core.windows.net");
+ AbfsThrottlingIntercept intercept;
+ AbfsClient abfsClient = ITestAbfsClient.createTestClientFromCurrentContext(getClient(fs), abfsConfiguration);
+ intercept = abfsClient.getIntercept();
+ Assertions.assertThat(intercept)
+ .describedAs("AbfsNoOpThrottlingIntercept instance expected")
+ .isInstanceOf(AbfsNoOpThrottlingIntercept.class);
+
+ configuration.setBoolean(FS_AZURE_ENABLE_AUTOTHROTTLING, true);
+ configuration.setBoolean(FS_AZURE_ACCOUNT_LEVEL_THROTTLING_ENABLED, true);
+ // On disabling throttling AbfsClientThrottlingIntercept object is returned
+ AbfsConfiguration abfsConfiguration1 = new AbfsConfiguration(configuration,
+ "dummy1.dfs.core.windows.net");
+ AbfsClient abfsClient1 = ITestAbfsClient.createTestClientFromCurrentContext(getClient(fs), abfsConfiguration1);
+ intercept = abfsClient1.getIntercept();
+ Assertions.assertThat(intercept)
+ .describedAs("AbfsClientThrottlingIntercept instance expected")
+ .isInstanceOf(AbfsClientThrottlingIntercept.class);
+ }
+
+ @Test
+ public void testCreateMultipleAccountThrottling() throws Exception {
+ Configuration config = new Configuration(getRawConfiguration());
+ String accountName = config.get(FS_AZURE_ACCOUNT_NAME);
+ if (accountName == null) {
+ // check if accountName is set using different config key
+ accountName = config.get(FS_AZURE_ABFS_ACCOUNT1_NAME);
+ }
+ assumeTrue("Not set: " + FS_AZURE_ABFS_ACCOUNT1_NAME,
+ accountName != null && !accountName.isEmpty());
+
+ Configuration rawConfig1 = new Configuration();
+ rawConfig1.addResource(TEST_CONFIGURATION_FILE_NAME);
+
+ AbfsRestOperation successOp = mock(AbfsRestOperation.class);
+ AbfsHttpOperation http500Op = mock(AbfsHttpOperation.class);
+ when(http500Op.getStatusCode()).thenReturn(HTTP_INTERNAL_ERROR);
+ when(successOp.getResult()).thenReturn(http500Op);
+
+ AbfsConfiguration configuration = Mockito.mock(AbfsConfiguration.class);
+ when(configuration.getAnalysisPeriod()).thenReturn(ANALYSIS_PERIOD);
+ when(configuration.isAutoThrottlingEnabled()).thenReturn(true);
+ when(configuration.accountThrottlingEnabled()).thenReturn(false);
+
+ AbfsThrottlingIntercept instance1 = AbfsThrottlingInterceptFactory.getInstance(accountName, configuration);
+ String accountName1 = config.get(FS_AZURE_ABFS_ACCOUNT1_NAME);
+
+ assumeTrue("Not set: " + FS_AZURE_ABFS_ACCOUNT1_NAME,
+ accountName1 != null && !accountName1.isEmpty());
+
+ AbfsThrottlingIntercept instance2 = AbfsThrottlingInterceptFactory.getInstance(accountName1, configuration);
+ //if singleton is enabled, for different accounts both the instances should return same value
+ Assertions.assertThat(instance1)
+ .describedAs(
+ "if singleton is enabled, for different accounts both the instances should return same value")
+ .isEqualTo(instance2);
+
+ when(configuration.accountThrottlingEnabled()).thenReturn(true);
+ AbfsThrottlingIntercept instance3 = AbfsThrottlingInterceptFactory.getInstance(accountName, configuration);
+ AbfsThrottlingIntercept instance4 = AbfsThrottlingInterceptFactory.getInstance(accountName1, configuration);
+ AbfsThrottlingIntercept instance5 = AbfsThrottlingInterceptFactory.getInstance(accountName, configuration);
+ //if singleton is not enabled, for different accounts instances should return different value
+ Assertions.assertThat(instance3)
+ .describedAs(
+ "iff singleton is not enabled, for different accounts instances should return different value")
+ .isNotEqualTo(instance4);
+
+ //if singleton is not enabled, for same accounts instances should return same value
+ Assertions.assertThat(instance3)
+ .describedAs(
+ "if singleton is not enabled, for same accounts instances should return same value")
+ .isEqualTo(instance5);
+ }
+
+ @Test
+ public void testOperationOnAccountIdle() throws Exception {
+ //Get the filesystem.
+ AzureBlobFileSystem fs = getFileSystem();
+ AbfsClient client = getClient(fs);
+ AbfsConfiguration configuration1 = client.getAbfsConfiguration();
+ Assume.assumeTrue(configuration1.isAutoThrottlingEnabled());
+ Assume.assumeTrue(configuration1.accountThrottlingEnabled());
+
+ AbfsClientThrottlingIntercept accountIntercept
+ = (AbfsClientThrottlingIntercept) client.getIntercept();
+ final byte[] b = new byte[2 * MIN_BUFFER_SIZE];
+ new Random().nextBytes(b);
+
+ Path testPath = path(TEST_PATH);
+
+ //Do an operation on the filesystem.
+ try (FSDataOutputStream stream = fs.create(testPath)) {
+ stream.write(b);
+ }
+
+ //Don't perform any operation on the account.
+ int sleepTime = (int) ((getAbfsConfig().getAccountOperationIdleTimeout()) * MULTIPLYING_FACTOR);
+ Thread.sleep(sleepTime);
+
+ try (FSDataInputStream streamRead = fs.open(testPath)) {
+ streamRead.read(b);
+ }
+
+ //Perform operations on another account.
+ AzureBlobFileSystem fs1 = new AzureBlobFileSystem();
+ Configuration config = new Configuration(getRawConfiguration());
+ String accountName1 = config.get(FS_AZURE_ABFS_ACCOUNT1_NAME);
+ assumeTrue("Not set: " + FS_AZURE_ABFS_ACCOUNT1_NAME,
+ accountName1 != null && !accountName1.isEmpty());
+ final String abfsUrl1 = this.getFileSystemName() + "12" + "@" + accountName1;
+ URI defaultUri1 = null;
+ defaultUri1 = new URI("abfss", abfsUrl1, null, null, null);
+ fs1.initialize(defaultUri1, getRawConfiguration());
+ AbfsClient client1 = getClient(fs1);
+ AbfsClientThrottlingIntercept accountIntercept1
+ = (AbfsClientThrottlingIntercept) client1.getIntercept();
+ try (FSDataOutputStream stream1 = fs1.create(testPath)) {
+ stream1.write(b);
+ }
+
+ //Verify the write analyzer for first account is idle but the read analyzer is not idle.
+ Assertions.assertThat(accountIntercept.getWriteThrottler()
+ .getIsOperationOnAccountIdle()
+ .get())
+ .describedAs("Write analyzer for first account should be idle the first time")
+ .isTrue();
+
+ Assertions.assertThat(
+ accountIntercept.getReadThrottler()
+ .getIsOperationOnAccountIdle()
+ .get())
+ .describedAs("Read analyzer for first account should not be idle")
+ .isFalse();
+
+ //Verify the write analyzer for second account is not idle.
+ Assertions.assertThat(
+ accountIntercept1.getWriteThrottler()
+ .getIsOperationOnAccountIdle()
+ .get())
+ .describedAs("Write analyzer for second account should not be idle")
+ .isFalse();
+
+ //Again perform an operation on the first account.
+ try (FSDataOutputStream stream2 = fs.create(testPath)) {
+ stream2.write(b);
+ }
+
+ //Verify the write analyzer on first account is not idle.
+ Assertions.assertThat(
+ accountIntercept.getWriteThrottler()
+ .getIsOperationOnAccountIdle()
+ .get())
+ .describedAs(
+ "Write analyzer for first account should not be idle second time")
+ .isFalse();
+ }
+
@Test
public void testAbfsConfigConstructor() throws Exception {
// Ensure we choose expected values that are not defaults
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/utils/DelegationSASGenerator.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/utils/DelegationSASGenerator.java
index 71e0326f79273d..3b7e3cacd0ec15 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/utils/DelegationSASGenerator.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/utils/DelegationSASGenerator.java
@@ -61,6 +61,7 @@ public String getDelegationSAS(String accountName, String containerName, String
case SASTokenProvider.WRITE_OPERATION:
case SASTokenProvider.SET_PROPERTIES_OPERATION:
case SASTokenProvider.LEASE_OPERATION:
+ case SASTokenProvider.SET_BLOB_METADATA_OPERATION:
sp = "w";
break;
case SASTokenProvider.DELETE_OPERATION:
@@ -96,6 +97,7 @@ public String getDelegationSAS(String accountName, String containerName, String
case SASTokenProvider.GET_BLOCK_LIST:
case SASTokenProvider.GET_BLOB_PROPERTIES_OPERATION:
case SASTokenProvider.GET_CONTAINER_PROPERTIES_OPERATION:
+ case SASTokenProvider.GET_BLOB_METADATA_OPERATION:
case SASTokenProvider.READ_OPERATION:
sp = "r";
break;