[Bug]Fix FlinkJobStatusWatcher deadlock & NullPointerException (#4327)

* Fix FlinkJobStatusWatcher deadlock

* Fix the issue where completed task status is not being updated

---------

Co-authored-by: liguoliang1 <liguoliang1@hisense.com>

Author: Li-GL

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 _ =>

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)