no need synchronized

no need synchronized

no need synchronized

Author: KeeProMise

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java

@@ -305,8 +305,9 @@ void checkResponse(RpcResponseHeaderProto header) throws IOException {
     }
   }
 
-  Call createCall(RPC.RpcKind rpcKind, Writable rpcRequest) {
-    return new Call(rpcKind, rpcRequest);
+  Call createCall(RPC.RpcKind rpcKind, Writable rpcRequest,
+      AlignmentContext alignmentContext) {
+    return new Call(rpcKind, rpcRequest, alignmentContext);
   }
 
   /** 
@@ -319,11 +320,13 @@ static class Call {
     private final CompletableFuture<Writable> rpcResponseFuture;
     final RPC.RpcKind rpcKind;      // Rpc EngineKind
     private final Object externalHandler;
-    private AlignmentContext alignmentContext;
+    private final AlignmentContext alignmentContext;
 
-    private Call(RPC.RpcKind rpcKind, Writable param) {
+    private Call(RPC.RpcKind rpcKind, Writable param,
+        AlignmentContext alignmentContext) {
       this.rpcKind = rpcKind;
       this.rpcRequest = param;
+      this.alignmentContext = alignmentContext;
 
       final Integer id = callId.get();
       if (id == null) {
@@ -351,44 +354,32 @@ public String toString() {
 
     /** Indicate when the call is complete and the
      * value or error are available.  Notifies by default.  */
-    protected synchronized void callComplete(Writable rpcResponse, IOException error) {
-      if (error != null) {
-        rpcResponseFuture.completeExceptionally(error);
-      } else {
-        rpcResponseFuture.complete(rpcResponse);
-      }
+    protected void callComplete() {
       if (externalHandler != null) {
         synchronized (externalHandler) {
           externalHandler.notify();
         }
       }
     }
 
-    /**
-     * Set an AlignmentContext for the call to update when call is done.
-     *
-     * @param ac alignment context to update.
-     */
-    public synchronized void setAlignmentContext(AlignmentContext ac) {
-      this.alignmentContext = ac;
-    }
-
     /** Set the exception when there is an error.
      * Notify the caller the call is done.
      * 
      * @param error exception thrown by the call; either local or remote
      */
-    public synchronized void setException(IOException error) {
-      callComplete(null, error);
+    public void setException(IOException error) {
+      rpcResponseFuture.completeExceptionally(error);
+      callComplete();
     }
 
     /** Set the return value when there is no error. 
      * Notify the caller the call is done.
      * 
      * @param rpcResponse return value of the rpc call.
      */
-    public synchronized void setRpcResponse(Writable rpcResponse) {
-      callComplete(rpcResponse, null);
+    public void setRpcResponse(Writable rpcResponse) {
+      rpcResponseFuture.complete(rpcResponse);
+      callComplete();
     }
   }
 
@@ -1503,8 +1494,7 @@ Writable call(RPC.RpcKind rpcKind, Writable rpcRequest,
       ConnectionId remoteId, int serviceClass,
       AtomicBoolean fallbackToSimpleAuth, AlignmentContext alignmentContext)
       throws IOException {
-    final Call call = createCall(rpcKind, rpcRequest);
-    call.setAlignmentContext(alignmentContext);
+    final Call call = createCall(rpcKind, rpcRequest, alignmentContext);
     final Connection connection = getConnection(remoteId, call, serviceClass,
         fallbackToSimpleAuth);
 

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestAsyncIPC.java

@@ -422,12 +422,13 @@ public void testCallIdAndRetry() throws IOException, InterruptedException,
     // Override client to store the call info and check response
     final Client client = new Client(LongWritable.class, conf) {
       @Override
-      Call createCall(RpcKind rpcKind, Writable rpcRequest) {
+      Call createCall(RpcKind rpcKind, Writable rpcRequest,
+          AlignmentContext alignmentContext) {
         // Set different call id and retry count for the next call
         Client.setCallIdAndRetryCount(Client.nextCallId(),
             TestIPC.RANDOM.nextInt(255), null);
 
-        final Call call = super.createCall(rpcKind, rpcRequest);
+        final Call call = super.createCall(rpcKind, rpcRequest, alignmentContext);
 
         CallInfo info = new CallInfo();
         info.id = call.id;
@@ -629,4 +630,4 @@ public void testAsyncCallWithCompletableFuture() throws IOException,
       server.stop();
     }
   }
-}
+}

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPC.java

@@ -1301,8 +1301,9 @@ public void testCallIdAndRetry() throws IOException {
     // Override client to store the call info and check response
     final Client client = new Client(LongWritable.class, conf) {
       @Override
-      Call createCall(RpcKind rpcKind, Writable rpcRequest) {
-        final Call call = super.createCall(rpcKind, rpcRequest);
+      Call createCall(RpcKind rpcKind, Writable rpcRequest,
+          AlignmentContext alignmentContext) {
+        final Call call = super.createCall(rpcKind, rpcRequest, alignmentContext);
         info.id = call.id;
         info.retry = call.retry;
         return call;