HADOOP-19569. S3A: stream write/close fails badly once FS is closed

Executors in hadoop-common to
 - pick up shutdown of inner executor and shut themselves down.
 - semaphore executor to decrement counters in this process so that
   queue state is updated
 This stops callers being able to submit work when the inner executor has
 shut down.

S3A code
- StoreImpl to IllegalStateException on  method invocation whene the service
  isn't running.
  Some methods are kept open as they do seem needed.
- WriteOperationHelper callbacks to raise IllegalStateException when invoked
  when FS is closed.

This is complex.

TODO:
- WriteOperationHelper MUST make all calls to the FS
  through its callback interface, rather than given a ref to S3AFS. This makes
  it easy to identify and lock down the methods.
- What is the correct exception to raise in write/close() failures?
  IOE or illegal state?

Author: steveloughran

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/BlockingThreadPoolExecutorService.java

@@ -130,21 +130,20 @@ public static BlockingThreadPoolExecutorService newInstance(
     slower than enqueueing. */
     final BlockingQueue<Runnable> workQueue =
         new LinkedBlockingQueue<>(waitingTasks + activeTasks);
+    final InnerExecutorRejection rejection = new InnerExecutorRejection();
     ThreadPoolExecutor eventProcessingExecutor =
         new ThreadPoolExecutor(activeTasks, activeTasks, keepAliveTime, unit,
             workQueue, newDaemonThreadFactory(prefixName),
-            new RejectedExecutionHandler() {
-              @Override
-              public void rejectedExecution(Runnable r,
-                  ThreadPoolExecutor executor) {
-                // This is not expected to happen.
-                LOG.error("Could not submit task to executor {}",
-                    executor.toString());
-              }
-            });
+            rejection);
     eventProcessingExecutor.allowCoreThreadTimeOut(true);
-    return new BlockingThreadPoolExecutorService(waitingTasks + activeTasks,
-        eventProcessingExecutor);
+    final BlockingThreadPoolExecutorService service =
+        new BlockingThreadPoolExecutorService(waitingTasks + activeTasks,
+            eventProcessingExecutor);
+    rejection.setDelegate((r, executor) -> {
+      service.shutdown();
+    });
+
+    return service;
   }
 
   /**
@@ -164,4 +163,28 @@ public String toString() {
         .append('}');
     return sb.toString();
   }
+
+  private static class InnerExecutorRejection implements RejectedExecutionHandler {
+
+    private RejectedExecutionHandler delegate;
+
+    private RejectedExecutionHandler getDelegate() {
+      return delegate;
+    }
+
+    private void setDelegate(final RejectedExecutionHandler delegate) {
+      this.delegate = delegate;
+    }
+
+    @Override
+    public void rejectedExecution(Runnable r,
+        ThreadPoolExecutor executor) {
+      // This is not expected to happen.
+      LOG.error("Could not submit task to executor {}",
+          executor.toString());
+      if (getDelegate() != null) {
+        delegate.rejectedExecution(r, executor);
+      }
+    }
+  }
 }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/SemaphoredDelegatingExecutor.java

@@ -31,6 +31,7 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -151,6 +152,7 @@ public <T> Future<T> submit(Runnable task, T result) {
 
   @Override
   public Future<?> submit(Runnable task) {
+    rejectWhenShutdown();
     try (DurationTracker ignored =
              trackerFactory.trackDuration(ACTION_EXECUTOR_ACQUIRED)) {
       queueingPermits.acquire();
@@ -163,6 +165,7 @@ public Future<?> submit(Runnable task) {
 
   @Override
   public void execute(Runnable command) {
+    rejectWhenShutdown();
     try (DurationTracker ignored =
              trackerFactory.trackDuration(ACTION_EXECUTOR_ACQUIRED)) {
       queueingPermits.acquire();
@@ -208,6 +211,12 @@ public String toString() {
     return sb.toString();
   }
 
+  private void rejectWhenShutdown() {
+    if (isShutdown()) {
+      throw new RejectedExecutionException("ExecutorService is shutdown");
+    }
+  }
+
   /**
    * Releases a permit after the task is executed.
    */
