Revert "HDDS-10384. RPC client Reusing thread resources. (#6270)"

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/AbstractCommitWatcher.java

@@ -73,7 +73,7 @@ SortedMap<Long, List<BUFFER>> getCommitIndexMap() {
     return commitIndexMap;
   }
 
-  synchronized void updateCommitInfoMap(long index, List<BUFFER> buffers) {
+  void updateCommitInfoMap(long index, List<BUFFER> buffers) {
     commitIndexMap.computeIfAbsent(index, k -> new LinkedList<>())
         .addAll(buffers);
   }

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java

@@ -25,6 +25,7 @@
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
@@ -181,7 +182,8 @@ public BlockOutputStream(
             (long) flushPeriod * streamBufferArgs.getStreamBufferSize() == streamBufferArgs
                 .getStreamBufferFlushSize());
 
-    this.responseExecutor = blockOutputStreamResourceProvider.get();
+    // A single thread executor handle the responses of async requests
+    responseExecutor = Executors.newSingleThreadExecutor();
     bufferList = null;
     totalDataFlushedLength = 0;
     writtenDataLength = 0;
@@ -655,6 +657,7 @@ public void cleanup(boolean invalidateClient) {
       bufferList.clear();
     }
     bufferList = null;
+    responseExecutor.shutdown();
   }
 
   /**

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ec/reconstruction/ECReconstructionCoordinator.java

@@ -101,7 +101,8 @@ public class ECReconstructionCoordinator implements Closeable {
 
   private static final int EC_RECONSTRUCT_STRIPE_READ_POOL_MIN_SIZE = 3;
 
-  private static final int EC_RECONSTRUCT_STRIPE_WRITE_POOL_MIN_SIZE = 5;
+  // TODO: Adjusts to the appropriate value when the ec-reconstruct-writer thread pool is used.
+  private static final int EC_RECONSTRUCT_STRIPE_WRITE_POOL_MIN_SIZE = 0;
 
   private final ECContainerOperationClient containerOperationClient;
 

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ECKeyOutputStream.java

@@ -43,6 +43,8 @@
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
@@ -64,6 +66,7 @@ public final class ECKeyOutputStream extends KeyOutputStream
   private final int numParityBlks;
   private final ByteBufferPool bufferPool;
   private final RawErasureEncoder encoder;
+  private final ExecutorService flushExecutor;
   private final Future<Boolean> flushFuture;
   private final AtomicLong flushCheckpoint;
 
@@ -116,13 +119,12 @@ private ECKeyOutputStream(Builder builder) {
     this.writeOffset = 0;
     this.encoder = CodecUtil.createRawEncoderWithFallback(
         builder.getReplicationConfig());
+    this.flushExecutor = Executors.newSingleThreadExecutor();
     S3Auth s3Auth = builder.getS3CredentialsProvider().get();
     ThreadLocal<S3Auth> s3CredentialsProvider =
         builder.getS3CredentialsProvider();
-    this.flushFuture = builder.getExecutorServiceSupplier().get().submit(() -> {
-      s3CredentialsProvider.set(s3Auth);
-      return flushStripeFromQueue();
-    });
+    flushExecutor.submit(() -> s3CredentialsProvider.set(s3Auth));
+    this.flushFuture = this.flushExecutor.submit(this::flushStripeFromQueue);
     this.flushCheckpoint = new AtomicLong(0);
     this.atomicKeyCreation = builder.getAtomicKeyCreation();
   }
@@ -493,6 +495,7 @@ public void close() throws IOException {
     } catch (InterruptedException e) {
       throw new IOException("Flushing thread was interrupted", e);
     } finally {
+      flushExecutor.shutdownNow();
       closeCurrentStreamEntry();
       blockOutputStreamEntryPool.cleanup();
     }

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java

@@ -196,7 +196,8 @@ public class RpcClient implements ClientProtocol {
   // for reconstruction.
   private static final int EC_RECONSTRUCT_STRIPE_READ_POOL_MIN_SIZE = 3;
 
-  private static final int WRITE_POOL_MIN_SIZE = 1;
+  // TODO: Adjusts to the appropriate value when the writeThreadPool is used.
+  private static final int WRITE_POOL_MIN_SIZE = 0;
 
   private final ConfigurationSource conf;
   private final OzoneManagerClientProtocol ozoneManagerClient;

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneAtRestEncryption.java

@@ -213,14 +213,6 @@ static void shutdown() throws IOException {
     }
   }
 
-  static void reInitClient() throws IOException {
-    ozClient = OzoneClientFactory.getRpcClient(conf);
-    store = ozClient.getObjectStore();
-    TestOzoneRpcClient.setOzClient(ozClient);
-    TestOzoneRpcClient.setStore(store);
-  }
-
-
   @ParameterizedTest
   @EnumSource
   void testPutKeyWithEncryption(BucketLayout bucketLayout) throws Exception {
@@ -778,7 +770,9 @@ void testGetKeyProvider() throws Exception {
 
     KeyProvider kp3 = ozClient.getObjectStore().getKeyProvider();
     assertNotEquals(kp3, kpSpy);
-    reInitClient();
+    // Restore ozClient and store
+    TestOzoneRpcClient.setOzClient(OzoneClientFactory.getRpcClient(conf));
+    TestOzoneRpcClient.setStore(ozClient.getObjectStore());
   }
 
   private static RepeatedOmKeyInfo getMatchedKeyInfo(