[pinpoint-apm#9932] Replace DefaultFuture with CompletableFuture

Author: emeroad

collector/src/main/java/com/navercorp/pinpoint/collector/controller/ClusterPointController.java

@@ -208,7 +208,12 @@ private CompletableFuture<ResponseMessage> request0(GrpcAgentConnection grpcAgen
             try {
                 responseFuture.get(3000, TimeUnit.MILLISECONDS);
                 return responseFuture;
-            } catch (InterruptedException | ExecutionException | TimeoutException e) {
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new PinpointSocketException(e);
+            } catch (ExecutionException e) {
+                throw new PinpointSocketException(e.getCause());
+            } catch (TimeoutException e) {
                 throw new PinpointSocketException(e);
             }
         }

collector/src/main/java/com/navercorp/pinpoint/collector/receiver/grpc/PinpointGrpcServer.java

@@ -29,7 +29,6 @@
 import com.navercorp.pinpoint.grpc.trace.PCmdRequest;
 import com.navercorp.pinpoint.grpc.trace.PCmdResponse;
 import com.navercorp.pinpoint.profiler.context.thrift.CommandGrpcToThriftMessageConverter;
-import com.navercorp.pinpoint.rpc.DefaultFuture;
 import com.navercorp.pinpoint.rpc.PinpointSocketException;
 import com.navercorp.pinpoint.rpc.ResponseMessage;
 import com.navercorp.pinpoint.rpc.client.RequestManager;
@@ -44,7 +43,6 @@
 import com.navercorp.pinpoint.rpc.stream.StreamChannel;
 import com.navercorp.pinpoint.rpc.stream.StreamChannelRepository;
 import com.navercorp.pinpoint.rpc.stream.StreamException;
-import com.navercorp.pinpoint.rpc.util.Futures;
 import com.navercorp.pinpoint.thrift.dto.command.TRouteResult;
 import com.navercorp.pinpoint.thrift.io.CommandHeaderTBaseSerializerFactory;
 import com.navercorp.pinpoint.thrift.util.SerializationUtils;
@@ -145,10 +143,9 @@ public CompletableFuture<ResponseMessage> request(GeneratedMessageV3 message) {
             return createFailedFuture(new PinpointSocketException(TRouteResult.NOT_SUPPORTED_REQUEST.name()));
         }
 
-        DefaultFuture<ResponseMessage> future = requestManager.register(request.getRequestId());
+        CompletableFuture<ResponseMessage> future = requestManager.register(request.getRequestId());
         requestObserver.onNext(request);
-
-        return Futures.toCompletableFuture(future);
+        return future;
     }
 
     // 1st message : client(collector) -> server(agent)
@@ -347,9 +344,9 @@ public void close(SocketStateCode toState) {
     }
 
     private void setFailMessageToFuture(int responseId, String message) {
-        DefaultFuture<ResponseMessage> future = requestManager.removeMessageFuture(responseId);
+        CompletableFuture<ResponseMessage> future = requestManager.removeMessageFuture(responseId);
         if (future != null) {
-            future.setFailure(new PinpointSocketException(message));
+            future.completeExceptionally(new PinpointSocketException(message));
         }
     }
 

profiler-test/src/main/java/com/navercorp/pinpoint/test/TestTcpDataSender.java

@@ -31,8 +31,7 @@
 import java.util.List;
 import java.util.Map.Entry;
 import java.util.NoSuchElementException;
