remove query Id truncation

Author: BrooksWalls

core/src/main/scala/org/apache/spark/scheduler/StructuredStreamingIdAwareSchedulerLogging.scala

@@ -150,13 +150,11 @@ private[scheduler] object StructuredStreamingIdAwareSchedulerLogging extends Log
       val queryId = Option(properties.getProperty(QUERY_ID_KEY))
       val batchId = Option(properties.getProperty(BATCH_ID_KEY))
 
-      // formatMessage truncates the queryId for readability
-      // so we use a blank messageWithContext to overwrite the full query Id to the context
       formatMessage(
         queryId,
         batchId,
         entry
-      ) + MessageWithContext("", constructStreamingContext(queryId, batchId))
+      )
     })
   }
 
@@ -200,7 +198,7 @@ private[scheduler] object StructuredStreamingIdAwareSchedulerLogging extends Log
                              batchId: Option[String],
                              msg: => String): String = {
     val msgWithBatchId = batchId.map(bid => s"[batchId = $bid] $msg").getOrElse(msg)
-    queryId.map(qId => s"[queryId = ${qId.take(5)}] $msgWithBatchId").getOrElse(msgWithBatchId)
+    queryId.map(qId => s"[queryId = $qId] $msgWithBatchId").getOrElse(msgWithBatchId)
   }
 
   private def formatMessage(
@@ -211,7 +209,7 @@ private[scheduler] object StructuredStreamingIdAwareSchedulerLogging extends Log
       bId => log"[batchId = ${MDC(LogKeys.BATCH_ID, bId)}] " + toMessageWithContext(msg)
     ).getOrElse(toMessageWithContext(msg))
     queryId.map(
-      qId => log"[queryId = ${MDC(LogKeys.QUERY_ID, qId.take(5))}] " + msgWithBatchId
+      qId => log"[queryId = ${MDC(LogKeys.QUERY_ID, qId)}] " + msgWithBatchId
     ).getOrElse(msgWithBatchId)
   }
 

core/src/test/scala/org/apache/spark/scheduler/PoolSuite.scala

@@ -395,7 +395,7 @@ class PoolSuite extends SparkFunSuite with LocalSparkContext {
     }
 
     val logs = logAppender.loggingEvents.map(_.getMessage.getFormattedMessage)
-    val expectedQueryPrefix = s"[queryId = ${testQueryId.take(5)}]"
+    val expectedQueryPrefix = s"[queryId = ${testQueryId}]"
     val expectedBatchPrefix = s"[batchId = $testBatchId]"
     val addedLogs = logs.filter(msg =>
       msg.contains("Added task set") &&

core/src/test/scala/org/apache/spark/scheduler/StructuredStreamingIdAwareSchedulerLoggingSuite.scala

@@ -68,7 +68,7 @@ class StructuredStreamingIdAwareSchedulerLoggingSuite extends SparkFunSuite {
     val result = StructuredStreamingIdAwareSchedulerLogging
       .constructStreamingLogEntry(propsWithBothIds(), "test message")
 
-    assertResult(s"[queryId = ${testQueryId.take(5)}] [batchId = $testBatchId] test message")(
+    assertResult(s"[queryId = $testQueryId] [batchId = $testBatchId] test message")(
       result.message)
     assertContextValue(result.context, LogKeys.QUERY_ID, testQueryId)
     assertContextValue(result.context, LogKeys.BATCH_ID, testBatchId)
@@ -78,7 +78,7 @@ class StructuredStreamingIdAwareSchedulerLoggingSuite extends SparkFunSuite {
     val result = StructuredStreamingIdAwareSchedulerLogging
       .constructStreamingLogEntry(propsWithQueryIdOnly(), "test message")
 
-    assertResult(s"[queryId = ${testQueryId.take(5)}] test message")(result.message)
+    assertResult(s"[queryId = $testQueryId] test message")(result.message)
     assertContextValue(result.context, LogKeys.QUERY_ID, testQueryId)
     assertContextAbsent(result.context, LogKeys.BATCH_ID)
   }
@@ -104,7 +104,7 @@ class StructuredStreamingIdAwareSchedulerLoggingSuite extends SparkFunSuite {
       .constructStreamingLogEntry(propsWithBothIds(),
         log"test message ${MDC(LogKeys.MESSAGE, "Dummy Context")}")
 
-    assertResult(s"[queryId = ${testQueryId.take(5)}] " +
+    assertResult(s"[queryId = $testQueryId] " +
       s"[batchId = $testBatchId] test message Dummy Context")(
       result.message)
     assertContextValue(result.context, LogKeys.QUERY_ID, testQueryId)
@@ -117,7 +117,7 @@ class StructuredStreamingIdAwareSchedulerLoggingSuite extends SparkFunSuite {
       .constructStreamingLogEntry(propsWithQueryIdOnly(),
         log"test message ${MDC(LogKeys.MESSAGE, "Dummy Context")}")
 
-    assertResult(s"[queryId = ${testQueryId.take(5)}] test message Dummy Context")(result.message)
+    assertResult(s"[queryId = $testQueryId] test message Dummy Context")(result.message)
     assertContextValue(result.context, LogKeys.QUERY_ID, testQueryId)
     assertContextAbsent(result.context, LogKeys.BATCH_ID)
     assertContextValue(result.context, LogKeys.MESSAGE, "Dummy Context")

core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala

@@ -2903,7 +2903,7 @@ class TaskSetManagerSuite
 
     val logs = logAppender.loggingEvents.map(_.getMessage.getFormattedMessage)
 
-    val expectedQueryPrefix = s"[queryId = ${testQueryId.take(5)}]"
+    val expectedQueryPrefix = s"[queryId = ${testQueryId}]"
     val expectedBatchPrefix = s"[batchId = $testBatchId]"
 
     // Verify the "Starting" log line includes query Id and batch Id