add new module common-grpc

kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala

@@ -2731,6 +2731,28 @@ object KyuubiConf {
       .stringConf
       .createWithDefault("ENGINE")
 
+  val ENGINE_SPARK_CONNECT_GRPC_BINDING_PORT: ConfigEntry[Int] =
+    buildConf("kyuubi.engine.spark.connect.grpc.bind.port")
+      .doc("The port is used in spark connect frontendService start GrpcServer")
+      .version("1.9.0")
+      .intConf
+      .createWithDefault(15002)
+
+  val ENGINE_SPARK_CONNECT_GRPC_MAX_INBOUND_MESSAGE_SIZE: ConfigEntry[Int] =
+    buildConf("kyuubi.engine.spark.connect.grpc.max.inbound.message.size")
+      .doc("Sets the maximum inbound message in bytes size for the gRPC requests." +
+        "Requests with a larger payload will fail.")
+      .version("1.9.0")
+      .intConf
+      .createWithDefault(128 * 1024 * 1024)
+
+  val ENGINE_SPARK_CONNECT_GRPC_BINDING_HOST: ConfigEntry[Option[String]] =
+    buildConf("kyuubi.engine.spark.connect.grpc.bind.host")
+      .doc("Hostname or IP of the machine on which to run the grpc server in frontend service ")
+      .version("1.9.0")
+      .serverOnly
+      .fallbackConf(FRONTEND_BIND_HOST)
+
   val ENGINE_SPARK_SHOW_PROGRESS: ConfigEntry[Boolean] =
     buildConf("kyuubi.session.engine.spark.showProgress")
       .doc("When true, show the progress bar in the Spark's engine log.")

kyuubi-grpc/src/main/scala/org/apache/kyuubi/grpc/events/OperationEventsManager.scala

@@ -0,0 +1,141 @@
+/*
+ * 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.grpc.events
+
+import org.apache.kyuubi.grpc.operation.GrpcOperation
+import org.apache.kyuubi.grpc.session.GrpcSession
+import org.apache.kyuubi.grpc.utils.Clock
+
+object OperationEventsManager {
+  // TODO: make this configurable
+  val MAX_STATEMENT_TEXT_SIZE = 65535
+}
+
+sealed abstract class OperationStatus(value: Int)
+
+object OperationStatus {
+  case object Pending extends OperationStatus(0)
+  case object Started extends OperationStatus(1)
+  case object Analyzed extends OperationStatus(2)
+  case object ReadyForExecution extends OperationStatus(3)
+  case object Finished extends OperationStatus(4)
+  case object Failed extends OperationStatus(5)
+  case object Canceled extends OperationStatus(6)
+  case object Closed extends OperationStatus(7)
+}
+abstract class OperationEventsManager(operation: GrpcOperation, clock: Clock) {
+  private def operationId: String = operation.operationKey.operationId
+
+  private def session: GrpcSession = operation.grpcSession
+
+  private def sessionId: String = session.sessionKey.sessionId
+
+  private def sessionStatus = session.sessionEventsManager.status
+
+  protected var _status: OperationStatus = OperationStatus.Pending
+
+  private var error = Option.empty[Boolean]
+
+  private var canceled = Option.empty[Boolean]
+
+  private var producedRowCount = Option.empty[Long]
+
+  private def status: OperationStatus = _status
+
+  private def hasCanceled: Option[Boolean] = canceled
+
+  private def hasError: Option[Boolean] = error
+
+  private def getProduceRowCount: Option[Long] = producedRowCount
+
+  def postStarted(): Unit = {
+    assertStatus(List(OperationStatus.Pending), OperationStatus.Started)
+  }
+
+  def postAnalyzed(analyzedPlan: Option[Any] = None): Unit = {
+    assertStatus(List(OperationStatus.Started, OperationStatus.Analyzed), OperationStatus.Analyzed)
+  }
+
+  def postReadyForExecution(): Unit = {
+    assertStatus(List(OperationStatus.Analyzed), OperationStatus.ReadyForExecution)
+  }
+
+  def postCanceled(): Unit = {
+    assertStatus(
+      List(
+        OperationStatus.Started,
+        OperationStatus.Analyzed,
+        OperationStatus.ReadyForExecution,
+        OperationStatus.Finished,
+        OperationStatus.Failed),
+      OperationStatus.Canceled)
+    canceled = Some(true)
+  }
+
+  def postFailed(errorMessage: String): Unit = {
+    assertStatus(
+      List(
+        OperationStatus.Started,
+        OperationStatus.Analyzed,
+        OperationStatus.ReadyForExecution,
+        OperationStatus.Finished),
+      OperationStatus.Failed)
+    error = Some(true)
+  }
+
+  def postFinished(producedRowCountOpt: Option[Long] = None): Unit = {
+    assertStatus(
+      List(
+        OperationStatus.Started,
+        OperationStatus.ReadyForExecution),
+      OperationStatus.Finished)
+    producedRowCount = producedRowCountOpt
+  }
+
+  def postClosed(): Unit = {
+    assertStatus(
+      List(
+        OperationStatus.Finished,
+        OperationStatus.Failed,
+        OperationStatus.Canceled),
+      OperationStatus.Closed)
+  }
+
+  def status_(operationStatus: OperationStatus): Unit = {
+    _status = operationStatus
+  }
+
+  private def assertStatus(
+      validStatuses: List[OperationStatus],
+      eventStatus: OperationStatus): Unit = {
+    if (!validStatuses.contains(status)) {
+      throw new IllegalStateException(
+        s"""
+           |operationId: $operationId with status ${status}
+           |is not within statuses $validStatuses for event $eventStatus
+           |""".stripMargin)
+    }
+//    if (sessionStatus != SessionStatus.Started) {
+//      throw new IllegalStateException(
+//        s"""
+//           |sessionId: $sessionId with status $sessionStatus
+//           |is not Started for event $eventStatus
+//           |""".stripMargin)
+//    }
+    _status = eventStatus
+  }
+}

kyuubi-grpc/src/main/scala/org/apache/kyuubi/grpc/events/SessionEventsManager.scala

@@ -0,0 +1,60 @@
+/*
+ * 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.grpc.events
+
+import org.apache.kyuubi.grpc.session.GrpcSession
+import org.apache.kyuubi.grpc.utils.Clock
+
+sealed abstract class SessionStatus(value: Int)
+
+object SessionStatus {
+  case object Pending extends SessionStatus(0)
+  case object Started extends SessionStatus(1)
+  case object Closed extends SessionStatus(2)
+}
+
+abstract class SessionEventsManager(session: GrpcSession, clock: Clock) {
+  private def sessionId: String = session.sessionKey.sessionId
+
+  private var _status: SessionStatus = SessionStatus.Pending
+
+  protected def status_(sessionStatus: SessionStatus): Unit = {
+    _status = sessionStatus
+  }
+
+  def status: SessionStatus = _status
+
+  def postStarted(): Unit = {
+    assertStatus(List(SessionStatus.Pending), SessionStatus.Started)
+    status_(SessionStatus.Started)
+  }
+
+  def postClosed(): Unit = {
+    assertStatus(List(SessionStatus.Started), SessionStatus.Closed)
+    status_(SessionStatus.Closed)
+  }
+
+  private def assertStatus(validStatuses: List[SessionStatus], eventStatus: SessionStatus): Unit = {
+    if (!validStatuses.contains(status)) {
+      throw new IllegalStateException(
+        s"""
+           |sessionId: $sessionId with status ${status}
+           |is not within statuses $validStatuses for event $eventStatus
+           |""".stripMargin)
+    }
+  }
+}

kyuubi-grpc/src/main/scala/org/apache/kyuubi/grpc/operation/AbstractGrpcOperation.scala

@@ -0,0 +1,78 @@
+/*
+ * 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.grpc.operation
+
+import java.util.concurrent.locks.ReentrantLock
+
+import org.apache.kyuubi.{KyuubiSQLException, Logging, Utils}
+import org.apache.kyuubi.grpc.session.GrpcSession
+
+abstract class AbstractGrpcOperation[S <: GrpcSession](session: S) extends GrpcOperation
+  with Logging {
+  final protected val opType: String = getClass.getSimpleName
+  final protected val createTime = System.currentTimeMillis()
+  protected def key: OperationKey
+  final private val operationTimeout: Long = 1000
+  private var lock: ReentrantLock = new ReentrantLock()
+
+  protected def withLockRequired[T](block: => T): T = Utils.withLockRequired(lock)(block)
+
+  @volatile protected var startTime: Long = _
+  @volatile protected var completedTime: Long = _
+  @volatile protected var lastAccessTime: Long = createTime
+
+  @volatile protected var operationException: KyuubiSQLException = _
+
+  protected def setOperationException(ex: KyuubiSQLException): Unit = {
+    this.operationException = ex
+  }
+
+  protected def runInternal(): Unit
+
+  protected def beforeRun(): Unit
+
+  protected def afterRun(): Unit
+
+  override def run(): Unit = {
+    beforeRun()
+    try {
+      runInternal()
+    } finally {
+      afterRun()
+    }
+  }
+
+  override def close(): Unit
+
+  override def operationKey: OperationKey = key
+
+  override def grpcSession: S = session
+
+}
+
+object OperationJobTag {
+  def apply(prefix: String, operationKey: OperationKey): String = {
+    s"${prefix}_" +
+      s"User_${operationKey.userId}_" +
+      s"Session_${operationKey.sessionId}_" +
+      s"Operation_${operationKey.operationId}"
+  }
+
+  def unapply(jobTag: String, prefix: String): Option[String] = {
+    if (jobTag.startsWith(prefix)) Some(jobTag) else None
+  }
+}

kyuubi-grpc/src/main/scala/org/apache/kyuubi/grpc/operation/GrpcOperation.scala

@@ -0,0 +1,33 @@
+/*
+ * 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.grpc.operation
+
+import org.apache.kyuubi.grpc.events.OperationEventsManager
+import org.apache.kyuubi.grpc.session.GrpcSession
+import org.apache.kyuubi.operation.log.OperationLog
+
+trait GrpcOperation {
+  def run(): Unit
+  def interrupt(): Unit
+  def close(): Unit
+
+  def getOperationLog: Option[OperationLog]
+  def isTimedOut: Boolean
+  def grpcSession: GrpcSession
+  def operationKey: OperationKey
+  def operationEventsManager: OperationEventsManager
+}