[KYUUBI #4806][FLINK] Support time-out incremental result fetch for Flink engine

[link3280]

Why are the changes needed?

As titled.

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

• Run test locally before make a pull request

[codecov-commenter]

Codecov Report

Merging #5134 (1563fa9) into master (14d0dab) will not change coverage.
Report is 20 commits behind head on master.
The diff coverage is 0.00%.

❗ Current head 1563fa9 differs from pull request most recent head a1b7478. Consider uploading reports for the commit a1b7478 to get more accurate results

@@          Coverage Diff           @@
##           master   #5134   +/-   ##
======================================
  Coverage    0.00%   0.00%           
======================================
  Files         566     566           
  Lines       31654   31703   +49     
  Branches     4124    4134   +10     
======================================
- Misses      31654   31703   +49     

Files Changed	Coverage Δ
...in/scala/org/apache/kyuubi/config/KyuubiConf.scala	0.00% <0.00%> (ø)

... and 7 files with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

[link3280]

@pan3793 @bowenliang123 PTAL.

[bowenliang123]

If no data are fetched, a TimeoutException would be thrown.
I would prefer not to throw a TimeoutException and return an empty result set instead.
An empty result for time-out fetching is expected if the timeout config is set.

[link3280]

We had a related discussion here: #4806 (comment)

[bowenliang123]

Well, it's acceptable, while the discussion was treated as a hint for internal behaviour than for a feature for me at the time.
Let's see what others think.

[link3280]

Any further comments? @pan3793

[pan3793]

@link3280 sorry for the late response, just back to the office, will finish review today.

[link3280]

@pan3793 No worries. Please take a look whenever it works for you.

[pan3793]

LGTM, thanks!

[pan3793]

what's the style for other configurations, use 0 or None to represent inf? let's algin the style

[link3280]

I found kyuubi.operation.query.timeout uses None.

kyuubi/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala

Lines 1851 to 1863 in 9e3ac23

	val OPERATION_QUERY_TIMEOUT: OptionalConfigEntry[Long] =
	buildConf("kyuubi.operation.query.timeout")
	.doc("Timeout for query executions at server-side, take effect with client-side timeout(" +
	"`java.sql.Statement.setQueryTimeout`) together, a running query will be cancelled" +
	" automatically if timeout. It's off by default, which means only client-side take full" +
	" control of whether the query should timeout or not." +
	" If set, client-side timeout is capped at this point." +
	" To cancel the queries right away without waiting for task to finish," +
	s" consider enabling ${OPERATION_FORCE_CANCEL.key} together.")
	.version("1.2.0")
	.timeConf
	.checkValue(_ >= 1000, "must >= 1s if set")
	.createOptional

Plus, I think 0 sometimes denotes an immediate timeout instead of an infinite timeout, thus None is a better choice.

[link3280]

Merged. Thanks a lot!

[bowenliang123]

Thanks for the efforts!