[Improvement-17330][K8s] Replace job watcher with informer

[det101]

Was this PR generated or assisted by AI?

YES

Purpose of the pull request

fix #17330

Brief change log

Verify this pull request

This change added tests and can be verified as follows:

./mvnw -pl dolphinscheduler-task-plugin/dolphinscheduler-task-api clean test -Dtest=K8sTaskExecutorTest

• [√ ] Manual K8s smoke test with busybox short task (sleep 10) succeeds end-to-end
• [√] long task (sleep 2400) does not fail with too old resource version ([Improvement][K8s] too old resource version #17330)

Pull Request Notice

Pull Request Notice

If your pull request contains incompatible change, you should also add it to docs/docs/en/guide/upgrade/incompatible.md

[sonarqubecloud]

Quality Gate failed

Failed conditions
0.0% Coverage on New Code (required ≥ 60%)

See analysis details on SonarQube Cloud

[det101]

Upgrading Fabric8 from version 6.4 to 6.0: The BOM upgrade was found to affect all Kubernetes clients, resulting in a significant impact. Modifications were made, but still based on version 6.0. @SbloodyS @ruanwenjun

[det101]

Manual verification (minikube)
Tested on standalone + minikube with busybox:1.30.1, two sequential K8S tasks.

Short task (sleep 15): Job submitted; informer logged event received, job: ..., action: ADD/UPDATE; terminal status 0 → succeed in k8s. Task SUCCESS.

Long task (sleep 2400, ~40 min): Informer kept receiving ADD/UPDATE for the full run; pod finished after 40 min; status 0 → succeed in k8s. Workflow SUCCESS. No too old resource version or fail in k8s.

[SbloodyS]

This still does not fully replace the old Watcher.onClose failure path. In Fabric8 6.0, SharedIndexInformer.start() completes from Reflector.listSyncAndWatch(), but Reflector does not compose the watch future returned by startWatcher(); later non-HttpGone watch closures only set running=false and do not complete this start future exceptionally. For non-timeout tasks, awaitJobCompletion() can therefore block forever instead of failing the task as the old onClose(WatcherException) did. We need an explicit monitor/failure path for informer/watch stopping, or another way to count down the latch when the informer can no longer observe the Job.

[det101]

This still does not fully replace the old Watcher.onClose failure path. In Fabric8 6.0, SharedIndexInformer.start() completes from Reflector.listSyncAndWatch(), but Reflector does not compose the watch future returned by startWatcher(); later non-HttpGone watch closures only set running=false and do not complete this start future exceptionally. For non-timeout tasks, awaitJobCompletion() can therefore block forever instead of failing the task as the old onClose(WatcherException) did. We need an explicit monitor/failure path for informer/watch stopping, or another way to count down the latch when the informer can no longer observe the Job.

Hi, to address the concern that awaitJobCompletion() may block forever when the informer stops observing the Job in Fabric8 6.0, my approach is:

Primary: SharedIndexInformer handles ADD/UPDATE/DELETE.
Safety net: poll Job status via GET every 30s; count down the latch on terminal state or Job deletion if informer events are missed.
We intentionally do not fail on isWatching()==false to avoid false failures during relist gaps or while the Job is still running. Task timeout remains the final fallback.

Does this approach work for you?

[SbloodyS]

I agree that polling the Job status via GET is a useful safety net when informer events are missed and the Kubernetes API is still reachable.

However, I think this still does not fully cover the old Watcher.onClose failure path. In the current implementation, poll failures are only logged and do not count down the latch. Also, task timeout is only a fallback when the timeout strategy is FAILED or WARNFAILED; for tasks without those timeout strategies, awaitJobCompletion() can still block indefinitely if the informer stops delivering events and GET keeps failing.

Could we add a bounded failure policy for continuous polling errors, or another explicit fatal/stopped informer path, so the task can fail instead of waiting forever? A unit test for “informer started, no terminal event, GET keeps failing, no timeout strategy” would also help cover this case.

[det101]

Good catch — you're right that logging poll errors alone doesn't cover the old Watcher.onClose fatal path, and without a FAILED/WARNFAILED timeout strategy awaitJobCompletion() could block indefinitely.
I've added a bounded failure policy: after 3 consecutive GET poll failures (30s interval, ~90s total), the task fails and counts down the latch. A successful GET resets the counter, so transient errors don't immediately fail the task. This mirrors the intent of the old watcher close path when the API stays unreachable.

[ruanwenjun]

Please don't put all the implementation into a single class, as it makes the code harder to maintain. Consider introducing a dedicated K8sJobMonitor class to monitor the job status after submission, which would help keep the responsibilities better separated.