-import java.util.concurrent.CompletableFuture;
-import java.util.function.Consumer;
+import java.util.function.BiConsumer;
 
 /**
  * @author Jongho Moon
@@ -140,7 +139,7 @@ public boolean request(MetaDataType data, int retry) {
     }
 
     @Override
-    public boolean request(MetaDataType data, Consumer<CompletableFuture<ResponseMessage>> listener) {
+    public boolean request(MetaDataType data, BiConsumer<ResponseMessage, Throwable> listener) {
         addData(data);
         return true;
     }

profiler/src/main/java/com/navercorp/pinpoint/profiler/AgentInfoSender.java

@@ -29,7 +29,9 @@
 import java.util.Objects;
 import java.util.Timer;
 import java.util.TimerTask;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /**
@@ -176,9 +178,9 @@ private boolean sendAgentInfo() {
                 AgentInfo agentInfo = agentInfoFactory.createAgentInfo();
 
                 logger.info("Sending AgentInfo {}", agentInfo);
-                ResponseFutureListener<ResponseMessage> listener = new ResponseFutureListener<>();
+                ResponseFutureListener<ResponseMessage, Throwable> listener = new ResponseFutureListener<>();
                 dataSender.request(agentInfo, listener);
-                ResponseMessage responseMessage = listener.getResponseFeture().get(3000, TimeUnit.MILLISECONDS);
+                ResponseMessage responseMessage = listener.getResponseFuture().get(3000, TimeUnit.MILLISECONDS);
                 if (responseMessage == null) {
                     logger.warn("result not set");
                     return false;
@@ -188,11 +190,20 @@ private boolean sendAgentInfo() {
                     logger.warn("request unsuccessful. Cause : {}", result.getMessage());
                 }
                 return result.isSuccess();
-            } catch (Throwable th) {
-                logger.warn("failed to send agent info", th);
+            } catch (ExecutionException ex) {
+                logError(ex.getCause());
+            } catch (InterruptedException ex) {
+                Thread.currentThread().interrupt();
+                logError(ex);
+            } catch (TimeoutException ex) {
+                logError(ex);
             }
             return false;
         }
+
+        private void logError(Throwable cause) {
+            logger.warn("failed to send agent info", cause);
+        }
     }
 
     public static class Builder {

profiler/src/main/java/com/navercorp/pinpoint/profiler/ResponseFutureListener.java

@@ -17,9 +17,9 @@
 package com.navercorp.pinpoint.profiler;
 
 import java.util.concurrent.CompletableFuture;
-import java.util.function.Consumer;
+import java.util.function.BiConsumer;
 
-public class ResponseFutureListener<T> implements Consumer<CompletableFuture<T>> {
+public class ResponseFutureListener<T, U  extends Throwable> implements BiConsumer<T, U> {
 
     private final CompletableFuture<T> future;
 
@@ -28,22 +28,15 @@ public ResponseFutureListener() {
     }
 
     @Override
-    public void accept(CompletableFuture<T> cf) {
-        if (!cf.isDone()) {
-            this.future.completeExceptionally(new IllegalStateException("ResponseMessage future is not complete"));
-            return;
+    public void accept(T message, U th) {
+        if (message != null) {
+            future.complete(message);
+        } else {
+            future.completeExceptionally(th);
         }
-
-        cf.whenComplete((responseMessage, throwable) -> {
-            if (throwable != null) {
-                future.completeExceptionally(throwable);
-            } else {
-                future.complete(responseMessage);
-            }
-        });
     }
 
-    public CompletableFuture<T> getResponseFeture() {
+    public CompletableFuture<T> getResponseFuture() {
         return future;
     }
 }

profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/EmptyDataSender.java

@@ -18,8 +18,7 @@
 
 import com.navercorp.pinpoint.rpc.ResponseMessage;
 
-import java.util.concurrent.CompletableFuture;
-import java.util.function.Consumer;
+import java.util.function.BiConsumer;
 
 
 /**
@@ -56,7 +55,7 @@ public boolean request(T data, int retry) {
 
 
     @Override
-    public boolean request(T data, Consumer<CompletableFuture<ResponseMessage>> listener) {
+    public boolean request(T data, BiConsumer<ResponseMessage, Throwable> listener) {
         return false;
     }
 

profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/EnhancedDataSender.java

@@ -18,8 +18,7 @@
 
 import com.navercorp.pinpoint.rpc.ResponseMessage;
 
-import java.util.concurrent.CompletableFuture;
-import java.util.function.Consumer;
+import java.util.function.BiConsumer;
 
 
 /**
@@ -31,7 +30,7 @@ public interface EnhancedDataSender<T> extends DataSender<T> {
 
     boolean request(T data, int retry);
 
-    boolean request(T data, Consumer<CompletableFuture<ResponseMessage>> listener);
+    boolean request(T data, BiConsumer<ResponseMessage, Throwable> listener);
 
 
 }

profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/ListenerableRequestMessage.java

@@ -20,8 +20,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
-import java.util.concurrent.CompletableFuture;
-import java.util.function.Consumer;
+import java.util.function.BiConsumer;
 
 
 /**
@@ -58,7 +57,7 @@ public boolean request(T data, int retry) {
 
 
     @Override
-    public boolean request(T data, Consumer<CompletableFuture<ResponseMessage>> listener) {
+    public boolean request(T data, BiConsumer<ResponseMessage, Throwable> listener) {
         logger.info("request tBase:{} FutureListener:{}", data, listener);
         return false;
     }

profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/LoggingDataSender.java

@@ -18,8 +18,7 @@
 
 import com.navercorp.pinpoint.rpc.ResponseMessage;
 
-import java.util.concurrent.CompletableFuture;
-import java.util.function.Consumer;
+import java.util.function.BiConsumer;
 
 /**
  * @author Woonduk Kang(emeroad)
@@ -30,6 +29,6 @@ public interface RequestMessage<M> {
 
     int getRetryCount();
 
-    Consumer<CompletableFuture<ResponseMessage>> getFutureListener();
+    BiConsumer<ResponseMessage, Throwable> getFutureListener();
 
 }

profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/RequestMessage.java

@@ -19,8 +19,7 @@
 
 import com.navercorp.pinpoint.rpc.ResponseMessage;
 
-import java.util.concurrent.CompletableFuture;
-import java.util.function.Consumer;
+import java.util.function.BiConsumer;
 
 /**
  * @author Woonduk Kang(emeroad)
@@ -31,11 +30,11 @@ private RequestMessageFactory() {
     }
 
     public static <T> RequestMessage<T> request(T message, int retryCount) {
-        return new RetryRequestMessage<>(message, retryCount);
+        return new RetryRequestMessage<>(message, retryCount, null);
     }
 
-    public static <T> RequestMessage<T> request(T message, Consumer<CompletableFuture<ResponseMessage>> futureListener) {
-        return new ListenerableRequestMessage<>(message, futureListener);
+    public static <T> RequestMessage<T> request(T message, int retryCount, BiConsumer<ResponseMessage, Throwable> futureListener) {
+        return new RetryRequestMessage<>(message, retryCount, futureListener);
     }
 
 }

profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/RequestMessageFactory.java

@@ -16,23 +16,23 @@
 
 package com.navercorp.pinpoint.profiler.sender;
 
-
 import com.navercorp.pinpoint.rpc.ResponseMessage;
 
-import java.util.concurrent.CompletableFuture;
-import java.util.function.Consumer;
+import java.util.function.BiConsumer;
 
 /**
  * @author Woonduk Kang(emeroad)
  */
-public class RetryRequestMessage<T> implements RequestMessage<T> {
+class RetryRequestMessage<T> implements RequestMessage<T> {
     private final T message;
     private final int retryCount;
+    private final BiConsumer<ResponseMessage, Throwable> futureListener;
 
 
-    RetryRequestMessage(T message, int retryCount) {
+    RetryRequestMessage(T message, int retryCount, BiConsumer<ResponseMessage, Throwable> futureListener) {
         this.message = message;
         this.retryCount = retryCount;
+        this.futureListener = futureListener;
     }
 
 
@@ -47,7 +47,7 @@ public int getRetryCount() {
     }
 
     @Override
-    public Consumer<CompletableFuture<ResponseMessage>> getFutureListener() {
-        return null;
+    public BiConsumer<ResponseMessage, Throwable> getFutureListener() {
+        return futureListener;
     }
 }