[SPARK-53242][CORE][DSTREAM] Move stackTraceToString to JavaUtils and use it to replace Throwables.getStackTraceAsString

### What changes were proposed in this pull request?
This pr move `stackTraceToString` to `JavaUtils` and use it to replace `Throwables.getStackTraceAsString`.

### Why are the changes needed?
Reuse Spark's existing error handling functions.

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

### How was this patch tested?
Pass Github Actions

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

Closes #51969 from LuciferYang/JavaUtils-stackTraceToString.

Authored-by: yangjie01 <yangjie01@baidu.com>
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

Author: LuciferYang

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

@@ -19,7 +19,6 @@
 
 import java.net.SocketAddress;
 
-import com.google.common.base.Throwables;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
@@ -36,6 +35,7 @@
 import org.apache.spark.network.protocol.ChunkFetchRequest;
 import org.apache.spark.network.protocol.ChunkFetchSuccess;
 import org.apache.spark.network.protocol.Encodable;
+import org.apache.spark.network.util.JavaUtils;
 
 import static org.apache.spark.network.util.NettyUtils.*;
 
@@ -114,7 +114,7 @@ public void processFetchRequest(
         MDC.of(LogKeys.STREAM_CHUNK_ID, msg.streamChunkId),
         MDC.of(LogKeys.HOST_PORT, getRemoteAddress(channel)));
       respond(channel, new ChunkFetchFailure(msg.streamChunkId,
-        Throwables.getStackTraceAsString(e)));
+        JavaUtils.stackTraceToString(e)));
       return;
     }
 

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

@@ -21,7 +21,6 @@
 import java.net.SocketAddress;
 import java.nio.ByteBuffer;
 
-import com.google.common.base.Throwables;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 
@@ -33,6 +32,7 @@
 import org.apache.spark.network.buffer.NioManagedBuffer;
 import org.apache.spark.network.client.*;
 import org.apache.spark.network.protocol.*;
+import org.apache.spark.network.util.JavaUtils;
 import org.apache.spark.network.util.TransportFrameDecoder;
 
 import static org.apache.spark.network.util.NettyUtils.getRemoteAddress;
@@ -145,7 +145,7 @@ private void processStreamRequest(final StreamRequest req) {
       logger.error("Error opening stream {} for request from {}", e,
         MDC.of(LogKeys.STREAM_ID, req.streamId),
         MDC.of(LogKeys.HOST_PORT, getRemoteAddress(channel)));
-      respond(new StreamFailure(req.streamId, Throwables.getStackTraceAsString(e)));
+      respond(new StreamFailure(req.streamId, JavaUtils.stackTraceToString(e)));
       return;
     }
 
@@ -172,14 +172,14 @@ public void onSuccess(ByteBuffer response) {
 
         @Override
         public void onFailure(Throwable e) {
-          respond(new RpcFailure(req.requestId, Throwables.getStackTraceAsString(e)));
+          respond(new RpcFailure(req.requestId, JavaUtils.stackTraceToString(e)));
         }
       });
     } catch (Exception e) {
       logger.error("Error while invoking RpcHandler#receive() on RPC id {} from {}", e,
         MDC.of(LogKeys.REQUEST_ID, req.requestId),
         MDC.of(LogKeys.HOST_PORT, getRemoteAddress(channel)));
-      respond(new RpcFailure(req.requestId, Throwables.getStackTraceAsString(e)));
+      respond(new RpcFailure(req.requestId, JavaUtils.stackTraceToString(e)));
     } finally {
       req.body().release();
     }
@@ -199,7 +199,7 @@ public void onSuccess(ByteBuffer response) {
 
         @Override
         public void onFailure(Throwable e) {
-          respond(new RpcFailure(req.requestId, Throwables.getStackTraceAsString(e)));
+          respond(new RpcFailure(req.requestId, JavaUtils.stackTraceToString(e)));
         }
       };
       TransportFrameDecoder frameDecoder = (TransportFrameDecoder)
@@ -266,7 +266,7 @@ public String getID() {
         logger.error("Error while invoking RpcHandler#receive() on RPC id {} from {}", e,
           MDC.of(LogKeys.REQUEST_ID, req.requestId),
           MDC.of(LogKeys.HOST_PORT, getRemoteAddress(channel)));
-        respond(new RpcFailure(req.requestId, Throwables.getStackTraceAsString(e)));
+        respond(new RpcFailure(req.requestId, JavaUtils.stackTraceToString(e)));
       }
       // We choose to totally fail the channel, rather than trying to recover as we do in other
       // cases.  We don't know how many bytes of the stream the client has already sent for the
