This is an automated email from the ASF dual-hosted git repository.
chengpan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kyuubi.git
The following commit(s) were added to refs/heads/master by this push:
new 75c7a74e5 [KYUUBI #4920] Add OperationRestApi
75c7a74e5 is described below
commit 75c7a74e52811dc489e9b6a97ab7cbd1c86f6a32
Author: Tianlin Liao <[email protected]>
AuthorDate: Mon Jun 26 13:36:25 2023 +0800
[KYUUBI #4920] Add OperationRestApi
### _Why are the changes needed?_
close https://github.com/apache/kyuubi/issues/4920
### _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
- [ ] [Run
test](https://kyuubi.readthedocs.io/en/master/develop_tools/testing.html#running-tests)
locally before make a pull request
Closes #4921 from lightning-L/kyuubi-4920.
Closes #4920
d395f8a8a [Tianlin Liao] remove KyuubiEvent
3a6730f9b [Tianlin Liao] [KYUUBI #4920] add OperationRestApi
Authored-by: Tianlin Liao <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>
---
.../org/apache/kyuubi/client/OperationRestApi.java | 77 +++++
.../kyuubi/client/api/v1/dto/KyuubiEvent.java | 20 --
.../client/api/v1/dto/KyuubiOperationEvent.java | 343 +++++++++++++++++++++
.../client/api/v1/dto/KyuubiSessionEvent.java | 2 +-
.../server/rest/client/OperationRestApiSuite.scala | 123 ++++++++
5 files changed, 544 insertions(+), 21 deletions(-)
diff --git
a/kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/OperationRestApi.java
b/kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/OperationRestApi.java
new file mode 100644
index 000000000..ad659a5d4
--- /dev/null
+++
b/kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/OperationRestApi.java
@@ -0,0 +1,77 @@
+/*
+ * 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.client;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.kyuubi.client.api.v1.dto.*;
+import org.apache.kyuubi.client.util.JsonUtils;
+
+public class OperationRestApi {
+
+ private KyuubiRestClient client;
+
+ private static final String API_BASE_PATH = "operations";
+
+ private OperationRestApi() {}
+
+ public OperationRestApi(KyuubiRestClient client) {
+ this.client = client;
+ }
+
+ public KyuubiOperationEvent getOperationEvent(String operationHandleStr) {
+ String path = String.format("%s/%s/event", API_BASE_PATH,
operationHandleStr);
+ return this.getClient()
+ .get(path, new HashMap<>(), KyuubiOperationEvent.class,
client.getAuthHeader());
+ }
+
+ public String applyOperationAction(OpActionRequest request, String
operationHandleStr) {
+ String path = String.format("%s/%s", API_BASE_PATH, operationHandleStr);
+ return this.getClient().put(path, JsonUtils.toJson(request),
client.getAuthHeader());
+ }
+
+ public ResultSetMetaData getResultSetMetadata(String operationHandleStr) {
+ String path = String.format("%s/%s/resultsetmetadata", API_BASE_PATH,
operationHandleStr);
+ return this.getClient()
+ .get(path, new HashMap<>(), ResultSetMetaData.class,
client.getAuthHeader());
+ }
+
+ public OperationLog getOperationLog(String operationHandleStr, int maxRows) {
+ String path = String.format("%s/%s/log", API_BASE_PATH,
operationHandleStr);
+ Map<String, Object> params = new HashMap<>();
+ params.put("maxrows", maxRows);
+ return this.getClient().get(path, params, OperationLog.class,
client.getAuthHeader());
+ }
+
+ public ResultRowSet getNextRowSet(String operationHandleStr) {
+ return getNextRowSet(operationHandleStr, null, null);
+ }
+
+ public ResultRowSet getNextRowSet(
+ String operationHandleStr, String fetchOrientation, Integer maxRows) {
+ String path = String.format("%s/%s/rowset", API_BASE_PATH,
operationHandleStr);
+ Map<String, Object> params = new HashMap<>();
+ if (fetchOrientation != null) params.put("fetchorientation",
fetchOrientation);
+ if (maxRows != null) params.put("maxrows", maxRows);
+ return this.getClient().get(path, params, ResultRowSet.class,
client.getAuthHeader());
+ }
+
+ private IRestClient getClient() {
+ return this.client.getHttpClient();
+ }
+}
diff --git
a/kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/api/v1/dto/KyuubiEvent.java
b/kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/api/v1/dto/KyuubiEvent.java
deleted file mode 100644
index 8de125089..000000000
---
a/kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/api/v1/dto/KyuubiEvent.java
+++ /dev/null
@@ -1,20 +0,0 @@
-/*
- * 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.client.api.v1.dto;
-
-public interface KyuubiEvent {}
diff --git
a/kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/api/v1/dto/KyuubiOperationEvent.java
b/kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/api/v1/dto/KyuubiOperationEvent.java
new file mode 100644
index 000000000..13c40eecf
--- /dev/null
+++
b/kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/api/v1/dto/KyuubiOperationEvent.java
@@ -0,0 +1,343 @@
+/*
+ * 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.client.api.v1.dto;
+
+import java.util.Map;
+
+public class KyuubiOperationEvent {
+
+ private String statementId;
+
+ private String remoteId;
+
+ private String statement;
+
+ private boolean shouldRunAsync;
+
+ private String state;
+
+ private long eventTime;
+
+ private long createTime;
+
+ private long startTime;
+
+ private long completeTime;
+
+ private Throwable exception;
+
+ private String sessionId;
+
+ private String sessionUser;
+
+ private String sessionType;
+
+ private String kyuubiInstance;
+
+ private Map<String, String> metrics;
+
+ public KyuubiOperationEvent() {}
+
+ public KyuubiOperationEvent(
+ String statementId,
+ String remoteId,
+ String statement,
+ boolean shouldRunAsync,
+ String state,
+ long eventTime,
+ long createTime,
+ long startTime,
+ long completeTime,
+ Throwable exception,
+ String sessionId,
+ String sessionUser,
+ String sessionType,
+ String kyuubiInstance,
+ Map<String, String> metrics) {
+ this.statementId = statementId;
+ this.remoteId = remoteId;
+ this.statement = statement;
+ this.shouldRunAsync = shouldRunAsync;
+ this.state = state;
+ this.eventTime = eventTime;
+ this.createTime = createTime;
+ this.startTime = startTime;
+ this.completeTime = completeTime;
+ this.exception = exception;
+ this.sessionId = sessionId;
+ this.sessionUser = sessionUser;
+ this.sessionType = sessionType;
+ this.kyuubiInstance = kyuubiInstance;
+ this.metrics = metrics;
+ }
+
+ public static KyuubiOperationEvent.KyuubiOperationEventBuilder builder() {
+ return new KyuubiOperationEvent.KyuubiOperationEventBuilder();
+ }
+
+ public static class KyuubiOperationEventBuilder {
+ private String statementId;
+
+ private String remoteId;
+
+ private String statement;
+
+ private boolean shouldRunAsync;
+
+ private String state;
+
+ private long eventTime;
+
+ private long createTime;
+
+ private long startTime;
+
+ private long completeTime;
+
+ private Throwable exception;
+
+ private String sessionId;
+
+ private String sessionUser;
+
+ private String sessionType;
+
+ private String kyuubiInstance;
+
+ private Map<String, String> metrics;
+
+ public KyuubiOperationEventBuilder() {}
+
+ public KyuubiOperationEvent.KyuubiOperationEventBuilder statementId(final
String statementId) {
+ this.statementId = statementId;
+ return this;
+ }
+
+ public KyuubiOperationEvent.KyuubiOperationEventBuilder remoteId(final
String remoteId) {
+ this.remoteId = remoteId;
+ return this;
+ }
+
+ public KyuubiOperationEvent.KyuubiOperationEventBuilder statement(final
String statement) {
+ this.statement = statement;
+ return this;
+ }
+
+ public KyuubiOperationEvent.KyuubiOperationEventBuilder shouldRunAsync(
+ final boolean shouldRunAsync) {
+ this.shouldRunAsync = shouldRunAsync;
+ return this;
+ }
+
+ public KyuubiOperationEvent.KyuubiOperationEventBuilder state(final String
state) {
+ this.state = state;
+ return this;
+ }
+
+ public KyuubiOperationEvent.KyuubiOperationEventBuilder eventTime(final
long eventTime) {
+ this.eventTime = eventTime;
+ return this;
+ }
+
+ public KyuubiOperationEvent.KyuubiOperationEventBuilder createTime(final
long createTime) {
+ this.createTime = createTime;
+ return this;
+ }
+
+ public KyuubiOperationEvent.KyuubiOperationEventBuilder startTime(final
long startTime) {
+ this.startTime = startTime;
+ return this;
+ }
+
+ public KyuubiOperationEvent.KyuubiOperationEventBuilder completeTime(final
long completeTime) {
+ this.completeTime = completeTime;
+ return this;
+ }
+
+ public KyuubiOperationEvent.KyuubiOperationEventBuilder exception(final
Throwable exception) {
+ this.exception = exception;
+ return this;
+ }
+
+ public KyuubiOperationEvent.KyuubiOperationEventBuilder sessionId(final
String sessionId) {
+ this.sessionId = sessionId;
+ return this;
+ }
+
+ public KyuubiOperationEvent.KyuubiOperationEventBuilder sessionUser(final
String sessionUser) {
+ this.sessionUser = sessionUser;
+ return this;
+ }
+
+ public KyuubiOperationEvent.KyuubiOperationEventBuilder sessionType(final
String sessionType) {
+ this.sessionType = sessionType;
+ return this;
+ }
+
+ public KyuubiOperationEvent.KyuubiOperationEventBuilder kyuubiInstance(
+ final String kyuubiInstance) {
+ this.kyuubiInstance = kyuubiInstance;
+ return this;
+ }
+
+ public KyuubiOperationEvent.KyuubiOperationEventBuilder metrics(
+ final Map<String, String> metrics) {
+ this.metrics = metrics;
+ return this;
+ }
+
+ public KyuubiOperationEvent build() {
+ return new KyuubiOperationEvent(
+ statementId,
+ remoteId,
+ statement,
+ shouldRunAsync,
+ state,
+ eventTime,
+ createTime,
+ startTime,
+ completeTime,
+ exception,
+ sessionId,
+ sessionUser,
+ sessionType,
+ kyuubiInstance,
+ metrics);
+ }
+ }
+
+ public String getStatementId() {
+ return statementId;
+ }
+
+ public void setStatementId(String statementId) {
+ this.statementId = statementId;
+ }
+
+ public String getRemoteId() {
+ return remoteId;
+ }
+
+ public void setRemoteId(String remoteId) {
+ this.remoteId = remoteId;
+ }
+
+ public String getStatement() {
+ return statement;
+ }
+
+ public void setStatement(String statement) {
+ this.statement = statement;
+ }
+
+ public boolean isShouldRunAsync() {
+ return shouldRunAsync;
+ }
+
+ public void setShouldRunAsync(boolean shouldRunAsync) {
+ this.shouldRunAsync = shouldRunAsync;
+ }
+
+ public String getState() {
+ return state;
+ }
+
+ public void setState(String state) {
+ this.state = state;
+ }
+
+ public long getEventTime() {
+ return eventTime;
+ }
+
+ public void setEventTime(long eventTime) {
+ this.eventTime = eventTime;
+ }
+
+ public long getCreateTime() {
+ return createTime;
+ }
+
+ public void setCreateTime(long createTime) {
+ this.createTime = createTime;
+ }
+
+ public long getStartTime() {
+ return startTime;
+ }
+
+ public void setStartTime(long startTime) {
+ this.startTime = startTime;
+ }
+
+ public long getCompleteTime() {
+ return completeTime;
+ }
+
+ public void setCompleteTime(long completeTime) {
+ this.completeTime = completeTime;
+ }
+
+ public Throwable getException() {
+ return exception;
+ }
+
+ public void setException(Throwable exception) {
+ this.exception = exception;
+ }
+
+ public String getSessionId() {
+ return sessionId;
+ }
+
+ public void setSessionId(String sessionId) {
+ this.sessionId = sessionId;
+ }
+
+ public String getSessionUser() {
+ return sessionUser;
+ }
+
+ public void setSessionUser(String sessionUser) {
+ this.sessionUser = sessionUser;
+ }
+
+ public String getSessionType() {
+ return sessionType;
+ }
+
+ public void setSessionType(String sessionType) {
+ this.sessionType = sessionType;
+ }
+
+ public String getKyuubiInstance() {
+ return kyuubiInstance;
+ }
+
+ public void setKyuubiInstance(String kyuubiInstance) {
+ this.kyuubiInstance = kyuubiInstance;
+ }
+
+ public Map<String, String> getMetrics() {
+ return metrics;
+ }
+
+ public void setMetrics(Map<String, String> metrics) {
+ this.metrics = metrics;
+ }
+}
diff --git
a/kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/api/v1/dto/KyuubiSessionEvent.java
b/kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/api/v1/dto/KyuubiSessionEvent.java
index 4c3cbcfd5..34d306fed 100644
---
a/kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/api/v1/dto/KyuubiSessionEvent.java
+++
b/kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/api/v1/dto/KyuubiSessionEvent.java
@@ -19,7 +19,7 @@ package org.apache.kyuubi.client.api.v1.dto;
import java.util.Map;
-public class KyuubiSessionEvent implements KyuubiEvent {
+public class KyuubiSessionEvent {
private String sessionId;
diff --git
a/kyuubi-server/src/test/scala/org/apache/kyuubi/server/rest/client/OperationRestApiSuite.scala
b/kyuubi-server/src/test/scala/org/apache/kyuubi/server/rest/client/OperationRestApiSuite.scala
new file mode 100644
index 000000000..fed685c44
--- /dev/null
+++
b/kyuubi-server/src/test/scala/org/apache/kyuubi/server/rest/client/OperationRestApiSuite.scala
@@ -0,0 +1,123 @@
+/*
+ * 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.server.rest.client
+
+import scala.collection.JavaConverters._
+
+import
org.apache.hive.service.rpc.thrift.TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V2
+import org.scalatest.concurrent.PatienceConfiguration.Timeout
+import org.scalatest.time.SpanSugar.convertIntToGrainOfTime
+
+import org.apache.kyuubi.RestClientTestHelper
+import org.apache.kyuubi.client.{KyuubiRestClient, OperationRestApi}
+import org.apache.kyuubi.client.api.v1.dto.OpActionRequest
+import org.apache.kyuubi.client.exception.KyuubiRestException
+import org.apache.kyuubi.operation.OperationState
+
+class OperationRestApiSuite extends RestClientTestHelper {
+
+ test("get an operation event") {
+ val statementHandleStr = getOpHandleStr()
+
+ withOperationRestApi { operationRestApi =>
+ val kyuubiEvent = operationRestApi.getOperationEvent(statementHandleStr)
+ assert("show tables".equals(kyuubiEvent.getStatement))
+ assert(kyuubiEvent.isShouldRunAsync == true)
+ }
+ }
+
+ test("apply operation action") {
+ val statementHandleStr = getOpHandleStr(
+ "SELECT java_method('java.lang.Thread', 'sleep', 10000l)")
+
+ withOperationRestApi { operationRestApi =>
+ // successful request
+ operationRestApi.applyOperationAction(new OpActionRequest("cancel"),
statementHandleStr)
+ eventually(Timeout(5.seconds)) {
+ val kyuubiEvent =
operationRestApi.getOperationEvent(statementHandleStr)
+ assert(kyuubiEvent.getState === OperationState.CANCELED.name)
+ }
+
+ operationRestApi.applyOperationAction(new OpActionRequest("close"),
statementHandleStr)
+ // failed request
+ assertThrows[KyuubiRestException] {
+ operationRestApi.applyOperationAction(new OpActionRequest("close"),
statementHandleStr)
+ }
+
+ // invalid operation
+ assertThrows[KyuubiRestException] {
+ operationRestApi.applyOperationAction(new OpActionRequest("fake"),
statementHandleStr)
+ }
+ }
+ }
+
+ test("get result set metadata/get operation log/get result row set") {
+ val statementHandleStr = getOpHandleStr("select \"test_value\", 1, 0.32d,
true")
+
+ withOperationRestApi { operationRestApi =>
+ // wait for complete
+ eventually(Timeout(5.seconds)) {
+ val kyuubiEvent =
operationRestApi.getOperationEvent(statementHandleStr)
+ assert(kyuubiEvent.getState === OperationState.FINISHED.name)
+ }
+
+ val resultSetMetadata =
operationRestApi.getResultSetMetadata(statementHandleStr)
+ assert(resultSetMetadata.getColumns.size == 4)
+
assert(resultSetMetadata.getColumns.get(0).getColumnName.equals("test_value"))
+
+ val logRowSet = operationRestApi.getOperationLog(statementHandleStr, 10)
+ assert(logRowSet.getLogRowSet.asScala.exists(
+ _.contains("select \"test_value\", 1, 0.32d, true")))
+ assert(logRowSet.getRowCount === 10)
+
+ val resultRowSet = operationRestApi.getNextRowSet(statementHandleStr)
+
assert("test_value".equals(resultRowSet.getRows.asScala.head.getFields.asScala.head.getValue))
+ assert(resultRowSet.getRowCount == 1)
+ }
+ }
+
+ def withOperationRestApi[T](f: OperationRestApi => T): T = {
+ val basicKyuubiRestClient: KyuubiRestClient =
+ KyuubiRestClient.builder(baseUri.toString)
+ .authHeaderMethod(KyuubiRestClient.AuthHeaderMethod.BASIC)
+ .username(ldapUser)
+ .password(ldapUserPasswd)
+ .socketTimeout(30000)
+ .build()
+ val operationRestApi = new OperationRestApi(basicKyuubiRestClient)
+ f(operationRestApi)
+ }
+
+ def getOpHandleStr(statement: String = "show tables"): String = {
+ val sessionHandle = fe.be.openSession(
+ HIVE_CLI_SERVICE_PROTOCOL_V2,
+ "admin",
+ "123456",
+ "localhost",
+ Map("testConfig" -> "testValue"))
+
+ val op =
+ if (statement.nonEmpty) {
+ fe.be.executeStatement(sessionHandle, statement, Map.empty, runAsync =
true, 3000)
+ } else {
+ fe.be.getCatalogs(sessionHandle)
+ }
+
+ op.identifier.toString
+ }
+}
