[SPARK-51023][CORE] log remote address on RPC exception

### What changes were proposed in this pull request?
Add the remote address to the RPC exception log lines. It's already logged for `TransportRequestHandler.processStreamRequest()`, but not for other types of requests.

### Why are the changes needed?
To simplify troubleshooting. We hit this in our production deployments for two cases:
* when the executor and driver are running different Spark versions
* when vulnerability scanner (internal security tool) sends malformed messages

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
By verifying the logs of `RpcIntegrationSuite` -
```
25/01/28 15:06:07.439 shuffle-server-3-15 ERROR TransportRequestHandler: Error while invoking RpcHandler#receive() on RPC id 5952848813688034638 from /127.0.0.1:61697
java.lang.RuntimeException: Thrown: the
	at org.apache.spark.network.RpcIntegrationSuite$1.receive(RpcIntegrationSuite.java:73)
	at org.apache.spark.network.server.TransportRequestHandler.processRpcRequest(TransportRequestHandler.java:167)
	at org.apache.spark.network.server.TransportRequestHandler.handle(TransportRequestHandler.java:111)
```

### Was this patch authored or co-authored using generative AI tooling?
No

Closes apache#49718 from fe2s/SPARK-51023-log_remote_address_on_RPC_exception.

Authored-by: oleksii.diagiliev <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: oleksii.diagiliev

common/network-common/src/main/java/org/apache/spark/network/server/TransportRequestHandler.java

@@ -176,8 +176,9 @@ public void onFailure(Throwable e) {
         }
       });
     } catch (Exception e) {
-      logger.error("Error while invoking RpcHandler#receive() on RPC id {}", e,
-        MDC.of(LogKeys.REQUEST_ID$.MODULE$, req.requestId));
+      logger.error("Error while invoking RpcHandler#receive() on RPC id {} from {}", e,
+        MDC.of(LogKeys.REQUEST_ID$.MODULE$, req.requestId),
+        MDC.of(LogKeys.HOST_PORT$.MODULE$, getRemoteAddress(channel)));
       respond(new RpcFailure(req.requestId, Throwables.getStackTraceAsString(e)));
     } finally {
       req.body().release();
@@ -262,8 +263,9 @@ public String getID() {
         respond(new RpcResponse(req.requestId,
           new NioManagedBuffer(blockPushNonFatalFailure.getResponse())));
       } else {
-        logger.error("Error while invoking RpcHandler#receive() on RPC id {}", e,
-          MDC.of(LogKeys.REQUEST_ID$.MODULE$, req.requestId));
+        logger.error("Error while invoking RpcHandler#receive() on RPC id {} from {}", e,
+          MDC.of(LogKeys.REQUEST_ID$.MODULE$, req.requestId),
+          MDC.of(LogKeys.HOST_PORT$.MODULE$, getRemoteAddress(channel)));
         respond(new RpcFailure(req.requestId, Throwables.getStackTraceAsString(e)));
       }
       // We choose to totally fail the channel, rather than trying to recover as we do in other
@@ -279,7 +281,8 @@ private void processOneWayMessage(OneWayMessage req) {
     try {
       rpcHandler.receive(reverseClient, req.body().nioByteBuffer());
     } catch (Exception e) {
-      logger.error("Error while invoking RpcHandler#receive() for one-way message.", e);
+      logger.error("Error while invoking RpcHandler#receive() for one-way message from {}.", e,
+        MDC.of(LogKeys.HOST_PORT$.MODULE$, getRemoteAddress(channel)));
     } finally {
       req.body().release();
     }
@@ -304,9 +307,10 @@ public void onFailure(Throwable e) {
       });
     } catch (Exception e) {
       logger.error("Error while invoking receiveMergeBlockMetaReq() for appId {} shuffleId {} "
-        + "reduceId {}", e, MDC.of(LogKeys.APP_ID$.MODULE$, req.appId),
+        + "reduceId {} from {}", e, MDC.of(LogKeys.APP_ID$.MODULE$, req.appId),
           MDC.of(LogKeys.SHUFFLE_ID$.MODULE$, req.shuffleId),
-          MDC.of(LogKeys.REDUCE_ID$.MODULE$, req.reduceId));
+          MDC.of(LogKeys.REDUCE_ID$.MODULE$, req.reduceId),
+          MDC.of(LogKeys.HOST_PORT$.MODULE$, getRemoteAddress(channel)));
       respond(new RpcFailure(req.requestId, Throwables.getStackTraceAsString(e)));
     }
   }