[KYUUBI #1798] Add EventBus module to unify the distribution and subscription of Kyuubi's events

### _Why are the changes needed?_

Use case is simple:
```
    // scribe sync
    EventBus.register[Test0KyuubiEvent](new Test0Handler with EventHandler)

    // scribe async
     EventBus.registerAsync[Test0KyuubiEvent](new Test0Handler with EventHandler)

    // send event
    EventBus.post(KyuubiEvent("test0"))
    EventBus.post(KyuubiEvent("test1"))
```

### _How was this patch tested?_
- [x] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [x] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #1853 from iodone/dev.

Closes #1798

7c01700 [odone] Updated
d2ecb42 [odone] [Fixed] address some reviews
09a467f [odone] [Removed] Remove some mistakenly changed content
5136fe4 [odone] [Update] JsonLoggerEventHandler instead of LoggingService
29b4f8a [odone] [Update] JsonLoggerEventHandler instead of LoggingService
a2f90bb [odone] [KYUUBI #1798] Added: Add async event handler execution
21be135 [odone] Added: more test case
10906cf [odone] Added: eventHandler can scribe subclass event
2c09b75 [odone] Added: Event Bus
bb05f77 [odone] Added: Event Bus

Authored-by: odone <odone.zhang@gmail.com>
Signed-off-by: ulysses-you <ulyssesyou@apache.org>

Author: iodone

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/SparkSQLEngine.scala

@@ -34,8 +34,9 @@ import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf._
 import org.apache.kyuubi.config.KyuubiReservedKeys.KYUUBI_ENGINE_SUBMIT_TIME_KEY
 import org.apache.kyuubi.engine.spark.SparkSQLEngine.{countDownLatch, currentEngine}
-import org.apache.kyuubi.engine.spark.events.{EngineEvent, EngineEventsStore, SparkEventLoggingService}
-import org.apache.kyuubi.events.EventLogging
+import org.apache.kyuubi.engine.spark.events.{EngineEvent, EngineEventsStore}
+import org.apache.kyuubi.engine.spark.events.handler.{SparkHistoryLoggingEventHandler, SparkJsonLoggingEventHandler}
+import org.apache.kyuubi.events.{EventBus, EventLoggerType, KyuubiEvent}
 import org.apache.kyuubi.ha.HighAvailabilityConf._
 import org.apache.kyuubi.ha.client.RetryPolicies
 import org.apache.kyuubi.service.Serverable
@@ -144,20 +145,18 @@ object SparkSQLEngine extends Logging {
   def startEngine(spark: SparkSession): Unit = {
     currentEngine = Some(new SparkSQLEngine(spark))
     currentEngine.foreach { engine =>
-      // start event logging ahead so that we can capture all statuses
-      val eventLogging = new SparkEventLoggingService(spark.sparkContext)
       try {
-        eventLogging.initialize(kyuubiConf)
-        eventLogging.start()
+        // start event logging ahead so that we can capture all statuses
+        initLoggerEventHandler(kyuubiConf)
       } catch {
         case NonFatal(e) =>
-          // Don't block the main process if the `EventLoggingService` failed to start
-          warn(s"Failed to initialize EventLoggingService: ${e.getMessage}", e)
+          // Don't block the main process if the `LoggerEventHandler` failed to start
+          warn(s"Failed to initialize LoggerEventHandler: ${e.getMessage}", e)
       }
 
       try {
         engine.initialize(kyuubiConf)
-        EventLogging.onEvent(EngineEvent(engine))
+        EventBus.post(EngineEvent(engine))
       } catch {
         case t: Throwable =>
           throw new KyuubiException(s"Failed to initialize SparkSQLEngine: ${t.getMessage}", t)
@@ -173,15 +172,37 @@ object SparkSQLEngine extends Logging {
           kyuubiConf)
         val event = EngineEvent(engine)
         info(event)
-        EventLogging.onEvent(event)
+        EventBus.post(event)
       } catch {
         case t: Throwable =>
           throw new KyuubiException(s"Failed to start SparkSQLEngine: ${t.getMessage}", t)
       }
       // Stop engine before SparkContext stopped to avoid calling a stopped SparkContext
       addShutdownHook(() => engine.stop(), SPARK_CONTEXT_SHUTDOWN_PRIORITY + 2)
-      addShutdownHook(() => eventLogging.stop(), SPARK_CONTEXT_SHUTDOWN_PRIORITY + 1)
     }
+
+    def initLoggerEventHandler(conf: KyuubiConf): Unit = {
+      conf.get(ENGINE_EVENT_LOGGERS)
+        .map(EventLoggerType.withName)
+        .foreach {
+          case EventLoggerType.SPARK =>
+            EventBus.register[KyuubiEvent](new SparkHistoryLoggingEventHandler(spark.sparkContext))
+          case EventLoggerType.JSON =>
+            val handler = SparkJsonLoggingEventHandler(
+              spark.sparkContext.applicationAttemptId
+                .map(id => s"${spark.sparkContext.applicationId}_$id")
+                .getOrElse(spark.sparkContext.applicationId),
+              ENGINE_EVENT_JSON_LOG_PATH,
+              spark.sparkContext.hadoopConfiguration,
+              conf)
+
+            // register JsonLogger as a event handler for default event bus
+            EventBus.register[KyuubiEvent](handler)
+          case _ =>
+        }
+
+    }
+
   }
 
   def main(args: Array[String]): Unit = {
@@ -211,7 +232,7 @@ object SparkSQLEngine extends Logging {
               case Some(engine) =>
                 engine.stop()
                 val event = EngineEvent(engine).copy(diagnostic = e.getMessage)
-                EventLogging.onEvent(event)
+                EventBus.post(event)
                 error(event, e)
               case _ => error("Current SparkSQLEngine is not created.")
             }

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/events/SparkEventLoggingService.scala

@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.engine.spark.events.handler
+
+import org.apache.spark.SparkContext
+import org.apache.spark.kyuubi.SparkContextHelper
+
+import org.apache.kyuubi.events.KyuubiEvent
+import org.apache.kyuubi.events.handler.EventHandler
+
+class SparkHistoryLoggingEventHandler(sc: SparkContext) extends EventHandler[KyuubiEvent] {
+  override def apply(event: KyuubiEvent): Unit = {
+    SparkContextHelper.postEventToSparkListenerBus(event, sc)
+  }
+}

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/events/handler/SparkHistoryLoggingEventHandler.scala

@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.engine.spark.events.handler
+
+import org.apache.hadoop.conf.Configuration
+
+import org.apache.kyuubi.config.{ConfigEntry, KyuubiConf}
+import org.apache.kyuubi.events.handler.JsonLoggingEventHandler
+
+case class SparkJsonLoggingEventHandler(
+    logName: String,
+    logPath: ConfigEntry[String],
+    hadoopConf: Configuration,
+    kyuubiConf: KyuubiConf)
+  extends JsonLoggingEventHandler(logName, logPath, hadoopConf, kyuubiConf)

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/events/handler/SparkJsonLoggingEventHandler.scala

@@ -29,7 +29,7 @@ import org.apache.kyuubi.{KyuubiSQLException, Logging}
 import org.apache.kyuubi.config.KyuubiConf.OPERATION_RESULT_MAX_ROWS
 import org.apache.kyuubi.engine.spark.KyuubiSparkUtil._
 import org.apache.kyuubi.engine.spark.events.SparkOperationEvent
-import org.apache.kyuubi.events.EventLogging
+import org.apache.kyuubi.events.EventBus
 import org.apache.kyuubi.operation.{ArrayFetchIterator, IterableFetchIterator, OperationState, OperationType}
 import org.apache.kyuubi.operation.OperationState.OperationState
 import org.apache.kyuubi.operation.log.OperationLog
@@ -51,7 +51,7 @@ class ExecuteStatement(
 
   private val operationListener: SQLOperationListener = new SQLOperationListener(this, spark)
 
-  EventLogging.onEvent(SparkOperationEvent(this))
+  EventBus.post(SparkOperationEvent(this))
 
   override protected def resultSchema: StructType = {
     if (result == null || result.schema.isEmpty) {
@@ -155,7 +155,7 @@ class ExecuteStatement(
 
   override def setState(newState: OperationState): Unit = {
     super.setState(newState)
-    EventLogging.onEvent(
+    EventBus.post(
       SparkOperationEvent(this, operationListener.getExecutionId))
   }
 }

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecuteStatement.scala

@@ -23,7 +23,7 @@ import org.apache.spark.sql.{AnalysisException, SparkSession}
 import org.apache.kyuubi.engine.spark.events.SessionEvent
 import org.apache.kyuubi.engine.spark.operation.SparkSQLOperationManager
 import org.apache.kyuubi.engine.spark.udf.KDFRegistry
-import org.apache.kyuubi.events.EventLogging
+import org.apache.kyuubi.events.EventBus
 import org.apache.kyuubi.operation.{Operation, OperationHandle}
 import org.apache.kyuubi.session.{AbstractSession, SessionManager}
 
@@ -56,7 +56,7 @@ class SparkSessionImpl(
       case (key, value) => setModifiableConfig(key, value)
     }
     KDFRegistry.registerAll(spark)
-    EventLogging.onEvent(sessionEvent)
+    EventBus.post(sessionEvent)
     super.open()
   }
 
@@ -67,7 +67,7 @@ class SparkSessionImpl(
 
   override def close(): Unit = {
     sessionEvent.endTime = System.currentTimeMillis()
-    EventLogging.onEvent(sessionEvent)
+    EventBus.post(sessionEvent)
     super.close()
     spark.sessionState.catalog.getTempViewNames().foreach(spark.catalog.uncacheTable(_))
     sessionManager.operationManager.asInstanceOf[SparkSQLOperationManager].closeILoop(handle)

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/session/SparkSessionImpl.scala

@@ -29,16 +29,19 @@ import org.apache.spark.ui.SparkUI
 
 import org.apache.kyuubi.Logging
 import org.apache.kyuubi.config.KyuubiReservedKeys.KYUUBI_STATEMENT_ID_KEY
-import org.apache.kyuubi.events.{EventLogger, KyuubiEvent}
+import org.apache.kyuubi.events.KyuubiEvent
 
 /**
  * A place to invoke non-public APIs of [[SparkContext]], anything to be added here need to
  * think twice
  */
 object SparkContextHelper extends Logging {
 
-  def createSparkHistoryLogger(sc: SparkContext): EventLogger = {
-    new SparkHistoryEventLogger(sc)
+  def postEventToSparkListenerBus(event: KyuubiEvent, sc: SparkContext) {
+    event match {
+      case s: SparkListenerEvent => sc.listenerBus.post(s)
+      case _ => // ignore
+    }
   }
 
   def getKvStore(sc: SparkContext): ElementTrackingStore = {
@@ -81,16 +84,3 @@ object SparkContextHelper extends Logging {
   }
 
 }
-
-/**
- * A [[EventLogger]] that logs everything to SparkHistory
- * @param sc SparkContext
- */
-private class SparkHistoryEventLogger(sc: SparkContext) extends EventLogger {
-  override def logEvent(kyuubiEvent: KyuubiEvent): Unit = {
-    kyuubiEvent match {
-      case s: SparkListenerEvent => sc.listenerBus.post(s)
-      case _ => // ignore
-    }
-  }
-}

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/kyuubi/SparkContextHelper.scala

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.kyuubi.engine.spark.events
+package org.apache.kyuubi.engine.spark.events.handler
 
 import java.io.{BufferedReader, InputStreamReader}
 import java.nio.file.Paths
@@ -28,17 +28,18 @@ import org.scalatest.time.SpanSugar._
 import org.apache.kyuubi.Utils
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.engine.spark.{KyuubiSparkUtil, WithSparkSQLEngine}
+import org.apache.kyuubi.engine.spark.events.{EngineEvent, SessionEvent}
 import org.apache.kyuubi.events.EventLoggerType._
 import org.apache.kyuubi.events.JsonProtocol
 import org.apache.kyuubi.operation.{HiveJDBCTestHelper, OperationHandle}
 
-class SparkEventLoggingServiceSuite extends WithSparkSQLEngine with HiveJDBCTestHelper {
+class SparkJsonLoggingEventHandlerSuite extends WithSparkSQLEngine with HiveJDBCTestHelper {
 
   private val logRoot = "file://" + Utils.createTempDir().toString
   private val currentDate = Utils.getDateFromTimestamp(System.currentTimeMillis())
 
   override def withKyuubiConf: Map[String, String] = Map(
-    KyuubiConf.ENGINE_EVENT_LOGGERS.key -> s"$JSON,$SPARK",
+    KyuubiConf.ENGINE_EVENT_LOGGERS.key -> s"$JSON",
     KyuubiConf.ENGINE_EVENT_JSON_LOG_PATH.key -> logRoot,
     "spark.eventLog.enabled" -> "true",
     "spark.eventLog.dir" -> logRoot)
@@ -56,15 +57,13 @@ class SparkEventLoggingServiceSuite extends WithSparkSQLEngine with HiveJDBCTest
       "session",
       s"day=$currentDate",
       KyuubiSparkUtil.engineId + ".json")
-
     val fileSystem: FileSystem = FileSystem.get(new Configuration())
     val fs: FSDataInputStream = fileSystem.open(new Path(engineEventPath.toString))
     val engineEventReader = new BufferedReader(new InputStreamReader(fs))
 
     val readEvent =
       JsonProtocol.jsonToEvent(engineEventReader.readLine(), classOf[EngineEvent])
     assert(readEvent.isInstanceOf[EngineEvent])
-
     withJdbcStatement() { statement =>
       val table = engineEventPath.getParent
       val resultSet = statement.executeQuery(s"SELECT * FROM `json`.`$table`")