[kyuubi] branch branch-1.6 updated: [KYUUBI #4164] Accumulate the operation terminal state

[feiwang]

This is an automated email from the ASF dual-hosted git repository.

feiwang pushed a commit to branch branch-1.6
in repository https://gitbox.apache.org/repos/asf/kyuubi.git


The following commit(s) were added to refs/heads/branch-1.6 by this push:
     new be23ec805 [KYUUBI #4164] Accumulate the operation terminal state
be23ec805 is described below

commit be23ec8051880ce8aac2fc926d0b077fde3e2121
Author: fwang12 <fwan...@ebay.com>
AuthorDate: Fri Jan 13 21:15:18 2023 +0800

    [KYUUBI #4164] Accumulate the operation terminal state
    
    ### _Why are the changes needed?_
    
    For the KyuubiOperation, its final state should be `CLOSED`, and now the 
`FINISHED`, `ERROR` states, which are important for kyuubi administer, are not 
accumulated.
    
    In this pr, we check whether the old state is terminal state, if it is, do 
not decrease it.
    
    ### _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
    
    - [x] [Run 
test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests)
 locally before make a pull request
    
    Closes #4164 from turboFei/terminate_acc.
    
    Closes #4164
    
    16e559c6f [fwang12] comment
    c9d86f4b6 [fwang12] add ut
    4fcf6accc [fwang12] Accumulate the terminate state
    
    Authored-by: fwang12 <fwan...@ebay.com>
    Signed-off-by: fwang12 <fwan...@ebay.com>
---
 .../org/apache/kyuubi/operation/KyuubiOperation.scala  |  4 +++-
 .../kyuubi/operation/KyuubiOperationPerUserSuite.scala | 18 ++++++++++++++++++
 2 files changed, 21 insertions(+), 1 deletion(-)

diff --git 
a/kyuubi-server/src/main/scala/org/apache/kyuubi/operation/KyuubiOperation.scala
 
b/kyuubi-server/src/main/scala/org/apache/kyuubi/operation/KyuubiOperation.scala
index 2d28c767e..2ed170f99 100644
--- 
a/kyuubi-server/src/main/scala/org/apache/kyuubi/operation/KyuubiOperation.scala
+++ 
b/kyuubi-server/src/main/scala/org/apache/kyuubi/operation/KyuubiOperation.scala
@@ -158,7 +158,9 @@ abstract class KyuubiOperation(session: Session) extends 
AbstractOperation(sessi
 
   override def setState(newState: OperationState): Unit = {
     MetricsSystem.tracing { ms =>
-      ms.markMeter(MetricRegistry.name(OPERATION_STATE, opType, 
state.toString.toLowerCase), -1)
+      if (!OperationState.isTerminal(state)) {
+        ms.markMeter(MetricRegistry.name(OPERATION_STATE, opType, 
state.toString.toLowerCase), -1)
+      }
       ms.markMeter(MetricRegistry.name(OPERATION_STATE, opType, 
newState.toString.toLowerCase))
       ms.markMeter(MetricRegistry.name(OPERATION_STATE, 
newState.toString.toLowerCase))
     }
diff --git 
a/kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerUserSuite.scala
 
b/kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerUserSuite.scala
index b6aba4c09..868aacbf2 100644
--- 
a/kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerUserSuite.scala
+++ 
b/kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerUserSuite.scala
@@ -28,6 +28,7 @@ import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf.KYUUBI_ENGINE_ENV_PREFIX
 import org.apache.kyuubi.engine.SemanticVersion
 import org.apache.kyuubi.jdbc.hive.KyuubiStatement
+import org.apache.kyuubi.metrics.{MetricsConstants, MetricsSystem}
 import org.apache.kyuubi.session.{KyuubiSessionImpl, KyuubiSessionManager, 
SessionHandle}
 
 class KyuubiOperationPerUserSuite
@@ -288,4 +289,21 @@ class KyuubiOperationPerUserSuite
       assert(!result.next())
     }
   }
+
+  test("accumulate the operation terminal state") {
+    val opType = classOf[ExecuteStatement].getSimpleName
+    val finishedMetric = s"${MetricsConstants.OPERATION_STATE}.$opType" +
+      s".${OperationState.FINISHED.toString.toLowerCase}"
+    val closedMetric = s"${MetricsConstants.OPERATION_STATE}.$opType" +
+      s".${OperationState.CLOSED.toString.toLowerCase}"
+    val finishedCount = MetricsSystem.meterValue(finishedMetric).getOrElse(0L)
+    val closedCount = MetricsSystem.meterValue(finishedMetric).getOrElse(0L)
+    withJdbcStatement() { statement =>
+      statement.executeQuery("select engine_name()")
+    }
+    eventually(timeout(5.seconds), interval(100.milliseconds)) {
+      assert(MetricsSystem.meterValue(finishedMetric).getOrElse(0L) > 
finishedCount)
+      assert(MetricsSystem.meterValue(closedMetric).getOrElse(0L) > 
closedCount)
+    }
+  }
 }