Address review feedback v2

Author: fuatbasik

hadoop-tools/hadoop-aws/pom.xml

@@ -528,7 +528,7 @@
     <dependency>
       <groupId>software.amazon.s3.analyticsaccelerator</groupId>
       <artifactId>analyticsaccelerator-s3</artifactId>
-      <version>0.0.1</version>
+      <version>0.0.2</version>
       <scope>compile</scope>
     </dependency>
     <dependency>

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

@@ -326,7 +326,7 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
 
   /**
    * CRT-Based S3Client created of analytics accelerator library is enabled
-   * and managed by the ClientManager. Analytics accelerator library can be
+   * and managed by the S3AStoreImpl. Analytics accelerator library can be
    * enabled with {@link Constants#ANALYTICS_ACCELERATOR_ENABLED_KEY}
    */
   private S3AsyncClient s3AsyncClient;
@@ -545,7 +545,7 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
   private boolean s3AccessGrantsEnabled;
 
   /**
-   * Factory to create S3SeekableInputStream if {@link this#analyticsAcceleratorEnabled} is true
+   * Factory to create S3SeekableInputStream if {@link this#analyticsAcceleratorEnabled} is true.
    */
   private S3SeekableInputStreamFactory s3SeekableInputStreamFactory;
 
@@ -705,8 +705,10 @@ public void initialize(URI name, Configuration originalConf)
       this.prefetchBlockCount =
           intOption(conf, PREFETCH_BLOCK_COUNT_KEY, PREFETCH_BLOCK_DEFAULT_COUNT, 1);
 
-      this.analyticsAcceleratorEnabled = conf.getBoolean(ANALYTICS_ACCELERATOR_ENABLED_KEY, ANALYTICS_ACCELERATOR_ENABLED_DEFAULT);
-      this.analyticsAcceleratorCRTEnabled = conf.getBoolean(ANALYTICS_ACCELERATOR_CRT_ENABLED, ANALYTICS_ACCELERATOR_CRT_ENABLED_DEFAULT);
+      this.analyticsAcceleratorEnabled =
+          conf.getBoolean(ANALYTICS_ACCELERATOR_ENABLED_KEY, ANALYTICS_ACCELERATOR_ENABLED_DEFAULT);
+      this.analyticsAcceleratorCRTEnabled =
+          conf.getBoolean(ANALYTICS_ACCELERATOR_CRT_ENABLED, ANALYTICS_ACCELERATOR_CRT_ENABLED_DEFAULT);
 
       this.isMultipartUploadEnabled = conf.getBoolean(MULTIPART_UPLOADS_ENABLED,
               DEFAULT_MULTIPART_UPLOAD_ENABLED);
