This is an automated email from the ASF dual-hosted git repository.
benjobs pushed a commit to branch dev-2.1.7
in repository https://gitbox.apache.org/repos/asf/streampark.git
The following commit(s) were added to refs/heads/dev-2.1.7 by this push:
new 16bfa389d [Bug]Fix FlinkJobStatusWatcher deadlock &
NullPointerException (#4327)
16bfa389d is described below
commit 16bfa389d4dde6b09a2b08ce5bb7331c385a8f9b
Author: Lane Li <[email protected]>
AuthorDate: Mon Jan 26 10:35:28 2026 +0800
[Bug]Fix FlinkJobStatusWatcher deadlock & NullPointerException (#4327)
* Fix FlinkJobStatusWatcher deadlock
* Fix the issue where completed task status is not being updated
---------
Co-authored-by: liguoliang1 <[email protected]>
---
.../streampark/flink/kubernetes/watcher/FlinkJobStatusWatcher.scala | 4 ++--
.../org/apache/streampark/flink/kubernetes/watcher/FlinkWatcher.scala | 4 ++--
2 files changed, 4 insertions(+), 4 deletions(-)
diff --git
a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkJobStatusWatcher.scala
b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkJobStatusWatcher.scala
index 39114d197..0569dac64 100644
---
a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkJobStatusWatcher.scala
+++
b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkJobStatusWatcher.scala
@@ -314,8 +314,8 @@ class FlinkJobStatusWatcher(conf: JobStatusWatcherConfig =
JobStatusWatcherConfi
)
val jobState = trackId match {
case id
- if watchController.canceling.has(id) || latest.jobState.equals(
- FlinkJobState.CANCELLING) =>
+ if watchController.canceling.has(id) || Option(latest).exists(
+ _.jobState == FlinkJobState.CANCELLING) =>
logger.info(s"trackId ${trackId.toString} is canceling")
if (deployExists) FlinkJobState.CANCELLING else FlinkJobState.CANCELED
case _ =>
diff --git
a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkWatcher.scala
b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkWatcher.scala
index e6f836af5..1a1d2a304 100644
---
a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkWatcher.scala
+++
b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkWatcher.scala
@@ -28,11 +28,11 @@ import scala.language.implicitConversions
trait FlinkWatcher extends AutoCloseable {
// see org.apache.flink.client.cli.ClientOptions.CLIENT_TIMEOUT}
- lazy val FLINK_CLIENT_TIMEOUT_SEC: Timeout =
+ val FLINK_CLIENT_TIMEOUT_SEC: Timeout =
Timeout.ofMilliseconds(Duration.ofSeconds(60).toMillis).toTimeout
// see org.apache.flink.configuration.RestOptions.AWAIT_LEADER_TIMEOUT
- lazy val FLINK_REST_AWAIT_TIMEOUT_SEC: Timeout =
Timeout.ofMilliseconds(30000L)
+ val FLINK_REST_AWAIT_TIMEOUT_SEC: Timeout = Timeout.ofMilliseconds(30000L)
private[this] val started: AtomicBoolean = new AtomicBoolean(false)
