Skip to content

Commit 8f529aa

Browse files
committed
[KYUUBI #5799] [FLINK] Fix fetch timeout in session conf doesn't support ISO-8601
# 🔍 Description ## Issue References 🔗 Currently, Flink engine supports overwriting result fetch timeout in session conf, but in that way IOS-8601 time format is not supported. This PR fixes the problem. This pull request fixes # ## Describe Your Solution 🔧 Apply ConfigOption's time value conf parsing in session conf parsing. ## Types of changes 🔖 - [x] Bugfix (non-breaking change which fixes an issue) - [ ] New feature (non-breaking change which adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) ## Test Plan 🧪 #### Behavior Without This Pull Request ⚰️ #### Behavior With This Pull Request 🎉 #### Related Unit Tests --- # Checklists ## 📝 Author Self Checklist - [x] My code follows the [style guidelines](https://kyuubi.readthedocs.io/en/master/contributing/code/style.html) of this project - [x] I have performed a self-review - [ ] I have commented my code, particularly in hard-to-understand areas - [ ] I have made corresponding changes to the documentation - [ ] My changes generate no new warnings - [x] I have added tests that prove my fix is effective or that my feature works - [x] New and existing unit tests pass locally with my changes - [x] This patch was not authored or co-authored using [Generative Tooling](https://www.apache.org/legal/generative-tooling.html) ## 📝 Committer Pre-Merge Checklist - [x] Pull request title is okay. - [x] No license issues. - [x] Milestone correctly set? - [x] Test coverage is ok - [x] Assignees are selected. - [x] Minimum number of approvals - [x] No changes are requested **Be nice. Be informative.** Closes #5799 from link3280/timeconf_parsing. Closes #5799 417898a [Paul Lin] [FLINK] Use ISO-8601 time conf in unit test 99a4964 [Paul Lin] [FLINK] Fix fetch timeout in session conf doesn't support ISO-8601 Authored-by: Paul Lin <paullin3280@gmail.com> Signed-off-by: Paul Lin <paullin3280@gmail.com>
1 parent 3677f3a commit 8f529aa

File tree

2 files changed

+5
-2
lines changed

2 files changed

+5
-2
lines changed

externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/operation/FlinkSQLOperationManager.scala

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,10 @@ class FlinkSQLOperationManager extends OperationManager("FlinkSQLOperationManage
7373
resultMaxRowsDefault.toString).toInt
7474

7575
val resultFetchTimeout =
76-
flinkSession.normalizedConf.get(ENGINE_FLINK_FETCH_TIMEOUT.key).map(_.toLong milliseconds)
76+
flinkSession.normalizedConf
77+
.get(ENGINE_FLINK_FETCH_TIMEOUT.key)
78+
.map(ENGINE_FLINK_FETCH_TIMEOUT.valueConverter)
79+
.map(_.get milliseconds)
7780
.getOrElse(resultFetchTimeoutDefault)
7881

7982
val op = mode match {

externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/operation/FlinkOperationSuite.scala

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1255,7 +1255,7 @@ abstract class FlinkOperationSuite extends HiveJDBCTestHelper with WithFlinkTest
12551255

12561256
test("test result fetch timeout") {
12571257
val exception = intercept[KyuubiSQLException](
1258-
withSessionConf()(Map(ENGINE_FLINK_FETCH_TIMEOUT.key -> "60000"))() {
1258+
withSessionConf()(Map(ENGINE_FLINK_FETCH_TIMEOUT.key -> "PT60S"))() {
12591259
withJdbcStatement("tbl_a") { stmt =>
12601260
stmt.executeQuery("create table tbl_a (a int) " +
12611261
"with ('connector' = 'datagen', 'rows-per-second'='0')")

0 commit comments

Comments
 (0)