[SPARK-50338][CORE] Make LazyTry exceptions less verbose

core/src/main/scala/org/apache/spark/util/Utils.scala

@@ -1354,8 +1354,8 @@ private[spark] object Utils
   val TRY_WITH_CALLER_STACKTRACE_FULL_STACKTRACE =
     "Full stacktrace of original doTryWithCallerStacktrace caller"
 
-  val TRY_WITH_CALLER_STACKTRACE_TRY_STACKTRACE =
-    "Stacktrace under doTryWithCallerStacktrace"
+  class OriginalTryStackTraceException(val doTryWithCallerStacktraceDepth: Int)
+    extends Exception(TRY_WITH_CALLER_STACKTRACE_FULL_STACKTRACE)
 
   /**
    * Use Try with stacktrace substitution for the caller retrieving the error.
@@ -1383,12 +1383,7 @@ private[spark] object Utils
         val commonSuffixLen = origStackTrace.reverse.zip(currentStackTrace.reverse).takeWhile {
           case (exElem, currentElem) => exElem == currentElem
         }.length
-        val belowEx = new Exception(TRY_WITH_CALLER_STACKTRACE_TRY_STACKTRACE)
-        belowEx.setStackTrace(origStackTrace.dropRight(commonSuffixLen))
-        ex.addSuppressed(belowEx)
-
-        // keep the full original stack trace in a suppressed exception.
-        val fullEx = new Exception(TRY_WITH_CALLER_STACKTRACE_FULL_STACKTRACE)
+        val fullEx = new OriginalTryStackTraceException(origStackTrace.size - commonSuffixLen)
         fullEx.setStackTrace(origStackTrace)
         ex.addSuppressed(fullEx)
       case Success(_) => // nothing
@@ -1406,7 +1401,7 @@ private[spark] object Utils
    * Full stack trace of the original doTryWithCallerStacktrace caller can be retrieved with
    * ```
    * ex.getSuppressed.find { e =>
-   * e.getMessage == Utils.TRY_WITH_CALLER_STACKTRACE_FULL_STACKTRACE
+   *   e.isInstanceOff[Utils.OriginalTryStackTraceException]
    * }
    * ```
    *
@@ -1416,13 +1411,15 @@ private[spark] object Utils
    */
   def getTryWithCallerStacktrace[T](t: Try[T]): T = t match {
     case Failure(ex) =>
-      val belowStacktrace = ex.getSuppressed.find { e =>
+      val originalStacktraceEx = ex.getSuppressed.find { e =>
         // added in doTryWithCallerStacktrace
-        e.getMessage == TRY_WITH_CALLER_STACKTRACE_TRY_STACKTRACE
+        e.isInstanceOf[OriginalTryStackTraceException]
       }.getOrElse {
         // If we don't have the expected stacktrace information, just rethrow
         throw ex
-      }.getStackTrace
+      }.asInstanceOf[OriginalTryStackTraceException]
+      val belowStacktrace = originalStacktraceEx.getStackTrace
+        .take(originalStacktraceEx.doTryWithCallerStacktraceDepth)
       // We are modifying and throwing the original exception. It would be better if we could
       // return a copy, but we can't easily clone it and preserve. If this is accessed from
       // multiple threads that then look at the stack trace, this could break.

core/src/test/scala/org/apache/spark/util/UtilsSuite.scala

@@ -1581,26 +1581,14 @@ class UtilsSuite extends SparkFunSuite with ResetSystemProperties {
     //   at org.apache.spark.util.UtilsSuite.$anonfun$new$165(UtilsSuite.scala:1658)
     //   ... 56 more
     // scalastyle:on line.size.limit
-    val origSt = e1.getSuppressed.find(
-      _.getMessage == Utils.TRY_WITH_CALLER_STACKTRACE_FULL_STACKTRACE)
+    val origSt = e1.getSuppressed.find(_.isInstanceOf[Utils.OriginalTryStackTraceException])
     assert(origSt.isDefined)
     assert(origSt.get.getStackTrace.exists(_.getMethodName == "throwException"))
     assert(origSt.get.getStackTrace.exists(_.getMethodName == "callDoTry"))
 
-    // The stack trace under Try should be in the suppressed exceptions.
-    // Example:
-    // Suppressed: java.lang.Exception: Stacktrace under doTryWithCallerStacktrace
-    //   at org.apache.spark.util.UtilsSuite.throwException(UtilsSuite.scala: 1640)
-    //   at org.apache.spark.util.UtilsSuite.$anonfun$callDoTry$1(UtilsSuite.scala: 1645)
-    //   at scala.util.Try$.apply(Try.scala: 213)
-    //   at org.apache.spark.util.Utils$.doTryWithCallerStacktrace(Utils.scala: 1586)
-    val trySt = e1.getSuppressed.find(
-      _.getMessage == Utils.TRY_WITH_CALLER_STACKTRACE_TRY_STACKTRACE)
-    assert(trySt.isDefined)
-    // calls under callDoTry should be present.
-    assert(trySt.get.getStackTrace.exists(_.getMethodName == "throwException"))
-    // callDoTry should be removed.
-    assert(!trySt.get.getStackTrace.exists(_.getMethodName == "callDoTry"))
+    // Should save the depth of the stack trace under doTryWithCallerStacktrace.
+    assert(origSt.get.asInstanceOf[Utils.OriginalTryStackTraceException]
+      .doTryWithCallerStacktraceDepth == 4)
 
     val e2 = intercept[Exception] {
       callGetTryAgain(t)