[HADP-56069] Support writing spark-monitor events to EventLog file in Hermes (apache#654)

Author: fenzhu

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SessionAwareEventLoggingListener.scala

@@ -393,6 +393,31 @@ private[spark] class SessionAwareEventLoggingListener(
           sessionId)
       case s: SparkListenerQueryPlanSignature =>
         logEvent(event, flushLogger = true, s.sessionId)
+      case s: SparkListenerEvent if s.getClass.getSimpleName.startsWith("SparkMonitor") =>
+        try {
+          val sessionIdField = s.getClass.getDeclaredField("sessionId")
+          sessionIdField.setAccessible(true)
+          val sessionId = sessionIdField.get(s).asInstanceOf[String]
+          logInfo(s.getClass.getCanonicalName + ": sessionId = " + sessionId)
+          if (sessionId != null && sessionId.nonEmpty) {
+            logEvent(event, flushLogger = true, sessionId)
+          } else {
+            val executionIdField = s.getClass.getDeclaredField("executionId")
+            executionIdField.setAccessible(true)
+            val executionId = executionIdField.get(s).asInstanceOf[Long]
+            if (executionIdToSessionId.containsKey(String.valueOf(executionId))) {
+              val sessionId = executionIdToSessionId.get(String.valueOf(executionId))
+              logInfo(s"Flush ${s.getClass.getCanonicalName}: executionId = " +
+                s"$executionId, sessionId = $sessionId")
+              logEvent(event, flushLogger = true, sessionId)
+            } else {
+              logError(s"Execution $executionId in ${s.getClass.getCanonicalName} " +
+                s"doesn't belong to any session")
+            }
+          }
+        } catch {
+          case e: Exception => logError(s"Fail to persist ${s.getClass.getCanonicalName}: $e")
+        }
       case _ =>
     }
   }