Java client: update error handling

java/client/build.gradle

@@ -110,7 +110,7 @@ tasks.register('buildAll') {
     finalizedBy 'build'
 }
 
-compileJava.dependsOn('protobuf', 'buildRustRelease')
+compileJava.dependsOn('protobuf')
 clean.dependsOn('cleanProtobuf', 'cleanRust')
 test.dependsOn('buildRust')
 testFfi.dependsOn('buildRust')
@@ -127,4 +127,3 @@ tasks.withType(Test) {
     }
     jvmArgs "-Djava.library.path=${projectDir}/../target/debug"
 }
-

java/client/src/main/java/glide/connectors/handlers/CallbackDispatcher.java

@@ -1,19 +1,25 @@
 package glide.connectors.handlers;
 
+import glide.api.models.exceptions.ClosingException;
+import glide.api.models.exceptions.ConnectionException;
+import glide.api.models.exceptions.ExecAbortException;
+import glide.api.models.exceptions.RequestException;
+import glide.api.models.exceptions.TimeoutException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.atomic.AtomicInteger;
 import lombok.RequiredArgsConstructor;
 import org.apache.commons.lang3.tuple.Pair;
+import response.ResponseOuterClass.RequestError;
 import response.ResponseOuterClass.Response;
 
 /** Holder for resources required to dispatch responses and used by {@link ReadHandler}. */
 @RequiredArgsConstructor
 public class CallbackDispatcher {
 
     /** Unique request ID (callback ID). Thread-safe and overflow-safe. */
-    private final AtomicInteger nextAvailableRequestId = new AtomicInteger(0);
+    protected final AtomicInteger nextAvailableRequestId = new AtomicInteger(0);
 
     /**
      * Storage of Futures to handle responses. Map key is callback id, which starts from 0. The value
@@ -23,7 +29,7 @@ public class CallbackDispatcher {
      * Negative Java values would be shown as positive on Rust side. There is no data loss, because
      * callback ID remains unique.
      */
-    private final ConcurrentHashMap<Integer, CompletableFuture<Response>> responses =
+    protected final ConcurrentHashMap<Integer, CompletableFuture<Response>> responses =
             new ConcurrentHashMap<>();
 
     /**
@@ -32,7 +38,7 @@ public class CallbackDispatcher {
      */
     // TODO: Optimize to avoid growing up to 2e32 (16 Gb)
     // https://github.com/aws/glide-for-redis/issues/704
-    private final ConcurrentLinkedQueue<Integer> freeRequestIds = new ConcurrentLinkedQueue<>();
+    protected final ConcurrentLinkedQueue<Integer> freeRequestIds = new ConcurrentLinkedQueue<>();
 
     /**
      * Register a new request to be sent. Once response received, the given future completes with it.
@@ -58,21 +64,62 @@ public CompletableFuture<Response> registerConnection() {
     }
 
     /**
-     * Complete the corresponding client promise and free resources.
+     * Complete the corresponding client promise, handle error and free resources.
      *
      * @param response A response received
      */
     public void completeRequest(Response response) {
+        if (response.hasClosingError()) {
+            // According to https://github.com/aws/glide-for-redis/issues/851
+            // a response with a closing error may arrive with any/random callback ID (usually -1)
+            // CommandManager and ConnectionManager would close the UDS channel on ClosingException
+            responses
+                    .values()
+                    .forEach(f -> f.completeExceptionally(new ClosingException(response.getClosingError())));
+            responses.clear();
+            return;
+        }
         // Complete and return the response at callbackId
         // free up the callback ID in the freeRequestIds list
         int callbackId = response.getCallbackIdx();
         CompletableFuture<Response> future = responses.remove(callbackId);
-        freeRequestIds.add(callbackId);
         if (future != null) {
+            freeRequestIds.add(callbackId);
+            if (response.hasRequestError()) {
+                RequestError error = response.getRequestError();
+                String msg = error.getMessage();
+                switch (error.getType()) {
+                    case Unspecified:
+                        // Unspecified error on Redis service-side
+                        future.completeExceptionally(new RequestException(msg));
+                    case ExecAbort:
+                        // Transactional error on Redis service-side
+                        future.completeExceptionally(new ExecAbortException(msg));
+                    case Timeout:
+                        // Timeout from Glide to Redis service
+                        future.completeExceptionally(new TimeoutException(msg));
+                    case Disconnect:
+                        // Connection problem between Glide and Redis
+                        future.completeExceptionally(new ConnectionException(msg));
+                    default:
+                        // Request or command error from Redis
+                        future.completeExceptionally(new RequestException(msg));
+                }
+            }
             future.completeAsync(() -> response);
         } else {
-            // TODO: log an error.
+            // TODO: log an error thru logger.
             // probably a response was received after shutdown or `registerRequest` call was missing
+            System.err.printf(
+                    "Received a response for not registered callback id %d, request error = %s%n",
+                    callbackId, response.getRequestError());
+            responses
+                    .values()
+                    .forEach(
+                            f ->
+                                    f.completeExceptionally(
+                                            new ClosingException("Client is in an erroneous state and should close")));
+            responses.clear();
         }
     }
 

java/client/src/main/java/glide/connectors/handlers/ChannelHandler.java

@@ -24,8 +24,8 @@ public class ChannelHandler {
 
     private static final String THREAD_POOL_NAME = "glide-channel";
 
-    private final Channel channel;
-    private final CallbackDispatcher callbackDispatcher;
+    protected final Channel channel;
+    protected final CallbackDispatcher callbackDispatcher;
 
     /** Open a new channel for a new client. */
     public ChannelHandler(CallbackDispatcher callbackDispatcher, String socketPath)

java/client/src/main/java/glide/managers/BaseCommandResponseResolver.java

@@ -1,13 +1,7 @@
 package glide.managers;
 
-import glide.api.models.exceptions.ClosingException;
-import glide.api.models.exceptions.ConnectionException;
-import glide.api.models.exceptions.ExecAbortException;
 import glide.api.models.exceptions.RedisException;
-import glide.api.models.exceptions.RequestException;
-import glide.api.models.exceptions.TimeoutException;
 import lombok.AllArgsConstructor;
-import response.ResponseOuterClass.RequestError;
 import response.ResponseOuterClass.Response;
 
 /**
@@ -20,40 +14,15 @@ public class BaseCommandResponseResolver
     private RedisExceptionCheckedFunction<Long, Object> respPointerResolver;
 
     /**
-     * Extracts value from the RESP pointer. <br>
-     * Throws errors when the response is unsuccessful.
+     * Extracts value from the RESP pointer.
      *
-     * @return A generic Object with the Response | null if the response is empty
+     * @return A generic Object with the Response or null if the response is empty
      */
     public Object apply(Response response) throws RedisException {
-        if (response.hasRequestError()) {
-            RequestError error = response.getRequestError();
-            String msg = error.getMessage();
-            switch (error.getType()) {
-                case Unspecified:
-                    // Unspecified error on Redis service-side
-                    throw new RequestException(msg);
-                case ExecAbort:
-                    // Transactional error on Redis service-side
-                    throw new ExecAbortException(msg);
-                case Timeout:
-                    // Timeout from Glide to Redis service
-                    throw new TimeoutException(msg);
-                case Disconnect:
-                    // Connection problem between Glide and Redis
-                    throw new ConnectionException(msg);
-                default:
-                    // Request or command error from Redis
-                    throw new RequestException(msg);
-            }
-        }
-        if (response.hasClosingError()) {
-            // A closing error is thrown when Rust-core is not connected to Redis
-            // We want to close shop and throw a ClosingException
-            // TODO: close the channel on a closing error
-            // channel.close();
-            throw new ClosingException(response.getClosingError());
-        }
+        // Note: errors are already handled before in CallbackDispatcher
+        assert !response.hasClosingError() : "Unhandled response closing error";
+        assert !response.hasRequestError() : "Unhandled response request error";
+
         if (response.hasConstantResponse()) {
             // Return "OK"
             return response.getConstantResponse().toString();

java/client/src/main/java/glide/managers/CommandManager.java

@@ -1,5 +1,6 @@
 package glide.managers;
 
+import glide.api.models.exceptions.ClosingException;
 import glide.connectors.handlers.ChannelHandler;
 import glide.managers.models.Command;
 import java.util.concurrent.CompletableFuture;
@@ -31,9 +32,27 @@ public <T> CompletableFuture<T> submitNewCommand(
         // when complete, convert the response to our expected type T using the given responseHandler
         return channel
                 .write(prepareRedisRequest(command.getRequestType(), command.getArguments()), true)
+                .exceptionally(this::exceptionHandler)
                 .thenApplyAsync(response -> responseHandler.apply(response));
     }
 
+    /**
+     * Exception handler for future pipeline.
+     *
+     * @param e An exception thrown in the pipeline before
+     * @return Nothing, it rethrows the exception
+     */
+    private Response exceptionHandler(Throwable e) {
+        if (e instanceof ClosingException) {
+            channel.close();
+        }
+        if (e instanceof RuntimeException) {
+            // RedisException also goes here
+            throw (RuntimeException) e;
+        }
+        throw new RuntimeException(e);
+    }
+
     /**
      * Build a protobuf command/transaction request object.<br>
      * Used by {@link CommandManager}.

java/client/src/main/java/glide/managers/ConnectionManager.java

@@ -12,10 +12,8 @@
 import glide.api.models.exceptions.ClosingException;
 import glide.connectors.handlers.ChannelHandler;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import lombok.RequiredArgsConstructor;
-import response.ResponseOuterClass.RequestError;
 import response.ResponseOuterClass.Response;
 
 /**
@@ -38,7 +36,25 @@ public class ConnectionManager {
      */
     public CompletableFuture<Void> connectToRedis(BaseClientConfiguration configuration) {
         ConnectionRequest request = createConnectionRequest(configuration);
-        return channel.connect(request).thenApplyAsync(this::checkGlideRsResponse);
+        return channel
+                .connect(request)
+                .exceptionally(this::exceptionHandler)
+                .thenApplyAsync(this::checkGlideRsResponse);
+    }
+
+    /**
+     * Exception handler for future pipeline.
+     *
+     * @param e An exception thrown in the pipeline before
+     * @return Nothing, it always rethrows the exception
+     */
+    private Response exceptionHandler(Throwable e) {
+        channel.close();
+        if (e instanceof RuntimeException) {
+            // RedisException also goes here
+            throw (RuntimeException) e;
+        }
+        throw new RuntimeException(e);
     }
 
     /**
@@ -154,15 +170,15 @@ private ConnectionRequestOuterClass.ReadFrom mapReadFromEnum(ReadFrom readFrom)
 
     /** Check a response received from Glide. */
     private Void checkGlideRsResponse(Response response) {
+        // Note: errors are already handled before in CallbackDispatcher, but we double-check
         if (response.hasRequestError()) {
-            RequestError error = response.getRequestError();
-            throwClosingError("Unexpected request error in response: " + error.getMessage());
+            throwClosingError(
+                    "Unhandled request error in response: " + response.getRequestError().getMessage());
         }
         if (response.hasClosingError()) {
-            // A closing error is thrown when Rust-core is not connected to Redis
-            // We want to close shop and throw a ClosingException
-            throwClosingError(response.getClosingError());
+            throwClosingError("Unhandled closing error in response: " + response.getClosingError());
         }
+
         if (response.hasRespPointer()) {
             throwClosingError("Unexpected data in response");
         }
@@ -173,12 +189,8 @@ private Void checkGlideRsResponse(Response response) {
         return null;
     }
 
-    private void throwClosingError(String msg) throws ClosingException {
-        try {
-            closeConnection().get();
-        } catch (InterruptedException | ExecutionException exception) {
-            throw new RuntimeException(exception);
-        }
+    private void throwClosingError(String msg) {
+        closeConnection();
         throw new ClosingException(msg);
     }
 
@@ -188,6 +200,6 @@ private void throwClosingError(String msg) throws ClosingException {
      * @return a CompletableFuture to indicate the channel is closed
      */
     public Future<Void> closeConnection() {
-        return channel.close().syncUninterruptibly();
+        return channel.close();
     }
 }