[client] Check whether request is removed from inflight requests to avoid NullPointerException when connection is closed. (#2127)

Author: vamossagar12

fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java

@@ -166,7 +166,9 @@ private CompletableFuture<Void> close(Throwable cause) {
             // notify all the inflight requests
             for (int requestId : inflightRequests.keySet()) {
                 InflightRequest request = inflightRequests.remove(requestId);
-                request.responseFuture.completeExceptionally(requestCause);
+                if (request != null) {
+                    request.responseFuture.completeExceptionally(requestCause);
+                }
             }
 
             // notify all the pending requests
@@ -249,6 +251,12 @@ public void onFailure(Throwable cause) {
     private void establishConnection(ChannelFuture future, boolean isInnerClient) {
         synchronized (lock) {
             if (future.isSuccess()) {
+                if (state.isDisconnected()) {
+                    LOG.debug(
+                            "Connection established to {} but connection is already closed.", node);
+                    future.channel().close();
+                    return;
+                }
                 LOG.debug("Established connection to server {}.", node);
                 channel = future.channel();
                 channel.pipeline()