[KYUUBI #5333][JDBC] Adaptive call fetchLaunchEngineResult
### _Why are the changes needed?_ After https://github.com/apache/kyuubi/pull/3354, thrift `out of sequence` exception will be thrown when using a higher version of beeline to access a lower version of Kyuubi Server. ``` 2023-09-26 11:10:16.426 ERROR org.apache.kyuubi.server.KyuubiTBinaryFrontendService: Error fetching results: org.apache.thrift.protocol.TProtocolException: Required field 'operationHandle' is unset! Struct:TFetchResultsReq(operationHandle:null, orientation:FETCH_NEXT, maxRows:1000, fetchType:0) at org.apache.hive.service.rpc.thrift.TFetchResultsReq.validate(TFetchResultsReq.java:548) ~[hive-service-rpc-3.1.3.jar:3.1.3] at org.apache.hive.service.rpc.thrift.TCLIService$FetchResults_args.validate(TCLIService.java:15755) ~[hive-service-rpc-3.1.3.jar:3.1.3] at org.apache.hive.service.rpc.thrift.TCLIService$FetchResults_args$FetchResults_argsStandardScheme.write(TCLIService.java:15812) ~[hive-service-rpc-3.1.3.jar:3.1.3] at org.apache.hive.service.rpc.thrift.TCLIService$FetchResults_args$FetchResults_argsStandardScheme.write(TCLIService.java:15781) ~[hive-service-rpc-3.1.3.jar:3.1.3] at org.apache.hive.service.rpc.thrift.TCLIService$FetchResults_args.write(TCLIService.java:15732) ~[hive-service-rpc-3.1.3.jar:3.1.3] at org.apache.thrift.TServiceClient.sendBase(TServiceClient.java:71) ~[libthrift-0.9.3.jar:0.9.3] at org.apache.thrift.TServiceClient.sendBase(TServiceClient.java:62) ~[libthrift-0.9.3.jar:0.9.3] at org.apache.hive.service.rpc.thrift.TCLIService$Client.send_FetchResults(TCLIService.java:561) ~[hive-service-rpc-3.1.3.jar:3.1.3] at org.apache.hive.service.rpc.thrift.TCLIService$Client.FetchResults(TCLIService.java:553) ~[hive-service-rpc-3.1.3.jar:3.1.3] at org.apache.kyuubi.client.KyuubiSyncThriftClient.$anonfun$fetchResults$1(KyuubiSyncThriftClient.scala:393) ~[kyuubi-server_2.12-1.6.1-incubating.jar:1.6.1-incubating] at org.apache.kyuubi.client.KyuubiSyncThriftClient.$anonfun$withLockAcquiredAsyncRequest$2(KyuubiSyncThriftClient.scala:136) ~[kyuubi-server_2.12-1.6.1-incubating.jar:1.6.1-incubating] at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[?:1.8.0_152] at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) ~[?:1.8.0_152] at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) ~[?:1.8.0_152] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_152] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_152] at java.lang.Thread.run(Thread.java:748) ~[?:1.8.0_152] ``` This PR introduces a negotiation mechanism to make the JDBC client adaptively call `fetchLaunchEngineResult` to address such a compatible issue. It supposes that the higher version of JDBC client could seamlessly communicate with the lower version of Kyuubi Server after this patch, but the JDBC client fetches the engine information only when the connected Server applies this patch too. ### _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.readthedocs.io/en/master/contributing/code/testing.html#running-tests) locally before make a pull request ### _Was this patch authored or co-authored using generative AI tooling?_ No. Closes #5333 from pan3793/launch-engine-result. Closes #5333 ca9bd5361 [Cheng Pan] [JDBC] Adaptive call fetchLaunchEngineResult Authored-by: Cheng Pan <chengpan@apache.org> Signed-off-by: Cheng Pan <chengpan@apache.org>
This commit is contained in:
Cheng Pan
parent
8ef6ca329e
commit
36eb81be26
@ -38,6 +38,8 @@ object KyuubiReservedKeys {
|
||||
"kyuubi.session.engine.launch.handle.guid"
|
||||
final val KYUUBI_SESSION_ENGINE_LAUNCH_HANDLE_SECRET =
|
||||
"kyuubi.session.engine.launch.handle.secret"
|
||||
final val KYUUBI_SESSION_ENGINE_LAUNCH_SUPPORT_RESULT =
|
||||
"kyuubi.session.engine.launch.support.result"
|
||||
final val KYUUBI_OPERATION_SET_CURRENT_CATALOG = "kyuubi.operation.set.current.catalog"
|
||||
final val KYUUBI_OPERATION_GET_CURRENT_CATALOG = "kyuubi.operation.get.current.catalog"
|
||||
final val KYUUBI_OPERATION_SET_CURRENT_DATABASE = "kyuubi.operation.set.current.database"
|
||||
|
||||
@ -104,6 +104,7 @@ public class KyuubiConnection implements SQLConnection, KyuubiLoggable {
|
||||
private Thread engineLogThread;
|
||||
private boolean engineLogInflight = true;
|
||||
private volatile boolean launchEngineOpCompleted = false;
|
||||
private boolean launchEngineOpSupportResult = false;
|
||||
private String engineId = "";
|
||||
private String engineName = "";
|
||||
private String engineUrl = "";
|
||||
@ -770,6 +771,10 @@ public class KyuubiConnection implements SQLConnection, KyuubiLoggable {
|
||||
String launchEngineOpHandleSecret =
|
||||
openRespConf.get("kyuubi.session.engine.launch.handle.secret");
|
||||
|
||||
launchEngineOpSupportResult =
|
||||
Boolean.parseBoolean(
|
||||
openRespConf.getOrDefault("kyuubi.session.engine.launch.support.result", "false"));
|
||||
|
||||
if (launchEngineOpHandleGuid != null && launchEngineOpHandleSecret != null) {
|
||||
try {
|
||||
byte[] guidBytes = Base64.getMimeDecoder().decode(launchEngineOpHandleGuid);
|
||||
@ -1353,7 +1358,7 @@ public class KyuubiConnection implements SQLConnection, KyuubiLoggable {
|
||||
}
|
||||
|
||||
private void fetchLaunchEngineResult() {
|
||||
if (launchEngineOpHandle == null) return;
|
||||
if (launchEngineOpHandle == null || !launchEngineOpSupportResult) return;
|
||||
|
||||
TFetchResultsReq tFetchResultsReq =
|
||||
new TFetchResultsReq(
|
||||
|
||||
@ -92,6 +92,8 @@ final class KyuubiTBinaryFrontendService(
|
||||
KYUUBI_SESSION_ENGINE_LAUNCH_HANDLE_SECRET,
|
||||
Base64.getMimeEncoder.encodeToString(opHandleIdentifier.getSecret))
|
||||
|
||||
respConfiguration.put(KYUUBI_SESSION_ENGINE_LAUNCH_SUPPORT_RESULT, true.toString)
|
||||
|
||||
resp.setSessionHandle(sessionHandle.toTSessionHandle)
|
||||
resp.setConfiguration(respConfiguration)
|
||||
resp.setStatus(OK_STATUS)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user