@@ -847,10 +849,10 @@ public void initialize(URI name, Configuration originalConf)
       if (this.analyticsAcceleratorEnabled) {
         LOG.info("Using S3SeekableInputStream");
         if(this.analyticsAcceleratorCRTEnabled) {
-          LOG.info("Using S3CrtClient");
+          LOG.info("Using S3 CRT client for analytics accelerator S3");
           this.s3AsyncClient = S3CrtAsyncClient.builder().maxConcurrency(600).build();
         } else {
-          LOG.info("Using S3Client");
+          LOG.info("Using S3 async client for analytics accelerator S3");
           this.s3AsyncClient = store.getOrCreateAsyncClient();
         }
 

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ASeekableStream.java

@@ -38,6 +38,7 @@ public class S3ASeekableStream extends FSInputStream implements StreamCapabiliti
   private S3SeekableInputStream inputStream;
   private long lastReadCurrentPos = 0;
   private final String key;
+  private volatile boolean closed;
 
   public static final Logger LOG = LoggerFactory.getLogger(S3ASeekableStream.class);
 
@@ -84,7 +85,7 @@ public void seek(long pos) throws IOException {
 
   @Override
   public synchronized long getPos() {
-    if (!isClosed()) {
+    if (!closed) {
       lastReadCurrentPos = inputStream.getPos();
     }
     return lastReadCurrentPos;
@@ -139,8 +140,9 @@ public int available() throws IOException {
   }
 
   @Override
-  public void close() throws IOException {
-    if (inputStream != null) {
+  public synchronized void close() throws IOException {
+    if(!closed) {
+      closed = true;
       try {
         inputStream.close();
         inputStream = null;
@@ -174,12 +176,8 @@ private void onReadFailure(IOException ioe) throws IOException {
 
 
   protected void throwIfClosed() throws IOException {
-    if (isClosed()) {
+    if (closed) {
       throw new IOException(key + ": " + FSExceptionMessages.STREAM_IS_CLOSED);
     }
   }
-
-  protected boolean isClosed() {
-    return inputStream == null;
-  }
 }

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/ITestS3AContractCreate.java

@@ -89,7 +89,9 @@ protected Configuration createConfiguration() {
 
   @Override
   public void testOverwriteExistingFile() throws Throwable {
-    skipIfAnalyticsAcceleratorEnabled(this.createConfiguration());
+    // Will remove this when Analytics Accelerator supports overwrites
+    skipIfAnalyticsAcceleratorEnabled(this.createConfiguration(),
+        "Analytics Accelerator does not support overwrites yet");
     super.testOverwriteExistingFile();
   }
 

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/ITestS3AContractDistCp.java

@@ -52,12 +52,6 @@ protected Configuration createConfiguration() {
     return newConf;
   }
 
-  @Override
-  public void setup() throws Exception {
-    super.setup();
-    skipIfAnalyticsAcceleratorEnabled(createConfiguration());
-  }
-
   @Override
   protected boolean shouldUseDirectWrite() {
     return true;
@@ -85,6 +79,14 @@ public void testNonDirectWrite() throws Exception {
         getRenameOperationCount() - renames);
   }
 
+  @Override
+  public void testDistCpUpdateCheckFileSkip() throws Exception {
+    //Will remove this when Analytics Accelerator supports overwrites
+    skipIfAnalyticsAcceleratorEnabled(createConfiguration(),
+        "Analytics Accelerator Library does not support update to existing files");
+    super.testDistCpUpdateCheckFileSkip();
+  }
+
   private long getRenameOperationCount() {
     return getFileSystem().getStorageStatistics()
         .getLong(StorageStatistics.CommonStatisticNames.OP_RENAME);

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/ITestS3AContractRename.java

@@ -49,7 +49,8 @@ public class ITestS3AContractRename extends AbstractContractRenameTest {
   @Override
   public void setup() throws Exception {
     super.setup();
-    skipIfAnalyticsAcceleratorEnabled(getContract().getConf());
+    skipIfAnalyticsAcceleratorEnabled(createConfiguration(),
+        "Analytics Accelerator does not support rename");
 
   }
   @Override

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/ITestS3AContractVectoredRead.java

@@ -86,12 +86,13 @@ protected AbstractFSContract createContract(Configuration conf) {
   }
 
   /**
-   * Analytics Accelerator Library for Amazon S3 does not support Vectored Reads
+   * Analytics Accelerator Library for Amazon S3 does not support Vectored Reads.
    * @throws Exception
    */
   @Override
   public void setup() throws Exception {
-    skipIfAnalyticsAcceleratorEnabled(createConfiguration());
+    skipIfAnalyticsAcceleratorEnabled(createConfiguration(),
+        "Analytics Accelerator does not support vectored reads");
     super.setup();
   }
 

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3ADelayedFNF.java

@@ -36,6 +36,7 @@
 import static org.apache.hadoop.fs.s3a.Constants.RETRY_INTERVAL;
 import static org.apache.hadoop.fs.s3a.Constants.RETRY_LIMIT;
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.skipIfAnalyticsAcceleratorEnabled;
 
 /**
  * Tests behavior of a FileNotFound error that happens after open(), i.e. on
@@ -65,6 +66,8 @@ protected Configuration createConfiguration() {
    */
   @Test
   public void testNotFoundFirstRead() throws Exception {
+    skipIfAnalyticsAcceleratorEnabled(getConfiguration(),
+        "Temporarily disabling to fix Exception handling on Analytics Accelerator");
     S3AFileSystem fs = getFileSystem();
     ChangeDetectionPolicy changeDetectionPolicy =
         fs.getChangeDetectionPolicy();

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AEncryptionSSEC.java

@@ -91,7 +91,8 @@ protected Configuration createConfiguration() {
   @Override
   public void setup() throws Exception {
     super.setup();
-    skipIfAnalyticsAcceleratorEnabled(getConfiguration());
+    skipIfAnalyticsAcceleratorEnabled(getConfiguration(),
+        "Analytics Accelerator does not support SSEC");
     assumeEnabled();
     // although not a root dir test, this confuses paths enough it shouldn't be run in
     // parallel with other jobs

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFSMainOperations.java

@@ -78,17 +78,19 @@ public void testCopyToLocalWithUseRawLocalFileSystemOption()
 
   @Override
   public void testWriteReadAndDeleteOneAndAHalfBlocks() throws Exception {
-    //Skipping this test for AnalyticsAccelerator as it is acting as an overwrite test
-    skipIfAnalyticsAcceleratorEnabled(this.contract.getConf());
+    // Skipping this test for AnalyticsAccelerator as it is acting as an overwrite test
+    skipIfAnalyticsAcceleratorEnabled(this.contract.getConf(),
+        "Analytics Accelerator does not support overwrites");
   }
 
   @Override
   public void testWriteReadAndDeleteTwoBlocks() throws Exception {
-    //Skipping this test for AnalyticsAccelerator as it is acting as an overwrite test
-    skipIfAnalyticsAcceleratorEnabled(this.contract.getConf());
+    // Skipping this test for AnalyticsAccelerator as it is acting as an overwrite test
+    skipIfAnalyticsAcceleratorEnabled(this.contract.getConf(),
+        "Analytics Accelerator does not support overwrites");
   }
 
-    @Override
+  @Override
   public void testOverwrite() throws IOException {
     boolean createPerformance = isCreatePerformanceEnabled(fSys);
     try {