gyfora commented on code in PR #994:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/994#discussion_r2218667511
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java:
##########
@@ -143,13 +181,41 @@ public DeleteControl
cleanupInternal(FlinkResourceContext<FlinkDeployment> ctx)
}
return DeleteControl.noFinalizerRemoval()
.rescheduleAfter(ctx.getOperatorConfig().getReconcileInterval().toMillis());
- } else {
- LOG.info("Stopping session cluster");
- var conf = ctx.getDeployConfig(ctx.getResource().getSpec());
- ctx.getFlinkService()
- .deleteClusterDeployment(
- deployment.getMetadata(), deployment.getStatus(),
conf, true);
- return DeleteControl.defaultDelete();
}
+
+ // Check for non-terminated jobs if the option is enabled (Enabled by
default) , after
+ // sessionJobs are deleted
+ boolean blockOnUnmanagedJobs =
+ ctx.getObserveConfig()
+
.getBoolean(KubernetesOperatorConfigOptions.BLOCK_ON_UNMANAGED_JOBS);
+ if (blockOnUnmanagedJobs) {
+ Set<JobID> nonTerminalJobs = getNonTerminalJobs(ctx);
+ if (!nonTerminalJobs.isEmpty()) {
+ var error =
+ String.format(
+ "The session cluster has non terminated jobs
%s that should be cancelled first",
+ nonTerminalJobs.stream()
+ .map(JobID::toHexString)
+ .collect(Collectors.toList()));
+ if (eventRecorder.triggerEvent(
+ deployment,
+ EventRecorder.Type.Warning,
+ EventRecorder.Reason.CleanupFailed,
+ EventRecorder.Component.Operator,
+ error,
+ ctx.getKubernetesClient())) {
+ LOG.warn(error);
+ }
+ return DeleteControl.noFinalizerRemoval()
+
.rescheduleAfter(ctx.getOperatorConfig().getReconcileInterval().toMillis());
+ }
+ }
+
+ LOG.info("Stopping session cluster");
+ var conf = ctx.getDeployConfig(ctx.getResource().getSpec());
+ ctx.getFlinkService()
+ .deleteClusterDeployment(
+ deployment.getMetadata(), deployment.getStatus(),
conf, true);
Review Comment:
I think there is a bug in this existing logic, instead of getting the
deployconfig, here we should just use `ctx.getObserveConfig()`
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java:
##########
@@ -143,13 +181,41 @@ public DeleteControl
cleanupInternal(FlinkResourceContext<FlinkDeployment> ctx)
}
return DeleteControl.noFinalizerRemoval()
.rescheduleAfter(ctx.getOperatorConfig().getReconcileInterval().toMillis());
- } else {
- LOG.info("Stopping session cluster");
- var conf = ctx.getDeployConfig(ctx.getResource().getSpec());
- ctx.getFlinkService()
- .deleteClusterDeployment(
- deployment.getMetadata(), deployment.getStatus(),
conf, true);
- return DeleteControl.defaultDelete();
}
+
+ // Check for non-terminated jobs if the option is enabled (Enabled by
default) , after
+ // sessionJobs are deleted
+ boolean blockOnUnmanagedJobs =
+ ctx.getObserveConfig()
+
.getBoolean(KubernetesOperatorConfigOptions.BLOCK_ON_UNMANAGED_JOBS);
+ if (blockOnUnmanagedJobs) {
+ Set<JobID> nonTerminalJobs = getNonTerminalJobs(ctx);
+ if (!nonTerminalJobs.isEmpty()) {
+ var error =
+ String.format(
+ "The session cluster has non terminated jobs
%s that should be cancelled first",
+ nonTerminalJobs.stream()
+ .map(JobID::toHexString)
+ .collect(Collectors.toList()));
+ if (eventRecorder.triggerEvent(
+ deployment,
+ EventRecorder.Type.Warning,
+ EventRecorder.Reason.CleanupFailed,
+ EventRecorder.Component.Operator,
+ error,
+ ctx.getKubernetesClient())) {
+ LOG.warn(error);
+ }
Review Comment:
we can remove the if branch and the logging. Event triggering already
creates logs we don't need both I think
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