@@ -302,7 +302,7 @@ public void onSuccess(int numChunks, ManagedBuffer buffer) {
           @Override
           public void onFailure(Throwable e) {
             logger.trace("Failed to send meta for {}", req);
-            respond(new RpcFailure(req.requestId, Throwables.getStackTraceAsString(e)));
+            respond(new RpcFailure(req.requestId, JavaUtils.stackTraceToString(e)));
           }
       });
     } catch (Exception e) {
@@ -311,7 +311,7 @@ public void onFailure(Throwable e) {
           MDC.of(LogKeys.SHUFFLE_ID, req.shuffleId),
           MDC.of(LogKeys.REDUCE_ID, req.reduceId),
           MDC.of(LogKeys.HOST_PORT, getRemoteAddress(channel)));
-      respond(new RpcFailure(req.requestId, Throwables.getStackTraceAsString(e)));
+      respond(new RpcFailure(req.requestId, JavaUtils.stackTraceToString(e)));
     }
   }
 

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

@@ -20,10 +20,9 @@
 import java.io.FileNotFoundException;
 import java.net.ConnectException;
 
-import com.google.common.base.Throwables;
-
 import org.apache.spark.annotation.Evolving;
 import org.apache.spark.network.server.BlockPushNonFatalFailure;
+import org.apache.spark.network.util.JavaUtils;
 
 /**
  * Plugs into {@link RetryingBlockTransferor} to further control when an exception should be retried
@@ -105,12 +104,12 @@ class BlockFetchErrorHandler implements ErrorHandler {
 
     @Override
     public boolean shouldRetryError(Throwable t) {
-      return !Throwables.getStackTraceAsString(t).contains(STALE_SHUFFLE_BLOCK_FETCH);
+      return !JavaUtils.stackTraceToString(t).contains(STALE_SHUFFLE_BLOCK_FETCH);
     }
 
     @Override
     public boolean shouldLogError(Throwable t) {
-      return !Throwables.getStackTraceAsString(t).contains(STALE_SHUFFLE_BLOCK_FETCH);
+      return !JavaUtils.stackTraceToString(t).contains(STALE_SHUFFLE_BLOCK_FETCH);
     }
   }
 }

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

@@ -643,4 +643,17 @@ public static String join(List<Object> arr, String sep) {
     }
     return joiner.toString();
   }
+
+  public static String stackTraceToString(Throwable t) {
+    if (t == null) {
+      return "";
+    }
+
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    try (PrintWriter writer = new PrintWriter(out)) {
+      t.printStackTrace(writer);
+      writer.flush();
+    }
+    return out.toString(StandardCharsets.UTF_8);
+  }
 }

common/utils/src/main/scala/org/apache/spark/util/SparkErrorUtils.scala

@@ -16,14 +16,14 @@
  */
 package org.apache.spark.util
 
-import java.io.{Closeable, IOException, PrintWriter}
-import java.nio.charset.StandardCharsets.UTF_8
+import java.io.{Closeable, IOException}
 
 import scala.annotation.tailrec
 import scala.collection.mutable
 import scala.util.control.NonFatal
 
 import org.apache.spark.internal.{Logging, LogKeys}
