Merge pull request #569 from alex268/master

Added validation of grpc context deadline before call execution

Author: alex268

core/src/main/java/tech/ydb/core/impl/BaseGrpcTransport.java

@@ -7,6 +7,8 @@
 
 import io.grpc.CallOptions;
 import io.grpc.ClientCall;
+import io.grpc.Context;
+import io.grpc.Deadline;
 import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
 import org.slf4j.Logger;
@@ -64,32 +66,46 @@ public void close() {
         }
     }
 
-    @Override
-    public <ReqT, RespT> CompletableFuture<Result<RespT>> unaryCall(
-            MethodDescriptor<ReqT, RespT> method,
-            GrpcRequestSettings settings,
-            ReqT request
-    ) {
-        if (isClosed.get()) {
-            return CompletableFuture.completedFuture(SHUTDOWN_RESULT.map(o -> null));
-        }
-
-        String traceId = settings.getTraceId();
+    private CallOptions prepareCallOptions(GrpcRequestSettings settings) {
         CallOptions options = getAuthCallOptions().getGrpcCallOptions();
         if (settings.getDeadlineAfter() != 0) {
             final long now = System.nanoTime();
             if (now >= settings.getDeadlineAfter()) {
-                return CompletableFuture.completedFuture(deadlineExpiredResult(method, settings));
+                return null; // DEADLINE
             }
             options = options.withDeadlineAfter(settings.getDeadlineAfter() - now, TimeUnit.NANOSECONDS);
         }
         if (settings.isDeadlineDisabled()) {
             options = options.withDeadline(null);
         }
 
+        Deadline deadline = Context.current().getDeadline();
+        if (deadline != null && deadline.isExpired()) {
+            return null; // DEADLINE
+        }
+
+        return options;
+    }
+
+    @Override
+    public <ReqT, RespT> CompletableFuture<Result<RespT>> unaryCall(
+            MethodDescriptor<ReqT, RespT> method,
+            GrpcRequestSettings settings,
+            ReqT request
+    ) {
+        if (isClosed.get()) {
+            return CompletableFuture.completedFuture(SHUTDOWN_RESULT.map(o -> null));
+        }
+
+        String traceId = settings.getTraceId();
         try {
             GrpcChannel channel = getChannel(settings);
             String endpoint = channel.getEndpoint().getHostAndPort();
+            CallOptions options = prepareCallOptions(settings);
+            if (options == null) {
+                return CompletableFuture.completedFuture(deadlineExpiredResult(method, settings));
+            }
+
             ClientCall<ReqT, RespT> call = channel.getReadyChannel().newCall(method, options);
             ChannelStatusHandler handler = new ChannelStatusHandler(channel, settings);
 
@@ -121,21 +137,14 @@ public <ReqT, RespT> GrpcReadStream<RespT> readStreamCall(
         }
 
         String traceId = settings.getTraceId();
-        CallOptions options = getAuthCallOptions().getGrpcCallOptions();
-        if (settings.getDeadlineAfter() != 0) {
-            final long now = System.nanoTime();
-            if (now >= settings.getDeadlineAfter()) {
-                return new EmptyStream<>(deadlineExpiredStatus(method, settings));
-            }
-            options = options.withDeadlineAfter(settings.getDeadlineAfter() - now, TimeUnit.NANOSECONDS);
-        }
-        if (settings.isDeadlineDisabled()) {
-            options = options.withDeadline(null);
-        }
-
         try {
             GrpcChannel channel = getChannel(settings);
             String endpoint = channel.getEndpoint().getHostAndPort();
+            CallOptions options = prepareCallOptions(settings);
+            if (options == null) {
+                return new EmptyStream<>(deadlineExpiredStatus(method, settings));
+            }
+
             ClientCall<ReqT, RespT> call = channel.getReadyChannel().newCall(method, options);
             ChannelStatusHandler handler = new ChannelStatusHandler(channel, settings);
 
@@ -159,7 +168,6 @@ public <ReqT, RespT> GrpcReadStream<RespT> readStreamCall(
         }
     }
 
-
     @Override
     public <ReqT, RespT> GrpcReadWriteStream<RespT, ReqT> readWriteStreamCall(
             MethodDescriptor<ReqT, RespT> method,
@@ -170,21 +178,14 @@ public <ReqT, RespT> GrpcReadWriteStream<RespT, ReqT> readWriteStreamCall(
         }
 
         String traceId = settings.getTraceId();
-        CallOptions options = getAuthCallOptions().getGrpcCallOptions();
-        if (settings.getDeadlineAfter() != 0) {
-            final long now = System.nanoTime();
-            if (now >= settings.getDeadlineAfter()) {
-                return new EmptyStream<>(deadlineExpiredStatus(method, settings));
-            }
-            options = options.withDeadlineAfter(settings.getDeadlineAfter() - now, TimeUnit.NANOSECONDS);
-        }
-        if (settings.isDeadlineDisabled()) {
-            options = options.withDeadline(null);
-        }
-
         try {
             GrpcChannel channel = getChannel(settings);
             String endpoint = channel.getEndpoint().getHostAndPort();
+            CallOptions options = prepareCallOptions(settings);
+            if (options == null) {
+                return new EmptyStream<>(deadlineExpiredStatus(method, settings));
+            }
+
             ClientCall<ReqT, RespT> call = channel.getReadyChannel().newCall(method, options);
             ChannelStatusHandler hdlr = new ChannelStatusHandler(channel, settings);