Zhen Wang

According to the description of your issue, we may be able to modify `BatchJobSubmission#close` like this: ``` override def close(): Unit = withLockRequired { if (!isClosedOrCanceled) { withClosingOperationLog { ```

It is not common to use jdbc to access redis. If the redis jdbc client supports the mysql dialect, I think we can use mysql type jdbc engine to access...

> how can end users get the metrics after adding hint ? Is there a pure sql way to show the metrics ? I would have preferred to display it...

> How about introducing a new syntax instead of hint, something like `CALL OBSERVE('observer1', sum(c1), count(1)) (select * from t1)`, it will return the result of observation This seems to...

> The behavior is decided by yourself. The new syntax also supports observe each operator as long as you inject CollectMetrics for each operator. How do we push down the...

> The difference with hint is that, we can change it's output. I might prefer to split it into two tasks: 1. add a hint to inject observer 2. provide...

> I'm fine with it. It should be a kind of pure sql for `dataset.observe`. Yes, `dataset.observe` also seems to just inject a observer.

> **Is there anything else that needs to be done with this issue? I see there is already a patch** We may need to make changes compatible with spark 3.5...

@zhaohehuhu Could you add a unit test?

I prefer to package it into distribution so we do not need to do extra work, we can clean extensions directory first in dist script.