+import org.apache.spark.network.util.JavaUtils
 
 private[spark] trait SparkErrorUtils extends Logging {
   /**
@@ -103,18 +103,7 @@ private[spark] trait SparkErrorUtils extends Logging {
     }
   }
 
-  def stackTraceToString(t: Throwable): String = {
-    Option(t) match {
-      case None => ""
-      case Some(throwable) =>
-        val out = new java.io.ByteArrayOutputStream
-        SparkErrorUtils.tryWithResource(new PrintWriter(out)) { writer =>
-          throwable.printStackTrace(writer)
-          writer.flush()
-        }
-        new String(out.toByteArray, UTF_8)
-    }
-  }
+  def stackTraceToString(t: Throwable): String = JavaUtils.stackTraceToString(t)
 
   /**
    * Walks the [[Throwable]] to obtain its root cause.

core/src/test/java/test/org/apache/spark/JavaAPISuite.java

@@ -47,7 +47,6 @@
 
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Iterators;
-import com.google.common.base.Throwables;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.IntWritable;
 import org.apache.hadoop.io.Text;
@@ -68,6 +67,7 @@
 import org.apache.spark.api.java.Optional;
 import org.apache.spark.api.java.function.*;
 import org.apache.spark.input.PortableDataStream;
+import org.apache.spark.network.util.JavaUtils;
 import org.apache.spark.partial.BoundedDouble;
 import org.apache.spark.partial.PartialResult;
 import org.apache.spark.rdd.RDD;
@@ -1503,7 +1503,7 @@ public void testAsyncActionErrorWrapping() throws Exception {
     JavaFutureAction<Long> future = rdd.map(new BuggyMapFunction<>()).countAsync();
     ExecutionException ee = assertThrows(ExecutionException.class,
       () -> future.get(2, TimeUnit.SECONDS));
-    assertTrue(Throwables.getStackTraceAsString(ee).contains("Custom exception!"));
+    assertTrue(JavaUtils.stackTraceToString(ee).contains("Custom exception!"));
     assertTrue(future.isDone());
   }
 

dev/checkstyle.xml

@@ -275,6 +275,10 @@
             <property name="format" value="ImmutableSet\.of"/>
             <property name="message" value="Use Set.of instead." />
         </module>
+        <module name="RegexpSinglelineJava">
+            <property name="format" value="Throwables\.getStackTraceAsString"/>
+            <property name="message" value="Use stackTraceToString of JavaUtils/SparkFileUtils/Utils instead." />
+        </module>
         <!-- support structured logging -->
         <module name="RegexpSinglelineJava">
             <property name="format" value="org\.slf4j\.(Logger|LoggerFactory)" />

scalastyle-config.xml

@@ -486,7 +486,7 @@ This file is divided into 3 sections:
 
   <check customId="commonslang3getstacktrace" level="error" class="org.scalastyle.file.RegexChecker" enabled="true">
     <parameters><parameter name="regex">\bExceptionUtils\.getStackTrace\b</parameter></parameters>
-    <customMessage>Use stackTraceToString of SparkErrorUtils or Utils instead</customMessage>
+    <customMessage>Use stackTraceToString of JavaUtils/SparkFileUtils/Utils instead.</customMessage>
   </check>
 
   <check customId="commonslang3strings" level="error" class="org.scalastyle.file.RegexChecker" enabled="true">
@@ -817,6 +817,11 @@ This file is divided into 3 sections:
     <customMessage>Use Java APIs (like java.util.Base64) instead.</customMessage>
   </check>
 
+  <check customId="googleThrowablesGetStackTraceAsString" level="error" class="org.scalastyle.file.RegexChecker" enabled="true">
+    <parameters><parameter name="regex">\bThrowables\.getStackTraceAsString\b</parameter></parameters>
+    <customMessage>Use stackTraceToString of JavaUtils/SparkFileUtils/Utils instead.</customMessage>
+  </check>
+
   <check customId="preconditionschecknotnull" level="error" class="org.scalastyle.file.RegexChecker" enabled="true">
     <parameters><parameter name="regex">\bPreconditions\.checkNotNull\b</parameter></parameters>
     <customMessage>Use requireNonNull of java.util.Objects instead.</customMessage>

streaming/src/main/scala/org/apache/spark/streaming/receiver/ReceiverSupervisorImpl.scala

@@ -24,7 +24,6 @@ import java.util.concurrent.atomic.AtomicLong
 import scala.collection.mutable.ArrayBuffer
 import scala.jdk.CollectionConverters._
 
-import com.google.common.base.Throwables
 import org.apache.hadoop.conf.Configuration
 
 import org.apache.spark.{SparkEnv, SparkException}
@@ -35,7 +34,7 @@ import org.apache.spark.storage.StreamBlockId
 import org.apache.spark.streaming.Time
 import org.apache.spark.streaming.scheduler._
 import org.apache.spark.streaming.util.WriteAheadLogUtils
-import org.apache.spark.util.RpcUtils
+import org.apache.spark.util.{RpcUtils, Utils}
 
 /**
  * Concrete implementation of [[org.apache.spark.streaming.receiver.ReceiverSupervisor]]
@@ -168,7 +167,7 @@ private[streaming] class ReceiverSupervisorImpl(
 
   /** Report error to the receiver tracker */
   def reportError(message: String, error: Throwable): Unit = {
-    val errorString = Option(error).map(Throwables.getStackTraceAsString).getOrElse("")
+    val errorString = Option(error).map(Utils.stackTraceToString).getOrElse("")
     trackerEndpoint.send(ReportError(streamId, message, errorString))
     logWarning(log"Reported error ${MDC(MESSAGE, message)} - ${MDC(ERROR, error)}")
   }
@@ -196,7 +195,7 @@ private[streaming] class ReceiverSupervisorImpl(
 
   override protected def onReceiverStop(message: String, error: Option[Throwable]): Unit = {
     logInfo(log"Deregistering receiver ${MDC(LogKeys.STREAM_ID, streamId)}")
-    val errorString = error.map(Throwables.getStackTraceAsString).getOrElse("")
+    val errorString = error.map(Utils.stackTraceToString).getOrElse("")
     trackerEndpoint.askSync[Boolean](DeregisterReceiver(streamId, message, errorString))
     logInfo(log"Stopped receiver ${MDC(LogKeys.STREAM_ID, streamId)}")
   }