[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.

[SbloodyS]

I found two correctness issues in the current implementation:

1. [P1] Treat a missing JobStatus as a running state

File: dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/k8s/impl/K8sJobMonitor.java
Around lines 235–240

Kubernetes may legitimately return a newly created Job whose status is still null. This is especially relevant now because onAdd evaluates the Job immediately.

getK8sJobStatus() currently dereferences job.getStatus() without a null check. The ADD handler will therefore throw a NullPointerException. The polling path has a more serious consequence: the same valid response is counted as a polling failure, and three responses with a null status will incorrectly fail the task even though the API requests succeeded and the Job may simply still be pending.

Please treat a null Job or JobStatus as RUNNING_CODE and add a test using a Job without a status object.

2. [P2] Seal the terminal result when timeout or monitor setup fails

File: dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/k8s/impl/K8sJobMonitor.java
Around lines 95–101

The exception handlers write EXIT_CODE_FAILURE directly but leave completed false. Consequently, after a timeout or another monitoring exception, an informer callback or an already-running polling request can still call completeOnce(...) and overwrite the failure with success.

This race is particularly possible at the timeout boundary: await() times out, the catch block records failure, and a concurrently queued terminal UPDATE then records success because it can still change completed from false to true.

Please route these failure paths through the same atomic completion mechanism, for example by calling completeOnce(completed, countDownLatch, taskResponse, EXIT_CODE_FAILURE, jobName). A timeout-versus-success concurrency test would also help prevent regressions.