[SPARK-53248][CORE] Support checkedCast in JavaUtils

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

This PR aims to support `checkedCast` in `JavaUtils`.

In addition, new Scalastyle and Checkstyle rules are added to prevent future regressions.

### Why are the changes needed?

To improve Spark's Java utility features.

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

No.

### How was this patch tested?

Pass the CIs.

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

No.

Closes #51977 from dongjoon-hyun/SPARK-53248.

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/util/TransportConf.java

@@ -22,7 +22,6 @@
 import java.util.Properties;
 import java.util.concurrent.TimeUnit;
 import com.google.common.base.Preconditions;
-import com.google.common.primitives.Ints;
 import io.netty.util.NettyRuntime;
 
 /**
@@ -171,7 +170,7 @@ public int ioRetryWaitTimeMs() {
    * memory mapping has high overhead for blocks close to or below the page size of the OS.
    */
   public int memoryMapBytes() {
-    return Ints.checkedCast(JavaUtils.byteStringAsBytes(
+    return JavaUtils.checkedCast(JavaUtils.byteStringAsBytes(
       conf.get("spark.storage.memoryMapThreshold", "2m")));
   }
 
@@ -248,7 +247,7 @@ public boolean saslEncryption() {
    * Maximum number of bytes to be encrypted at a time when SASL encryption is used.
    */
   public int maxSaslEncryptedBlockSize() {
-    return Ints.checkedCast(JavaUtils.byteStringAsBytes(
+    return JavaUtils.checkedCast(JavaUtils.byteStringAsBytes(
       conf.get("spark.network.sasl.maxEncryptedBlockSize", "64k")));
   }
 
@@ -263,7 +262,7 @@ public boolean saslServerAlwaysEncrypt() {
    * When Secure (SSL/TLS) Shuffle is enabled, the Chunk size to use for shuffling files.
    */
   public int sslShuffleChunkSize() {
-    return Ints.checkedCast(JavaUtils.byteStringAsBytes(
+    return JavaUtils.checkedCast(JavaUtils.byteStringAsBytes(
       conf.get("spark.network.ssl.maxEncryptedBlockSize", "64k")));
   }
 
@@ -567,7 +566,7 @@ public String mergedShuffleFileManagerImpl() {
    * service unnecessarily.
    */
   public int minChunkSizeInMergedShuffleFile() {
-    return Ints.checkedCast(JavaUtils.byteStringAsBytes(
+    return JavaUtils.checkedCast(JavaUtils.byteStringAsBytes(
       conf.get("spark.shuffle.push.server.minChunkSizeInMergedShuffleFile", "2m")));
   }
 

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

@@ -19,7 +19,6 @@
 
 import java.util.List;
 
-import com.google.common.primitives.Ints;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.FileRegion;
@@ -44,6 +43,7 @@
 import org.apache.spark.network.protocol.StreamRequest;
 import org.apache.spark.network.protocol.StreamResponse;
 import org.apache.spark.network.util.ByteArrayWritableChannel;
+import org.apache.spark.network.util.JavaUtils;
 import org.apache.spark.network.util.NettyUtils;
 
 public class ProtocolSuite {
@@ -115,7 +115,8 @@ private static class FileRegionEncoder extends MessageToMessageEncoder<FileRegio
     public void encode(ChannelHandlerContext ctx, FileRegion in, List<Object> out)
       throws Exception {
 
-      ByteArrayWritableChannel channel = new ByteArrayWritableChannel(Ints.checkedCast(in.count()));
+      ByteArrayWritableChannel channel =
+        new ByteArrayWritableChannel(JavaUtils.checkedCast(in.count()));
       while (in.transferred() < in.count()) {
         in.transferTo(channel, in.transferred());
       }

common/unsafe/src/main/java/org/apache/spark/unsafe/types/ByteArray.java

@@ -20,9 +20,8 @@
 import java.nio.ByteOrder;
 import java.util.Arrays;
 
-import com.google.common.primitives.Ints;
-
 import org.apache.spark.unsafe.Platform;
+import org.apache.spark.network.util.JavaUtils;
 
 public final class ByteArray {
 
@@ -169,7 +168,7 @@ public static byte[] concatWS(byte[] delimiter, byte[]... inputs) {
     }
     if (totalLength > 0) totalLength -= delimiter.length;
     // Allocate a new byte array, and copy the inputs one by one into it
-    final byte[] result = new byte[Ints.checkedCast(totalLength)];
+    final byte[] result = new byte[JavaUtils.checkedCast(totalLength)];
     int offset = 0;
     for (int i = 0; i < inputs.length; i++) {
       byte[] input = inputs[i];

common/utils/src/main/java/org/apache/spark/network/util/JavaUtils.java

@@ -656,4 +656,11 @@ public static String stackTraceToString(Throwable t) {
     }
     return out.toString(StandardCharsets.UTF_8);
   }
+
+  public static int checkedCast(long value) {
+    if (value > Integer.MAX_VALUE || value < Integer.MIN_VALUE) {
+      throw new IllegalArgumentException("Cannot cast to integer.");
+    }
+    return (int) value;
+  }
 }

core/src/main/java/org/apache/spark/util/collection/unsafe/sort/RadixSort.java

@@ -17,10 +17,9 @@
 
 package org.apache.spark.util.collection.unsafe.sort;
 
-import com.google.common.primitives.Ints;
-
 import org.apache.spark.unsafe.Platform;
 import org.apache.spark.unsafe.array.LongArray;
+import org.apache.spark.network.util.JavaUtils;
 
 public class RadixSort {
 
@@ -63,7 +62,7 @@ public static int sort(
         }
       }
     }
-    return Ints.checkedCast(inIndex);
+    return JavaUtils.checkedCast(inIndex);
   }
 
   /**
@@ -204,7 +203,7 @@ public static int sortKeyPrefixArray(
         }
       }
     }
-    return Ints.checkedCast(inIndex);
+    return JavaUtils.checkedCast(inIndex);
   }
 
   /**

core/src/test/scala/org/apache/spark/util/collection/unsafe/sort/RadixSortSuite.scala

@@ -22,9 +22,8 @@ import java.util.{Arrays, Comparator}
 
 import scala.util.Random
 
-import com.google.common.primitives.Ints
-
 import org.apache.spark.SparkFunSuite
+import org.apache.spark.network.util.JavaUtils.checkedCast
 import org.apache.spark.unsafe.array.LongArray
 import org.apache.spark.unsafe.memory.MemoryBlock
 import org.apache.spark.util.collection.Sorter
@@ -75,21 +74,21 @@ class RadixSortSuite extends SparkFunSuite {
       2, 4, false, false, true))
 
   private def generateTestData(size: Long, rand: => Long): (Array[JLong], LongArray) = {
-    val ref = Array.tabulate[Long](Ints.checkedCast(size)) { i => rand }
-    val extended = ref ++ createArray(Ints.checkedCast(size), 0L)
+    val ref = Array.tabulate[Long](checkedCast(size)) { i => rand }
+    val extended = ref ++ createArray(checkedCast(size), 0L)
     (ref.map(i => JLong.valueOf(i)), new LongArray(MemoryBlock.fromLongArray(extended)))
   }
 
   private def generateKeyPrefixTestData(size: Long, rand: => Long): (LongArray, LongArray) = {
-    val ref = Array.tabulate[Long](Ints.checkedCast(size * 2)) { i => rand }
-    val extended = ref ++ createArray(Ints.checkedCast(size * 2), 0L)
+    val ref = Array.tabulate[Long](checkedCast(size * 2)) { i => rand }
+    val extended = ref ++ createArray(checkedCast(size * 2), 0L)
     (new LongArray(MemoryBlock.fromLongArray(ref)),
      new LongArray(MemoryBlock.fromLongArray(extended)))
   }
 
   private def collectToArray(array: LongArray, offset: Int, length: Long): Array[Long] = {
     var i = 0
-    val out = new Array[Long](Ints.checkedCast(length))
+    val out = new Array[Long](checkedCast(length))
     while (i < length) {
       out(i) = array.get(offset + i)
       i += 1
@@ -112,7 +111,7 @@ class RadixSortSuite extends SparkFunSuite {
       refCmp: PrefixComparator): Unit = {
     val sortBuffer = new LongArray(MemoryBlock.fromLongArray(new Array[Long](buf.size().toInt)))
     new Sorter(new UnsafeSortDataFormat(sortBuffer)).sort(
-      buf, Ints.checkedCast(lo), Ints.checkedCast(hi),
+      buf, checkedCast(lo), checkedCast(hi),
       (r1: RecordPointerAndKeyPrefix, r2: RecordPointerAndKeyPrefix) =>
         refCmp.compare(r1.keyPrefix, r2.keyPrefix))
   }

dev/checkstyle.xml

@@ -199,6 +199,10 @@
             <property name="illegalClasses" value="com.google.common.io.BaseEncoding" />
             <property name="illegalClasses" value="com.google.common.io.Files" />
         </module>
+        <module name="RegexpSinglelineJava">
+            <property name="format" value="Ints\.checkedCast"/>
+            <property name="message" value="Use JavaUtils.checkedCast instead." />
+        </module>
         <module name="RegexpSinglelineJava">
             <property name="format" value="Charset\.defaultCharset"/>
             <property name="message" value="Use StandardCharsets.UTF_8 instead." />

scalastyle-config.xml

@@ -826,4 +826,9 @@ This file is divided into 3 sections:
     <parameters><parameter name="regex">\bPreconditions\.checkNotNull\b</parameter></parameters>
     <customMessage>Use requireNonNull of java.util.Objects instead.</customMessage>
   </check>
+
+  <check customId="intscheckedcast" level="error" class="org.scalastyle.file.RegexChecker" enabled="true">
+    <parameters><parameter name="regex">\bInts\.checkedCast\b</parameter></parameters>
+    <customMessage>Use JavaUtils.checkedCast instead.</customMessage>
+  </check>
 </scalastyle>