apache / apache/kyuubi

[Bug] Flink engine should return the result when fetching timeout but already fetched some data

Open
#5,959 0 comments 0 reactions 0 assignees View on GitHub
kind:bug priority:major
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.