@@ -222,6 +231,7 @@ class RunnableWithPermitRelease implements Runnable {
     @Override
     public void run() {
       try {
+        rejectWhenShutdown();
         delegatee.run();
       } finally {
         queueingPermits.release();
@@ -244,6 +254,7 @@ class CallableWithPermitRelease<T> implements Callable<T> {
     @Override
     public T call() throws Exception {
       try {
+        rejectWhenShutdown();
         return delegatee.call();
       } finally {
         queueingPermits.release();

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

@@ -1944,6 +1944,7 @@ private final class WriteOperationHelperCallbacksImpl
     @Retries.OnceRaw
     public CompleteMultipartUploadResponse completeMultipartUpload(
         CompleteMultipartUploadRequest request) {
+      checkRunning();
       return getStore().completeMultipartUpload(request);
     }
 
@@ -1954,6 +1955,7 @@ public UploadPartResponse uploadPart(
         final RequestBody body,
         final DurationTrackerFactory durationTrackerFactory)
         throws AwsServiceException, UncheckedIOException {
+      checkRunning();
       return getStore().uploadPart(request, body, durationTrackerFactory);
     }
 
@@ -3211,6 +3213,7 @@ void deleteObjectAtPath(Path f,
       String key,
       boolean isFile)
       throws SdkException, IOException {
+    checkRunning();
     if (isFile) {
       instrumentation.fileDeleted(1);
     } else {
@@ -3313,7 +3316,7 @@ PutObjectResponse putObjectDirect(PutObjectRequest putObjectRequest,
       S3ADataBlocks.BlockUploadData uploadData,
       DurationTrackerFactory durationTrackerFactory)
       throws SdkException {
-
+    checkRunning();
     long len = getPutRequestLength(putObjectRequest);
     LOG.debug("PUT {} bytes to {}", len, putObjectRequest.key());
     incrementPutStartStatistics(len);
@@ -4422,6 +4425,14 @@ private void checkNotClosed() throws PathIOException {
     }
   }
 
+  /**
+   * Check the FS is running.
+   * @throws IllegalStateException if closed
+   */
+  protected void checkRunning() throws IllegalStateException {
+    Preconditions.checkState(!isClosed, "FileSystem is closed");
+  }
+
   /**
    * Get the delegation token support for this filesystem;
    * not null iff delegation support is enabled.

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

@@ -252,6 +252,15 @@ protected void serviceStart() throws Exception {
     initLocalDirAllocator();
   }
 
+  /**
+   * Check the service is running.
+   * @throws IllegalStateException if not in STARTED.
+   */
+  protected void checkRunning() throws IllegalStateException {
+    Preconditions.checkState(isInState(STATE.STARTED),
+        "Store is in state %s", getServiceState());
+  }
+
   /**
    * Return the store path capabilities.
    * If the object stream factory is non-null, hands off the
@@ -262,6 +271,7 @@ protected void serviceStart() throws Exception {
    */
   @Override
   public boolean hasPathCapability(final Path path, final String capability) {
+    checkRunning();
     switch (toLowerCase(capability)) {
     case StreamCapabilities.IOSTATISTICS:
       return true;
@@ -278,6 +288,7 @@ public boolean hasPathCapability(final Path path, final String capability) {
    */
   @Override
   public boolean inputStreamHasCapability(final String capability) {
+    checkRunning();
     if (objectInputStreamFactory != null) {
       return objectInputStreamFactory.hasCapability(capability);
     }
@@ -298,26 +309,29 @@ private void initLocalDirAllocator() {
   /** Acquire write capacity for rate limiting {@inheritDoc}. */
   @Override
   public Duration acquireWriteCapacity(final int capacity) {
+    checkRunning();
     return writeRateLimiter.acquire(capacity);
   }
 
   /** Acquire read capacity for rate limiting {@inheritDoc}. */
   @Override
   public Duration acquireReadCapacity(final int capacity) {
+    checkRunning();
     return readRateLimiter.acquire(capacity);
-
   }
 
   /**
    * Create a new store context.
    * @return a new store context.
    */
   private StoreContext createStoreContext() {
+    checkRunning();
     return storeContextFactory.createStoreContext();
   }
 
   @Override
   public StoreContext getStoreContext() {
+    checkRunning();
     return storeContext;
   }
 
@@ -327,41 +341,49 @@ public StoreContext getStoreContext() {
    * @throws UncheckedIOException on any failure to create the client.
    */
   private S3Client getS3Client() throws UncheckedIOException {
+    checkRunning();
     return clientManager.getOrCreateS3ClientUnchecked();
   }
 
   @Override
   public S3TransferManager getOrCreateTransferManager() throws IOException {
+    checkRunning();
     return clientManager.getOrCreateTransferManager();
   }
 
   @Override
   public S3Client getOrCreateS3Client() throws IOException {
+    checkRunning();
     return clientManager.getOrCreateS3Client();
   }
 
   @Override
   public S3AsyncClient getOrCreateAsyncClient() throws IOException {
+    checkRunning();
     return clientManager.getOrCreateAsyncClient();
   }
 
   @Override
   public S3Client getOrCreateS3ClientUnchecked() throws UncheckedIOException {
+    checkRunning();
     return clientManager.getOrCreateS3ClientUnchecked();
   }
 
   @Override
   public S3Client getOrCreateAsyncS3ClientUnchecked() throws UncheckedIOException {
+    checkRunning();
     return clientManager.getOrCreateAsyncS3ClientUnchecked();
   }
 
   @Override
   public S3Client getOrCreateUnencryptedS3Client() throws IOException {
+    checkRunning();
     return clientManager.getOrCreateUnencryptedS3Client();
   }
 
   @Override
   public DurationTrackerFactory getDurationTrackerFactory() {
+    checkRunning();
     return durationTrackerFactory;
   }
 
@@ -380,6 +402,7 @@ private S3AStorageStatistics getStorageStatistics() {
 
   @Override
   public RequestFactory getRequestFactory() {
+    checkRunning();
     return requestFactory;
   }
 
@@ -389,6 +412,7 @@ public RequestFactory getRequestFactory() {
    */
   @Override
   public ClientManager clientManager() {
+    checkRunning();
     return clientManager;
   }
 
@@ -507,6 +531,7 @@ private void incrementBytesWritten(final long bytes) {
    */
   @Override
   public void incrementPutStartStatistics(long bytes) {
+    checkRunning();
     LOG.debug("PUT start {} bytes", bytes);
     incrementWriteOperations();
     incrementGauge(OBJECT_PUT_REQUESTS_ACTIVE, 1);
@@ -599,6 +624,7 @@ public Map.Entry<Duration, DeleteObjectsResponse> deleteObjects(
       final DeleteObjectsRequest deleteRequest)
       throws SdkException {
 
+    checkRunning();
     DeleteObjectsResponse response;
     BulkDeleteRetryHandler retryHandler = new BulkDeleteRetryHandler(createStoreContext());
 
@@ -671,6 +697,7 @@ public HeadObjectResponse headObject(String key,
       Invoker changeInvoker,
       S3AFileSystemOperations fsHandler,
       String operation) throws IOException {
+    checkRunning();
     HeadObjectResponse response = getStoreContext().getInvoker()
         .retryUntranslated("HEAD " + key, true,
             () -> {
@@ -729,6 +756,7 @@ public HeadObjectResponse headObject(String key,
   public ResponseInputStream<GetObjectResponse> getRangedS3Object(String key,
       long start,
       long end) throws IOException {
+    checkRunning();
     final GetObjectRequest request = getRequestFactory().newGetObjectRequestBuilder(key)
         .range(S3AUtils.formatRange(start, end))
         .build();
@@ -757,7 +785,7 @@ public ResponseInputStream<GetObjectResponse> getRangedS3Object(String key,
   public Map.Entry<Duration, Optional<DeleteObjectResponse>> deleteObject(
       final DeleteObjectRequest request)
       throws SdkException {
-
+    checkRunning();
     String key = request.key();
     blockRootDelete(key);
     DurationInfo d = new DurationInfo(LOG, false, "deleting %s", key);
@@ -810,6 +838,7 @@ public UploadPartResponse uploadPart(
       final RequestBody body,
       @Nullable final DurationTrackerFactory trackerFactory)
       throws AwsServiceException, UncheckedIOException {
+    checkRunning();
     long len = request.contentLength();
     incrementPutStartStatistics(len);
     try {
@@ -852,6 +881,7 @@ public UploadInfo putObject(
       PutObjectRequest putObjectRequest,
       File file,
       ProgressableProgressListener listener) throws IOException {
+    checkRunning();
     long len = getPutRequestLength(putObjectRequest);
     LOG.debug("PUT {} bytes to {} via transfer manager ", len, putObjectRequest.key());
     incrementPutStartStatistics(len);
@@ -882,6 +912,7 @@ public UploadInfo putObject(
   @Retries.OnceTranslated
   public CompletedFileUpload waitForUploadCompletion(String key, UploadInfo uploadInfo)
       throws IOException {
+    checkRunning();
     FileUpload upload = uploadInfo.getFileUpload();
     try {
       CompletedFileUpload result = upload.completionFuture().join();
@@ -931,6 +962,7 @@ public LocalDirAllocator getDirectoryAllocator() {
   public File createTemporaryFileForWriting(String pathStr,
       long size,
       Configuration conf) throws IOException {
+    checkRunning();
     requireNonNull(directoryAllocator, "directory allocator not initialized");
     Path path = directoryAllocator.getLocalPathForWrite(pathStr,
         size, conf);