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

[fe...@apache.org]

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 <fw...@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 <fw...@ebay.com>
    Signed-off-by: fwang12 <fw...@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)
+    }
+  }
 }