[KYUUBI #2028][FOLLOWUP] add engine stop event and fix the partition of initialized event

### _Why are the changes needed?_

add engine stop event;
fix initialized event partition;

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

- [ ] Add screenshots for manual tests if appropriate

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

Closes #2421 from zhaomin1423/fix_hive_engine.

Closes #2028

a967631 [Min Zhao] add engineStarTime
e7815bf [Min Zhao] add engineStarTime
953a9d8 [Min Zhao] post stop event in stopServer
0ec6911 [Min Zhao] fix conflict
4bf4159 [Min Zhao] fix conflict

Authored-by: Min Zhao <zhaomin1423@163.com>
Signed-off-by: ulysses-you <ulyssesyou@apache.org>

Author: zhaomin1423

externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/HiveSQLEngine.scala

@@ -26,18 +26,20 @@ import org.apache.hadoop.hive.conf.HiveConf
 import org.apache.kyuubi.{Logging, Utils}
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf.{ENGINE_EVENT_JSON_LOG_PATH, ENGINE_EVENT_LOGGERS}
+import org.apache.kyuubi.engine.hive.HiveSQLEngine.currentEngine
 import org.apache.kyuubi.engine.hive.events.HiveEngineEvent
 import org.apache.kyuubi.engine.hive.events.handler.HiveJsonLoggingEventHandler
 import org.apache.kyuubi.events.{EventBus, EventLoggerType, KyuubiEvent}
 import org.apache.kyuubi.ha.HighAvailabilityConf.HA_ZK_CONN_RETRY_POLICY
 import org.apache.kyuubi.ha.client.RetryPolicies
-import org.apache.kyuubi.service.{AbstractBackendService, AbstractFrontendService, Serverable}
+import org.apache.kyuubi.service.{AbstractBackendService, AbstractFrontendService, Serverable, ServiceState}
 import org.apache.kyuubi.util.{KyuubiHadoopUtils, SignalRegister}
 
 class HiveSQLEngine extends Serverable("HiveSQLEngine") {
   override val backendService: AbstractBackendService = new HiveBackendService(this)
   override val frontendServices: Seq[AbstractFrontendService] =
     Seq(new HiveTBinaryFrontendService(this))
+  private[hive] val engineStartTime = System.currentTimeMillis()
 
   override def start(): Unit = {
     super.start()
@@ -47,6 +49,12 @@ class HiveSQLEngine extends Serverable("HiveSQLEngine") {
   }
 
   override protected def stopServer(): Unit = {
+    currentEngine.foreach { engine =>
+      val event = HiveEngineEvent(engine)
+        .copy(state = ServiceState.STOPPED, endTime = System.currentTimeMillis())
+      EventBus.post(event)
+    }
+
     // #2351
     // https://issues.apache.org/jira/browse/HIVE-23164
     // Server is not properly terminated because of non-daemon threads
@@ -131,7 +139,8 @@ object HiveSQLEngine extends Logging {
       case t: Throwable => currentEngine match {
           case Some(engine) =>
             engine.stop()
-            val event = HiveEngineEvent(engine).copy(diagnostic = t.getMessage)
+            val event = HiveEngineEvent(engine)
+              .copy(endTime = System.currentTimeMillis(), diagnostic = t.getMessage)
             EventBus.post(event)
           case _ =>
             error(s"Failed to start Hive SQL engine: ${t.getMessage}.", t)

externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/events/HiveEngineEvent.scala

@@ -30,8 +30,9 @@ case class HiveEngineEvent(
     diagnostic: String,
     settings: Map[String, String]) extends KyuubiEvent {
 
-  override def partitions: Seq[(String, String)] =
+  override def partitions: Seq[(String, String)] = {
     ("day", Utils.getDateFromTimestamp(startTime)) :: Nil
+  }
 
   override def toString: String = {
     s"""
@@ -50,7 +51,6 @@ case class HiveEngineEvent(
 object HiveEngineEvent {
 
   def apply(engine: HiveSQLEngine): HiveEngineEvent = {
-    engine.getStartTime
     val connectionUrl =
       if (engine.getServiceState.equals(ServiceState.LATENT)) {
         null
@@ -60,7 +60,7 @@ object HiveEngineEvent {
 
     new HiveEngineEvent(
       connectionUrl = connectionUrl,
-      startTime = engine.getStartTime,
+      startTime = engine.engineStartTime,
       endTime = -1L,
       state = engine.getServiceState,
       diagnostic = "",

externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/events/SessionEvent.scala renamed to externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/events/HiveSessionEvent.scala

@@ -30,7 +30,7 @@ import org.apache.kyuubi.events.KyuubiEvent
  * @param serverIp Kyuubi Server IP address
  * @param totalOperations how many queries and meta calls
  */
-case class SessionEvent(
+case class HiveSessionEvent(
     sessionId: String,
     engineId: String,
     username: String,
@@ -45,10 +45,10 @@ case class SessionEvent(
 
 }
 
-object SessionEvent {
+object HiveSessionEvent {
 
-  def apply(session: HiveSessionImpl): SessionEvent = {
-    new SessionEvent(
+  def apply(session: HiveSessionImpl): HiveSessionEvent = {
+    new HiveSessionEvent(
       session.handle.identifier.toString,
       engineId = "",
       session.user,

externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/session/HiveSessionImpl.scala

@@ -24,7 +24,7 @@ import scala.collection.JavaConverters._
 import org.apache.hive.service.cli.session.HiveSession
 import org.apache.hive.service.rpc.thrift.TProtocolVersion
 
-import org.apache.kyuubi.engine.hive.events.SessionEvent
+import org.apache.kyuubi.engine.hive.events.HiveSessionEvent
 import org.apache.kyuubi.events.EventBus
 import org.apache.kyuubi.operation.{Operation, OperationHandle}
 import org.apache.kyuubi.session.{AbstractSession, SessionHandle, SessionManager}
@@ -41,7 +41,7 @@ class HiveSessionImpl(
     val hive: HiveSession)
   extends AbstractSession(protocol, user, password, ipAddress, conf, sessionManager) {
 
-  private val sessionEvent = SessionEvent(this)
+  private val sessionEvent = HiveSessionEvent(this)
 
   def serverIpAddress(): String = serverIpAddress
 

externals/kyuubi-hive-sql-engine/src/test/scala/org/apache/kyuubi/engine/hive/event/HiveEventLoggingServiceSuite.scala

@@ -28,9 +28,10 @@ import org.apache.kyuubi.Utils
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf.ENGINE_EVENT_JSON_LOG_PATH
 import org.apache.kyuubi.engine.hive.HiveSQLEngine
-import org.apache.kyuubi.engine.hive.events.{HiveEngineEvent, HiveOperationEvent, SessionEvent}
+import org.apache.kyuubi.engine.hive.events.{HiveEngineEvent, HiveOperationEvent, HiveSessionEvent}
 import org.apache.kyuubi.events.JsonProtocol
 import org.apache.kyuubi.operation.HiveJDBCTestHelper
+import org.apache.kyuubi.service.ServiceState
 
 class HiveEventLoggingServiceSuite extends HiveJDBCTestHelper {
 
@@ -69,9 +70,13 @@ class HiveEventLoggingServiceSuite extends HiveJDBCTestHelper {
     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 =
+    val initializedEvent =
       JsonProtocol.jsonToEvent(engineEventReader.readLine(), classOf[HiveEngineEvent])
-    assert(readEvent.isInstanceOf[HiveEngineEvent])
+    assert(initializedEvent.asInstanceOf[HiveEngineEvent].state.equals(ServiceState.INITIALIZED))
+
+    val startedEvent =
+      JsonProtocol.jsonToEvent(engineEventReader.readLine(), classOf[HiveEngineEvent])
+    assert(startedEvent.asInstanceOf[HiveEngineEvent].state.equals(ServiceState.STARTED))
   }
 
   test("test session event logging") {
@@ -82,15 +87,15 @@ class HiveEventLoggingServiceSuite extends HiveJDBCTestHelper {
       assert(!catalogs.next())
       val sessionEventPath = Paths.get(
         logRoot,
-        "session",
+        "hive_session",
         s"day=$currentDate",
         s"Hive-$hostName.json")
       val fileSystem: FileSystem = FileSystem.get(new Configuration())
       val fs: FSDataInputStream = fileSystem.open(new Path(sessionEventPath.toString))
       val engineEventReader = new BufferedReader(new InputStreamReader(fs))
       val readEvent =
-        JsonProtocol.jsonToEvent(engineEventReader.readLine(), classOf[SessionEvent])
-      assert(readEvent.isInstanceOf[SessionEvent])
+        JsonProtocol.jsonToEvent(engineEventReader.readLine(), classOf[HiveSessionEvent])
+      assert(readEvent.isInstanceOf[HiveSessionEvent])
     }
   }