Fix the active rpc metrics; Port [#17234] to branch-2.10

Cherry pick #17234 to branch-2.10

### What changes are proposed in this pull request?

mentioned in #16629 

### Why are the changes needed?

The `Cluster.ActiveRpcReadCount` metric could get an impossible value like a negative number. The main reason is that the request handling doesn't cover all situations.

### Does this PR introduce any user facing changes?

Nope


			pr-link: #18587
			change-id: cid-04f369309aaf5e17b37e13f30c07d2515ac4c599

core/server/worker/src/main/java/alluxio/worker/grpc/BlockReadHandler.java

@@ -464,29 +464,30 @@ private void runInternal() {
         }
         continue;
       }
-      if (error != null) {
+      if (eof || cancel || error != null) {
         try {
-          completeRequest(mContext, false);
-        } catch (Exception e) {
-          LOG.error("Failed to close the request.", e);
-        }
-        replyError(error);
-      } else if (eof || cancel) {
-        try {
-          boolean success = !cancel;
+          boolean success = !cancel && error == null;
           completeRequest(mContext, success);
         } catch (Exception e) {
-          LogUtils.warnWithException(LOG, "Exception occurred while completing read request, "
-                  + "EOF/CANCEL sessionId: {}. {}", mContext.getRequest().getSessionId(),
-              mContext.getRequest(), e);
-          setError(new Error(AlluxioStatusException.fromThrowable(e), true));
+          if (error != null) {
+            LOG.error("Failed to close the request.", e);
+          } else {
+            LogUtils.warnWithException(LOG, "Exception occurred while completing read request, "
+                            + "EOF/CANCEL sessionId: {}. {}", mContext.getRequest().getSessionId(),
+                    mContext.getRequest(), e);
+            error = new Error(AlluxioStatusException.fromThrowable(e), true);
+          }
         }
-        if (eof) {
+        if (error != null) {
+          replyError(error);
+        } else if (eof) {
           replyEof();
-        } else {
+        } else if (cancel) {
           replyCancel();
         }
       }
+      // Leave `!mResponse.isReady() && tooManyPendingChunks()` unhandled
+      // since the reader is not finished in that case and needs more rounds
     }
 
     /**