[hotfix] stop job without savepoint and drain bug fixed. (apache#821)

lvshaokang · web-flow · commit 16387d5ed130 · 2022-03-15T08:18:02.000+08:00
diff --git a/streamx-plugin/streamx-flink-submit/streamx-flink-submit-core/src/main/scala/com/streamxhub/streamx/flink/submit/impl/RemoteSubmit.scala b/streamx-plugin/streamx-flink-submit/streamx-flink-submit-core/src/main/scala/com/streamxhub/streamx/flink/submit/impl/RemoteSubmit.scala
@@ -82,7 +82,7 @@ object RemoteSubmit extends FlinkSubmitTrait {
         case (true, true) if savePointDir.nonEmpty => client.stopWithSavepoint(jobID, true, savePointDir).get()
         case (true, false) if savePointDir.nonEmpty => client.cancelWithSavepoint(jobID, savePointDir).get()
         case _ => client.cancel(jobID).get()
-          ""
+          null
       }
       StopResponse(actionResult)
     } catch {
diff --git a/streamx-plugin/streamx-flink-submit/streamx-flink-submit-core/src/main/scala/com/streamxhub/streamx/flink/submit/impl/YarnSessionSubmit.scala b/streamx-plugin/streamx-flink-submit/streamx-flink-submit-core/src/main/scala/com/streamxhub/streamx/flink/submit/impl/YarnSessionSubmit.scala
@@ -111,7 +111,7 @@ object YarnSessionSubmit extends YarnSubmitTrait {
         case (true, true) if savePointDir.nonEmpty => client.stopWithSavepoint(jobID, true, savePointDir).get()
         case (true, false) if savePointDir.nonEmpty => client.cancelWithSavepoint(jobID, savePointDir).get()
         case _ => client.cancel(jobID).get()
-          ""
+          null
       }
       StopResponse(actionResult)
     } catch {
diff --git a/streamx-plugin/streamx-flink-submit/streamx-flink-submit-core/src/main/scala/com/streamxhub/streamx/flink/submit/trait/KubernetesNativeSubmitTrait.scala b/streamx-plugin/streamx-flink-submit/streamx-flink-submit-core/src/main/scala/com/streamxhub/streamx/flink/submit/trait/KubernetesNativeSubmitTrait.scala
@@ -115,7 +115,7 @@ trait KubernetesNativeSubmitTrait extends FlinkSubmitTrait {
         case (true, true) if savePointDir.nonEmpty => client.stopWithSavepoint(jobID, true, savePointDir).get()
         case (true, false) if savePointDir.nonEmpty => client.cancelWithSavepoint(jobID, savePointDir).get()
         case _ => client.cancel(jobID).get()
-          ""
+          null
       }
       StopResponse(actionResult)
     } catch {

Original file line number	Diff line number	Diff line change
`@@ -82,7 +82,7 @@ object RemoteSubmit extends FlinkSubmitTrait {`
`82`	`82`	`case (true, true) if savePointDir.nonEmpty => client.stopWithSavepoint(jobID, true, savePointDir).get()`
`83`	`83`	`case (true, false) if savePointDir.nonEmpty => client.cancelWithSavepoint(jobID, savePointDir).get()`
`84`	`84`	`case _ => client.cancel(jobID).get()`
`85`		`- ""`
	`85`	`+ null`
`86`	`86`	`}`
`87`	`87`	`StopResponse(actionResult)`
`88`	`88`	`} catch {`
Original file line number	Diff line number	Diff line change
`@@ -111,7 +111,7 @@ object YarnSessionSubmit extends YarnSubmitTrait {`
`111`	`111`	`case (true, true) if savePointDir.nonEmpty => client.stopWithSavepoint(jobID, true, savePointDir).get()`
`112`	`112`	`case (true, false) if savePointDir.nonEmpty => client.cancelWithSavepoint(jobID, savePointDir).get()`
`113`	`113`	`case _ => client.cancel(jobID).get()`
`114`		`- ""`
	`114`	`+ null`
`115`	`115`	`}`
`116`	`116`	`StopResponse(actionResult)`
`117`	`117`	`} catch {`
Original file line number	Diff line number	Diff line change
`@@ -115,7 +115,7 @@ trait KubernetesNativeSubmitTrait extends FlinkSubmitTrait {`
`115`	`115`	`case (true, true) if savePointDir.nonEmpty => client.stopWithSavepoint(jobID, true, savePointDir).get()`
`116`	`116`	`case (true, false) if savePointDir.nonEmpty => client.cancelWithSavepoint(jobID, savePointDir).get()`
`117`	`117`	`case _ => client.cancel(jobID).get()`
`118`		`- ""`
	`118`	`+ null`
`119`	`119`	`}`
`120`	`120`	`StopResponse(actionResult)`
`121`	`121`	`} catch {`