hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ste...@apache.org
Subject [hadoop] branch branch-3.2 updated: HADOOP-16040. ABFS: Bug fix for tolerateOobAppends configuration.
Date Thu, 10 Jan 2019 12:00:07 GMT
This is an automated email from the ASF dual-hosted git repository.

stevel pushed a commit to branch branch-3.2
in repository https://gitbox.apache.org/repos/asf/hadoop.git


The following commit(s) were added to refs/heads/branch-3.2 by this push:
     new f7de630  HADOOP-16040. ABFS: Bug fix for tolerateOobAppends configuration.
f7de630 is described below

commit f7de630e85b8d3ed55b7f5e3eecba6c5dd76dc54
Author: Da Zhou <da.zhou@microsoft.com>
AuthorDate: Thu Jan 10 11:59:29 2019 +0000

    HADOOP-16040. ABFS: Bug fix for tolerateOobAppends configuration.
    
    Contributed by Da Zhou.
    
    (cherry picked from commit e8d19003695e3bc76bfa8e1187a238eec0220def)
---
 .../fs/azurebfs/AzureBlobFileSystemStore.java      |  3 +-
 .../fs/azurebfs/services/AbfsInputStream.java      |  3 +-
 .../fs/azurebfs/ITestAzureBlobFileSystemE2E.java   | 37 ++++++++++++++++++++--
 3 files changed, 39 insertions(+), 4 deletions(-)

diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystemStore.java
b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystemStore.java
index b2bd64f..222285f 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystemStore.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystemStore.java
@@ -374,7 +374,8 @@ public class AzureBlobFileSystemStore {
     // Add statistics for InputStream
     return new AbfsInputStream(client, statistics,
             AbfsHttpConstants.FORWARD_SLASH + getRelativePath(path), contentLength,
-                abfsConfiguration.getReadBufferSize(), abfsConfiguration.getReadAheadQueueDepth(),
eTag);
+                abfsConfiguration.getReadBufferSize(), abfsConfiguration.getReadAheadQueueDepth(),
+                abfsConfiguration.getTolerateOobAppends(), eTag);
   }
 
   public OutputStream openFileForWrite(final Path path, final boolean overwrite) throws
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsInputStream.java
b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsInputStream.java
index 9593679..fe48cb9 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsInputStream.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsInputStream.java
@@ -61,6 +61,7 @@ public class AbfsInputStream extends FSInputStream {
       final long contentLength,
       final int bufferSize,
       final int readAheadQueueDepth,
+      final boolean tolerateOobAppends,
       final String eTag) {
     this.client = client;
     this.statistics = statistics;
@@ -68,8 +69,8 @@ public class AbfsInputStream extends FSInputStream {
     this.contentLength = contentLength;
     this.bufferSize = bufferSize;
     this.readAheadQueueDepth = (readAheadQueueDepth >= 0) ? readAheadQueueDepth : Runtime.getRuntime().availableProcessors();
+    this.tolerateOobAppends = tolerateOobAppends;
     this.eTag = eTag;
-    this.tolerateOobAppends = false;
     this.readAheadEnabled = true;
   }
 
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemE2E.java
b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemE2E.java
index 9e22790..ebc9c07 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemE2E.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemE2E.java
@@ -25,12 +25,14 @@ import java.util.Random;
 
 import org.junit.Test;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys;
 
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_TOLERATE_CONCURRENT_APPEND;
 import static org.apache.hadoop.test.LambdaTestUtils.intercept;
 
 /**
@@ -66,7 +68,9 @@ public class ITestAzureBlobFileSystemE2E extends AbstractAbfsIntegrationTest
{
   }
 
   @Test (expected = IOException.class)
-  public void testOOBWrites() throws Exception {
+  public void testOOBWritesAndReadFail() throws Exception {
+    Configuration conf = this.getRawConfiguration();
+    conf.setBoolean(AZURE_TOLERATE_CONCURRENT_APPEND, false);
     final AzureBlobFileSystem fs = getFileSystem();
     int readBufferSize = fs.getAbfsStore().getAbfsConfiguration().getReadBufferSize();
 
@@ -83,7 +87,6 @@ public class ITestAzureBlobFileSystemE2E extends AbstractAbfsIntegrationTest
{
     try (FSDataInputStream readStream = fs.open(testFilePath)) {
       assertEquals(readBufferSize,
           readStream.read(bytesToRead, 0, readBufferSize));
-
       try (FSDataOutputStream writeStream = fs.create(testFilePath)) {
         writeStream.write(b);
         writeStream.flush();
@@ -95,6 +98,36 @@ public class ITestAzureBlobFileSystemE2E extends AbstractAbfsIntegrationTest
{
   }
 
   @Test
+  public void testOOBWritesAndReadSucceed() throws Exception {
+    Configuration conf = this.getRawConfiguration();
+    conf.setBoolean(AZURE_TOLERATE_CONCURRENT_APPEND, true);
+    final AzureBlobFileSystem fs = getFileSystem(conf);
+    int readBufferSize = fs.getAbfsStore().getAbfsConfiguration().getReadBufferSize();
+
+    byte[] bytesToRead = new byte[readBufferSize];
+    final byte[] b = new byte[2 * readBufferSize];
+    new Random().nextBytes(b);
+    final Path testFilePath = new Path(methodName.getMethodName());
+
+    try (FSDataOutputStream writeStream = fs.create(testFilePath)) {
+      writeStream.write(b);
+      writeStream.flush();
+    }
+
+    try (FSDataInputStream readStream = fs.open(testFilePath)) {
+      // Read
+      assertEquals(readBufferSize, readStream.read(bytesToRead, 0, readBufferSize));
+      // Concurrent write
+      try (FSDataOutputStream writeStream = fs.create(testFilePath)) {
+        writeStream.write(b);
+        writeStream.flush();
+      }
+
+      assertEquals(readBufferSize, readStream.read(bytesToRead, 0, readBufferSize));
+    }
+  }
+
+  @Test
   public void testWriteWithBufferOffset() throws Exception {
     final AzureBlobFileSystem fs = getFileSystem();
     final Path testFilePath = new Path(methodName.getMethodName());


---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org


Mime
View raw message