[Bug] Flink engine should return the result when fetching timeout but already fetched some data
- Dominant language
- Scala
- Stars
- 2.4k
- Forks
- 1k
- PR merge metrics
- No merged PRs in 30d
Description
### Code of Conduct
- [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)
### Search before asking
- [X] I have searched in the [issues](https://github.com/apache/kyuubi/issues?q=is%3Aissue) and found no similar issues.
### Describe the bug
There are already 3 rows in Kafka when we query from the earlist offset,
I tested some cases:
1. `kyuubi.session.engine.flink.max.rows=3` & `kyuubi.session.engine.flink.fetch.timeout=PT60S`
client return 3 rows,this meet our expect
2. `kyuubi.session.engine.flink.max.rows=5` & `kyuubi.session.engine.flink.fetch.timeout=PT60S`
client returns the Futures timed out after [60000 milliseconds],this does not meet our expect.
we shoud return the result when fetch timeout but already fetched some data
### Affects Version(s)
1.8.0
### Kyuubi Server Log Output
_No response_
### Kyuubi Engine Log Output
```logtalk
first time fetch req;
2024-01-10 17:13:54,674 DEBUG org.apache.kyuubi.engine.flink.FlinkTBinaryFrontendService [] - TFetchResultsReq(operationHandle:TOperationHandle(operationId:THandleIdentifier(guid:58 0C 2C EC F9 70 4D 54 8F 44 8F 0C 97 57 34 90, secret:C2 EE 5B 97 3E A0 41 FC AC 16 9B D7 08 ED 8F 38), operationType:EXECUTE_STATEMENT, hasResultSet:true), orientation:FETCH_NEXT, maxRows:1000, fetchType:0)
first time fetch resp:
2024-01-10 17:15:18,748 DEBUG org.apache.kyuubi.engine.flink.operation.ExecuteStatement [] - Timeout fetching more data for ExecuteStatement operation. Returning the current fetched data.
second fetch req:
2024-01-10 17:15:18,836 DEBUG org.apache.kyuubi.engine.flink.FlinkTBinaryFrontendService [] - TFetchResultsReq(operationHandle:TOperationHandle(operationId:THandleIdentifier(guid:58 0C 2C EC F9 70 4D 54 8F 44 8F 0C 97 57 34 90, secret:C2 EE 5B 97 3E A0 41 FC AC 16 9B D7 08 ED 8F 38), operationType:EXECUTE_STATEMENT, hasResultSet:true), orientation:FETCH_NEXT, maxRows:1000, fetchType:0)
second fetch resp:
2024-01-10 17:16:18,839 ERROR org.apache.kyuubi.engine.flink.FlinkTBinaryFrontendService [] - Error fetching results:
java.util.concurrent.TimeoutException: Futures timed out after [60000 milliseconds]
at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:255) ~[flink-scala_2.12-1.18.0-vip-SNAPSHOT.jar:1.18.0-vip-SNAPSHOT]
at scala.concurrent.impl.Promise$DefaultPromise.result(Promise.scala:259) ~[flink-scala_2.12-1.18.0-vip-SNAPSHOT.jar:1.18.0-vip-SNAPSHOT]
at scala.concurrent.Await$.$anonfun$result$1(package.scala:215) ~[flink-scala_2.12-1.18.0-vip-SNAPSHOT.jar:1.18.0-vip-SNAPSHOT]
at scala.concurrent.BlockContext$DefaultBlockContext$.blockOn(BlockContext.scala:53) ~[flink-scala_2.12-1.18.0-vip-SNAPSHOT.jar:1.18.0-vip-SNAPSHOT]
at scala.concurrent.Await$.result(package.scala:142) ~[flink-scala_2.12-1.18.0-vip-SNAPSHOT.jar:1.18.0-vip-SNAPSHOT]
at org.apache.kyuubi.engine.flink.result.IncrementalResultFetchIterator.fetchNext(IncrementalResultFetchIterator.scala:128) ~[kyuubi-flink-sql-engine_2.12-1.9.0-SNAPSHOT.jar:1.9.0-SNAPSHOT]
at org.apache.kyuubi.engine.flink.result.IncrementalResultFetchIterator.next(IncrementalResultFetchIterator.scala:164) ~[kyuubi-flink-sql-engine_2.12-1.9.0-SNAPSHOT.jar:1.9.0-SNAPSHOT]
at org.apache.kyuubi.engine.flink.result.IncrementalResultFetchIterator.next(IncrementalResultFetchIterator.scala:42) ~[kyuubi-flink-sql-engine_2.12-1.9.0-SNAPSHOT.jar:1.9.0-SNAPSHOT]
at org.apache.kyuubi.engine.flink.operation.FlinkOperation.getNextRowSetInternal(FlinkOperation.scala:118) ~[kyuubi-flink-sql-engine_2.12-1.9.0-SNAPSHOT.jar:1.9.0-SNAPSHOT]
at org.apache.kyuubi.operation.AbstractOperation.$anonfun$getNextRowSet$1(AbstractOperation.scala:191) ~[kyuubi-flink-sql-engine_2.12-1.9.0-SNAPSHOT.jar:1.9.0-SNAPSHOT]
at org.apache.kyuubi.Utils$.withLockRequired(Utils.scala:425) ~[kyuubi-flink-sql-engine_2.12-1.9.0-SNAPSHOT.jar:1.9.0-SNAPSHOT]
at org.apache.kyuubi.operation.AbstractOperation.withLockRequired(AbstractOperation.scala:52) ~[kyuubi-flink-sql-engine_2.12-1.9.0-SNAPSHOT.jar:1.9.0-SNAPSHOT]
at org.apache.kyuubi.operation.AbstractOperation.getNextRowSet(AbstractOperation.scala:191) ~[kyuubi-flink-sql-engine_2.12-1.9.0-SNAPSHOT.jar:1.9.0-SNAPSHOT]
at org.apache.kyuubi.operation.OperationManager.getOperationNextRowSet(OperationManager.scala:140) ~[kyuubi-flink-sql-engine_2.12-1.9.0-SNAPSHOT.jar:1.9.0-SNAPSHOT]
at org.apache.kyuubi.session.AbstractSession.fetchResults(AbstractSession.scala:239) ~[kyuubi-flink-sql-engine_2.12-1.9.0-SNAPSHOT.jar:1.9.0-SNAPSHOT]
at org.apache.kyuubi.service.AbstractBackendService.fetchResults(AbstractBackendService.scala:213) ~[kyuubi-flink-sql-engine_2.12-1.9.0-SNAPSHOT.jar:1.9.0-SNAPSHOT]
at org.apache.kyuubi.service.TFrontendService.FetchResults(TFrontendService.scala:530) [kyuubi-flink-sql-engine_2.12-1.9.0-SNAPSHOT.jar:1.9.0-SNAPSHOT]
at org.apache.kyuubi.shaded.hive.service.rpc.thrift.TCLIService$Processor$FetchResults.getResult(TCLIService.java:1837) [kyuubi-flink-sql-engine_2.12-1.9.0-SNAPSHOT.jar:1.9.0-SNAPSHOT]
at org.apache.kyuubi.shaded.hive.service.rpc.thrift.TCLIService$Processor$FetchResults.getResult(TCLIService.java:1822) [kyuubi-flink-sql-engine_2.12-1.9.0-SNAPSHOT.jar:1.9.0-SNAPSHOT]
at org.apache.kyuubi.shaded.thrift.ProcessFunction.process(ProcessFunction.java:39) [kyuubi-flink-sql-engine_2.12-1.9.0-SNAPSHOT.jar:1.9.0-SNAPSHOT]
at org.apache.kyuubi.shaded.thrift.TBaseProcessor.process(TBaseProcessor.java:39) [kyuubi-flink-sql-engine_2.12-1.9.0-SNAPSHOT.jar:1.9.0-SNAPSHOT]
at org.apache.kyuubi.service.authentication.TSetIpAddressProcessor.process(TSetIpAddressProcessor.scala:35) [kyuubi-flink-sql-engine_2.12-1.9.0-SNAPSHOT.jar:1.9.0-SNAPSHOT]
at org.apache.kyuubi.shaded.thrift.server.TThreadPoolServer$WorkerProcess.run(TThreadPoolServer.java:286) [kyuubi-flink-sql-engine_2.12-1.9.0-SNAPSHOT.jar:1.9.0-SNAPSHOT]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_201]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_201]
at java.lang.Thread.run(Thread.java:748) [?:1.8.0_201]
```
### Kyuubi Server Configurations
_No response_
### Kyuubi Engine Configurations
_No response_
### Additional context
$KYUUBI_HOME/bin/beeline -u 'jdbc:hive2://X.X.X.X:10009/#kyuubi.engine.type=FLINK_SQL;kyuubi.engine.type=FLINK_SQL;kyuubi.session.engine.flink.max.rows=100;kyuubi.session.engine.flink.fetch.timeout=PT60S' -n u_test
SQL:
select * from hive.test_db.test_tb
/*+ OPTIONS('properties.bootstrap.servers'='....',
'topic' = '....',
'properties.group.id'='test_kyuubi',
'scan.startup.mode' = 'earlist-offset',
'connector' = 'kafka'
)*/
limit 10;
### Are you willing to submit PR?
- [X] Yes. I would be willing to submit a PR with guidance from the Kyuubi community to fix.
- [ ] No. I cannot submit a PR at this time.
Contributor guide
Research direction
Start by reading IncrementalResultFetchIterator.scala around line 128 and FlinkOperation.scala around line 118, then trace the FETCH_RESULTS entry point shown in the stack trace. Reproduce the Kafka query with max.rows greater than the available rows and the configured fetch timeout. Done means a fetch that times out after receiving rows returns those rows instead of propagating a TimeoutException.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- kafka, scala, sql
- Domain
- api, backend, stream-processing
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100