[KYUUBI #4119] Return app submission time for batch

### _Why are the changes needed?_

Return the batch app submission time, it is useful to provide the app elapse time.

### _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 #4119 from turboFei/open_time.

Closes #4119

9c7ddef [fwang12] rebase
1c3c7e4 [fwang12] save
bf5177a [fwang12] ut
ae1c0be [fwang12] get from recovery
48b3a1e [fwang12] refactor
f8778e6 [fwang12] engine open time
1989aa4 [fwang12] add
8884ee5 [fwang12] save

Authored-by: fwang12 <fwang12@ebay.com>
Signed-off-by: fwang12 <fwang12@ebay.com>

kyuubi-ctl/src/main/scala/org/apache/kyuubi/ctl/util/Render.scala

@@ -111,6 +111,10 @@ private[ctl] object Render {
 
   private def buildBatchAppInfo(batch: Batch, showDiagnostic: Boolean = true): List[String] = {
     val batchAppInfo = ListBuffer[String]()
+    if (batch.getAppSubmissionTime > 0) {
+      batchAppInfo += s"App Submission Time:" +
+        s" ${millisToDateString(batch.getAppSubmissionTime, "yyyy-MM-dd HH:mm:ss")}"
+    }
     Option(batch.getAppId).foreach { _ =>
       batchAppInfo += s"App Id: ${batch.getAppId}"
     }

kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/api/v1/dto/Batch.java

@@ -26,6 +26,7 @@ public class Batch {
   private String user;
   private String batchType;
   private String name;
+  private long appSubmissionTime;
   private String appId;
   private String appUrl;
   private String appState;
@@ -42,6 +43,7 @@ public Batch(
       String user,
       String batchType,
       String name,
+      long appSubmissionTime,
       String appId,
       String appUrl,
       String appState,
@@ -54,6 +56,7 @@ public Batch(
     this.user = user;
     this.batchType = batchType;
     this.name = name;
+    this.appSubmissionTime = appSubmissionTime;
     this.appId = appId;
     this.appUrl = appUrl;
     this.appState = appState;
@@ -152,6 +155,14 @@ public void setCreateTime(long createTime) {
     this.createTime = createTime;
   }
 
+  public long getAppSubmissionTime() {
+    return appSubmissionTime;
+  }
+
+  public void setAppSubmissionTime(long appSubmissionTime) {
+    this.appSubmissionTime = appSubmissionTime;
+  }
+
   public long getEndTime() {
     return endTime;
   }

kyuubi-rest-client/src/test/java/org/apache/kyuubi/client/BatchRestClientTest.java

@@ -117,6 +117,7 @@ public void testNoPasswordBasicClient() {
     assertEquals(result.getUser(), expectedBatch.getUser());
     assertEquals(result.getBatchType(), expectedBatch.getBatchType());
     assertEquals(result.getName(), expectedBatch.getName());
+    assertEquals(result.getAppSubmissionTime(), expectedBatch.getAppSubmissionTime());
     assertEquals(result.getAppId(), expectedBatch.getAppId());
     assertEquals(result.getAppUrl(), expectedBatch.getAppUrl());
     assertEquals(result.getAppState(), expectedBatch.getAppState());
@@ -145,6 +146,7 @@ public void testAnonymousBasicClient() {
     assertEquals(result.getUser(), expectedBatch.getUser());
     assertEquals(result.getBatchType(), expectedBatch.getBatchType());
     assertEquals(result.getName(), expectedBatch.getName());
+    assertEquals(result.getAppSubmissionTime(), expectedBatch.getAppSubmissionTime());
     assertEquals(result.getAppId(), expectedBatch.getAppId());
     assertEquals(result.getAppUrl(), expectedBatch.getAppUrl());
     assertEquals(result.getAppState(), expectedBatch.getAppState());
@@ -176,6 +178,7 @@ public void createBatchTest() {
     assertEquals(result.getUser(), expectedBatch.getUser());
     assertEquals(result.getBatchType(), expectedBatch.getBatchType());
     assertEquals(result.getName(), expectedBatch.getName());
+    assertEquals(result.getAppSubmissionTime(), expectedBatch.getAppSubmissionTime());
     assertEquals(result.getAppId(), expectedBatch.getAppId());
     assertEquals(result.getAppUrl(), expectedBatch.getAppUrl());
     assertEquals(result.getAppState(), expectedBatch.getAppState());
@@ -207,6 +210,7 @@ public void getBatchByIdTest() {
     assertEquals(result.getUser(), expectedBatch.getUser());
     assertEquals(result.getBatchType(), expectedBatch.getBatchType());
     assertEquals(result.getName(), expectedBatch.getName());
+    assertEquals(result.getAppSubmissionTime(), expectedBatch.getAppSubmissionTime());
     assertEquals(result.getAppId(), expectedBatch.getAppId());
     assertEquals(result.getAppUrl(), expectedBatch.getAppUrl());
     assertEquals(result.getAppState(), expectedBatch.getAppState());

kyuubi-rest-client/src/test/java/org/apache/kyuubi/client/RestClientTestUtils.java

@@ -51,6 +51,7 @@ public static Batch generateTestBatch(String id) {
             TEST_USERNAME,
             "spark",
             "batch_name",
+            0,
             id,
             null,
             "RUNNING",

kyuubi-server/src/main/resources/sql/derby/002-KYUUBI-4119.derby.sql

@@ -0,0 +1 @@
+ALTER TABLE metadata ADD COLUMN engine_open_time bigint;

kyuubi-server/src/main/resources/sql/derby/metadata-store-schema-1.7.0.derby.sql

@@ -19,6 +19,7 @@ CREATE TABLE metadata(
     create_time BIGINT NOT NULL, -- the metadata create time
     engine_type varchar(32) NOT NULL, -- the engine type
     cluster_manager varchar(128), -- the engine cluster manager
+    engine_open_time bigint, -- the engine open time
     engine_id varchar(128), -- the engine application id
     engine_name clob, -- the engine application name
     engine_url varchar(1024), -- the engine tracking url

kyuubi-server/src/main/resources/sql/derby/upgrade-1.6.0-to-1.7.0.derby.sql

@@ -1 +1,2 @@
 RUN '001-KYUUBI-3967.derby.sql';
+RUN '002-KYUUBI-4119.derby.sql';

kyuubi-server/src/main/resources/sql/mysql/002-KYUUBI-4119.mysql.sql

@@ -0,0 +1,3 @@
+SELECT '< KYUUBI-4119: Return app submission time for batch >' AS ' ';
+
+ALTER TABLE metadata ADD COLUMN engine_open_time bigint COMMENT 'the engine open time';

kyuubi-server/src/main/resources/sql/mysql/metadata-store-schema-1.7.0.mysql.sql

@@ -17,6 +17,7 @@ CREATE TABLE IF NOT EXISTS metadata(
     create_time BIGINT NOT NULL COMMENT 'the metadata create time',
     engine_type varchar(32) NOT NULL COMMENT 'the engine type',
     cluster_manager varchar(128) COMMENT 'the engine cluster manager',
+    engine_open_time bigint COMMENT 'the engine open time',
     engine_id varchar(128) COMMENT 'the engine application id',
     engine_name mediumtext COMMENT 'the engine application name',
     engine_url varchar(1024) COMMENT 'the engine tracking url',

kyuubi-server/src/main/resources/sql/mysql/upgrade-1.6.0-to-1.7.0.mysql.sql

@@ -1,3 +1,4 @@
 SELECT '< Upgrading MetaStore schema from 1.6.0 to 1.7.0 >' AS ' ';
 SOURCE 001-KYUUBI-3967.mysql.sql;
+SOURCE 002-KYUUBI-4119.mysql.sql;
 SELECT '< Finished upgrading MetaStore schema from 1.6.0 to 1.7.0 >' AS ' ';

kyuubi-server/src/main/scala/org/apache/kyuubi/operation/BatchJobSubmission.scala

@@ -74,6 +74,9 @@ class BatchJobSubmission(
   private var killMessage: KillResponse = (false, "UNKNOWN")
   def getKillMessage: KillResponse = killMessage
 
+  @volatile private var _appSubmissionTime = recoveryMetadata.map(_.engineOpenTime).getOrElse(0L)
+  def appSubmissionTime: Long = _appSubmissionTime
+
   @VisibleForTesting
   private[kyuubi] val builder: ProcBuilder = {
     Option(batchType).map(_.toUpperCase(Locale.ROOT)) match {
@@ -96,7 +99,14 @@ class BatchJobSubmission(
 
   override private[kyuubi] def currentApplicationInfo: Option[ApplicationInfo] = {
     // only the ApplicationInfo with non-empty id is valid for the operation
-    applicationManager.getApplicationInfo(builder.clusterManager(), batchId).filter(_.id != null)
+    val applicationInfo =
+      applicationManager.getApplicationInfo(builder.clusterManager(), batchId).filter(_.id != null)
+    applicationInfo.foreach { _ =>
+      if (_appSubmissionTime <= 0) {
+        _appSubmissionTime = System.currentTimeMillis()
+      }
+    }
+    applicationInfo
   }
 
   private[kyuubi] def killBatchApplication(): KillResponse = {
@@ -127,6 +137,7 @@ class BatchJobSubmission(
       val metadataToUpdate = Metadata(
         identifier = batchId,
         state = state.toString,
+        engineOpenTime = appSubmissionTime,
         engineId = status.id,
         engineName = status.name,
         engineUrl = status.url.orNull,

kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/BatchesResource.scala

@@ -94,6 +94,7 @@ private[v1] class BatchesResource extends ApiRequestContext with Logging {
       session.user,
       batchOp.batchType,
       name,
+      batchOp.appSubmissionTime,
       appId,
       appUrl,
       appState,
@@ -130,6 +131,7 @@ private[v1] class BatchesResource extends ApiRequestContext with Logging {
         metadata.username,
         metadata.engineType,
         name,
+        metadata.engineOpenTime,
         appId,
         appUrl,
         appState,

kyuubi-server/src/main/scala/org/apache/kyuubi/server/metadata/MetadataManager.scala

@@ -311,6 +311,7 @@ object MetadataManager extends Logging {
       batchMetadata.username,
       batchMetadata.engineType,
       name,
+      batchMetadata.engineOpenTime,
       batchMetadata.engineId,
       batchMetadata.engineUrl,
       batchMetadata.engineState,

kyuubi-server/src/main/scala/org/apache/kyuubi/server/metadata/api/Metadata.scala

@@ -41,6 +41,7 @@ import org.apache.kyuubi.session.SessionType.SessionType
  * @param createTime the create time.
  * @param engineType the engine type.
  * @param clusterManager the engine cluster manager.
+ * @param engineOpenTime the engine open time
  * @param engineId the engine id.
  * @param engineName the engine name.
  * @param engineUrl the engine tracking url.
@@ -65,6 +66,7 @@ case class Metadata(
     createTime: Long = 0L,
     engineType: String = null,
     clusterManager: Option[String] = None,
+    engineOpenTime: Long = 0L,
     engineId: String = null,
     engineName: String = null,
     engineUrl: String = null,

kyuubi-server/src/main/scala/org/apache/kyuubi/server/metadata/jdbc/JDBCMetadataStore.scala

@@ -265,6 +265,10 @@ class JDBCMetadataStore(conf: KyuubiConf) extends MetadataStore with Logging {
       setClauses += " end_time = ? "
       params += metadata.endTime
     }
+    if (metadata.engineOpenTime > 0) {
+      setClauses += " engine_open_time = ? "
+      params += metadata.engineOpenTime
+    }
     Option(metadata.engineId).foreach { _ =>
       setClauses += " engine_id = ? "
       params += metadata.engineId

kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiBatchSessionImpl.scala

@@ -161,7 +161,7 @@ class KyuubiBatchSessionImpl(
 
   private[kyuubi] def onEngineOpened(): Unit = {
     if (sessionEvent.openedTime <= 0) {
-      sessionEvent.openedTime = System.currentTimeMillis()
+      sessionEvent.openedTime = batchJobSubmissionOp.appSubmissionTime
       EventBus.post(sessionEvent)
     }
   }

kyuubi-server/src/test/scala/org/apache/kyuubi/server/api/v1/BatchesResourceSuite.scala

@@ -100,6 +100,9 @@ class BatchesResourceSuite extends KyuubiFunSuite with RestFrontendTestHelper wi
     assert(batch.getName === sparkBatchTestAppName)
     assert(batch.getCreateTime > 0)
     assert(batch.getEndTime === 0)
+    if (batch.getAppId != null) {
+      assert(batch.getAppSubmissionTime > 0)
+    }
 
     // invalid batchId
     getBatchResponse = webTarget.path(s"api/v1/batches/invalidBatchId")