[SPARK-52874][CORE] Support o.a.s.util.Pair Java Record

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

This PR aims to add `o.a.s.util.Pair` Java Record for Java code and to enforce it.

```scala
scala> val pair = org.apache.spark.util.Pair.of(1,2);
val pair: org.apache.spark.util.Pair[Int,Int] = Pair[getLeft=1, getRight=2]

scala> pair.getLeft()
val res0: Int = 1

scala> pair.getRight()
val res1: Int = 2
```

### Why are the changes needed?

Since Java 16, we can use `Records` officially in Java.
| VERSION | JEP |
| - | - |
| Java 14 | [JEP 359: Records (Preview)](https://openjdk.org/jeps/359) |
| Java 15 | [JEP 384: Records (Second Preview)](https://openjdk.org/jeps/384) |
| Java 16 | [JEP 395: Records](https://openjdk.org/jeps/395) |

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

No behavior change.

### How was this patch tested?

Pass the CIs to make it sure all tests passed.

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

No.

Closes apache#51568 from dongjoon-hyun/SPARK-52874.

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

Author: dongjoon-hyun

common/network-common/src/main/java/org/apache/spark/network/client/TransportResponseHandler.java

@@ -26,8 +26,6 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import io.netty.channel.Channel;
-import org.apache.commons.lang3.tuple.ImmutablePair;
-import org.apache.commons.lang3.tuple.Pair;
 
 import org.apache.spark.internal.SparkLogger;
 import org.apache.spark.internal.SparkLoggerFactory;
@@ -45,6 +43,7 @@
 import org.apache.spark.network.server.MessageHandler;
 import static org.apache.spark.network.util.NettyUtils.getRemoteAddress;
 import org.apache.spark.network.util.TransportFrameDecoder;
+import org.apache.spark.util.Pair;
 
 /**
  * Handler that processes server responses, in response to requests issued from a
@@ -96,7 +95,7 @@ public void removeRpcRequest(long requestId) {
 
   public void addStreamCallback(String streamId, StreamCallback callback) {
     updateTimeOfLastRequest();
-    streamCallbacks.offer(ImmutablePair.of(streamId, callback));
+    streamCallbacks.offer(Pair.of(streamId, callback));
   }
 
   @VisibleForTesting
@@ -125,7 +124,7 @@ private void failOutstandingRequests(Throwable cause) {
     }
     for (Pair<String, StreamCallback> entry : streamCallbacks) {
       try {
-        entry.getValue().onFailure(entry.getKey(), cause);
+        entry.getRight().onFailure(entry.getLeft(), cause);
       } catch (Exception e) {
         logger.warn("StreamCallback.onFailure throws exception", e);
       }
@@ -236,7 +235,7 @@ public void handle(ResponseMessage message) throws Exception {
     } else if (message instanceof StreamResponse resp) {
       Pair<String, StreamCallback> entry = streamCallbacks.poll();
       if (entry != null) {
-        StreamCallback callback = entry.getValue();
+        StreamCallback callback = entry.getRight();
         if (resp.byteCount > 0) {
           StreamInterceptor<ResponseMessage> interceptor = new StreamInterceptor<>(
             this, resp.streamId, resp.byteCount, callback);
@@ -262,7 +261,7 @@ public void handle(ResponseMessage message) throws Exception {
     } else if (message instanceof StreamFailure resp) {
       Pair<String, StreamCallback> entry = streamCallbacks.poll();
       if (entry != null) {
-        StreamCallback callback = entry.getValue();
+        StreamCallback callback = entry.getRight();
         try {
           callback.onFailure(resp.streamId, new RuntimeException(resp.error));
         } catch (IOException ioe) {

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

@@ -26,13 +26,12 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import io.netty.channel.Channel;
-import org.apache.commons.lang3.tuple.ImmutablePair;
-import org.apache.commons.lang3.tuple.Pair;
 
 import org.apache.spark.internal.SparkLogger;
 import org.apache.spark.internal.SparkLoggerFactory;
 import org.apache.spark.network.buffer.ManagedBuffer;
 import org.apache.spark.network.client.TransportClient;
+import org.apache.spark.util.Pair;
 
 /**
  * StreamManager which allows registration of an Iterator<ManagedBuffer>, which are
@@ -127,7 +126,7 @@ public static Pair<Long, Integer> parseStreamChunkId(String streamChunkId) {
       "Stream id and chunk index should be specified.";
     long streamId = Long.valueOf(array[0]);
     int chunkIndex = Integer.valueOf(array[1]);
-    return ImmutablePair.of(streamId, chunkIndex);
+    return Pair.of(streamId, chunkIndex);
   }
 
   @Override

common/network-common/src/test/java/org/apache/spark/network/ChunkFetchRequestHandlerSuite.java

@@ -27,15 +27,14 @@
 
 import static org.mockito.Mockito.*;
 
-import org.apache.commons.lang3.tuple.ImmutablePair;
-import org.apache.commons.lang3.tuple.Pair;
 import org.apache.spark.network.buffer.ManagedBuffer;
 import org.apache.spark.network.client.TransportClient;
 import org.apache.spark.network.protocol.*;
 import org.apache.spark.network.server.ChunkFetchRequestHandler;
 import org.apache.spark.network.server.NoOpRpcHandler;
 import org.apache.spark.network.server.OneForOneStreamManager;
 import org.apache.spark.network.server.RpcHandler;
+import org.apache.spark.util.Pair;
 
 public class ChunkFetchRequestHandlerSuite {
 
@@ -54,7 +53,7 @@ public void handleChunkFetchRequest() throws Exception {
       .thenAnswer(invocationOnMock0 -> {
         Object response = invocationOnMock0.getArguments()[0];
         ExtendedChannelPromise channelFuture = new ExtendedChannelPromise(channel);
-        responseAndPromisePairs.add(ImmutablePair.of(response, channelFuture));
+        responseAndPromisePairs.add(Pair.of(response, channelFuture));
         return channelFuture;
       });
 

common/network-common/src/test/java/org/apache/spark/network/RpcIntegrationSuite.java

@@ -26,8 +26,6 @@
 
 import com.google.common.collect.Sets;
 import com.google.common.io.Files;
-import org.apache.commons.lang3.tuple.ImmutablePair;
-import org.apache.commons.lang3.tuple.Pair;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
@@ -41,6 +39,7 @@
 import org.apache.spark.network.util.JavaUtils;
 import org.apache.spark.network.util.MapConfigProvider;
 import org.apache.spark.network.util.TransportConf;
+import org.apache.spark.util.Pair;
 
 public class RpcIntegrationSuite {
   static TransportConf conf;
@@ -408,7 +407,7 @@ private Pair<Set<String>, Set<String>> checkErrorsContain(
         notFound.add(contain);
       }
     }
-    return new ImmutablePair<>(remainingErrors, notFound);
+    return new Pair<>(remainingErrors, notFound);
   }
 
   private static class VerifyingStreamCallback implements StreamCallbackWithID {

common/network-common/src/test/java/org/apache/spark/network/TransportRequestHandlerSuite.java

@@ -28,8 +28,6 @@
 import static org.junit.jupiter.api.Assertions.*;
 import static org.mockito.Mockito.*;
 
-import org.apache.commons.lang3.tuple.ImmutablePair;
-import org.apache.commons.lang3.tuple.Pair;
 import org.apache.spark.network.buffer.ManagedBuffer;
 import org.apache.spark.network.client.RpcResponseCallback;
 import org.apache.spark.network.client.TransportClient;
@@ -39,6 +37,7 @@
 import org.apache.spark.network.server.RpcHandler;
 import org.apache.spark.network.server.StreamManager;
 import org.apache.spark.network.server.TransportRequestHandler;
+import org.apache.spark.util.Pair;
 
 public class TransportRequestHandlerSuite {
 
@@ -53,7 +52,7 @@ public void handleStreamRequest() throws Exception {
       .thenAnswer(invocationOnMock0 -> {
         Object response = invocationOnMock0.getArguments()[0];
         ExtendedChannelPromise channelFuture = new ExtendedChannelPromise(channel);
-        responseAndPromisePairs.add(ImmutablePair.of(response, channelFuture));
+        responseAndPromisePairs.add(Pair.of(response, channelFuture));
         return channelFuture;
       });
 
@@ -145,7 +144,7 @@ public MergedBlockMetaReqHandler getMergedBlockMetaReqHandler() {
     when(channel.writeAndFlush(any())).thenAnswer(invocationOnMock0 -> {
       Object response = invocationOnMock0.getArguments()[0];
       ExtendedChannelPromise channelFuture = new ExtendedChannelPromise(channel);
-      responseAndPromisePairs.add(ImmutablePair.of(response, channelFuture));
+      responseAndPromisePairs.add(Pair.of(response, channelFuture));
       return channelFuture;
     });
 

common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolver.java

@@ -28,7 +28,6 @@
 
 import org.apache.commons.lang3.builder.ToStringBuilder;
 import org.apache.commons.lang3.builder.ToStringStyle;
-import org.apache.commons.lang3.tuple.Pair;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -56,6 +55,7 @@
 import org.apache.spark.network.util.JavaUtils;
 import org.apache.spark.network.util.NettyUtils;
 import org.apache.spark.network.util.TransportConf;
+import org.apache.spark.util.Pair;
 
 /**
  * Manages converting shuffle BlockIds into physical segments of local files, from a process outside
@@ -400,7 +400,7 @@ public Map<String, String[]> getLocalDirs(String appId, Set<String> execIds) {
         }
         return Pair.of(exec, info.localDirs);
       })
-      .collect(Collectors.toMap(Pair::getKey, Pair::getValue));
+      .collect(Collectors.toMap(Pair::getLeft, Pair::getRight));
   }
 
   /**

common/utils/src/main/java/org/apache/spark/util/Pair.java

@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.util;
+
+/**
+ * An immutable pair of values. Note that the fields are intentionally designed to be `getLeft` and
+ * `getRight` instead of `left` and `right` in order to mitigate the migration burden
+ * from `org.apache.commons.lang3.tuple.Pair`.
+ */
+public record Pair<L, R>(L getLeft, R getRight) {
+  public static <L, R> Pair<L, R> of(L left, R right) {
+    return new Pair<>(left, right);
+  }
+}

common/utils/src/test/java/org/apache/spark/util/SparkLoggerSuiteBase.java

@@ -22,7 +22,6 @@
 import java.nio.file.Files;
 import java.util.List;
 
-import org.apache.commons.lang3.tuple.Pair;
 import org.apache.logging.log4j.Level;
 import org.junit.jupiter.api.Test;
 

dev/checkstyle.xml

@@ -183,6 +183,7 @@
         <module name="IllegalImport">
             <property name="illegalPkgs" value="org.apache.log4j" />
             <property name="illegalPkgs" value="org.apache.commons.lang" />
+            <property name="illegalPkgs" value="org.apache.commons.lang3.tuple" />
             <property name="illegalClasses" value="org.apache.commons.lang3.JavaVersion" />
         </module>
         <module name="RegexpSinglelineJava">

scalastyle-config.xml

@@ -294,6 +294,11 @@ This file is divided into 3 sections:
       Commons Lang 3 JavaVersion (org.apache.commons.lang3.JavaVersion)</customMessage>
   </check>
 
+  <check customId="commonslang3tuple" level="error" class="org.scalastyle.file.RegexChecker" enabled="true">
+    <parameters><parameter name="regex">org\.apache\.commons\.lang3\.tuple</parameter></parameters>
+    <customMessage>Use org.apache.spark.util.Pair instead</customMessage>
+  </check>
+
   <check customId="uribuilder" level="error" class="org.scalastyle.file.RegexChecker" enabled="true">
     <parameters><parameter name="regex">UriBuilder\.fromUri</parameter></parameters>
     <customMessage>Use Utils.getUriBuilder instead.</customMessage>