gyfora commented on code in PR #994:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/994#discussion_r2204019063
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java:
##########
@@ -120,6 +126,61 @@ private void
recoverSession(FlinkResourceContext<FlinkDeployment> ctx) throws Ex
.setJobManagerDeploymentStatus(JobManagerDeploymentStatus.DEPLOYING);
}
+ /**
+ * Detects unmanaged jobs running in the session cluster. Unmanaged jobs
are jobs that exist in
+ * the Flink cluster but are not managed by FlinkSessionJob resources.
+ */
+ private Set<JobID> getUnmanagedJobs(
+ FlinkResourceContext<FlinkDeployment> ctx, Set<FlinkSessionJob>
sessionJobs) {
Review Comment:
I don't think that we need to pass sessionJobs here. if the flag is enabled,
any running job should simply block it. We can simplify this logic a lot
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java:
##########
@@ -143,13 +204,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 unmanaged jobs if the option is enabled (Enabled by
default)
+ boolean blockOnUnmanagedJobs =
+ ctx.getObserveConfig()
+
.getBoolean(KubernetesOperatorConfigOptions.BLOCK_ON_UNMANAGED_JOBS);
+ if (blockOnUnmanagedJobs) {
+ Set<JobID> unmanagedJobs = getUnmanagedJobs(ctx, sessionJobs);
+ if (!unmanagedJobs.isEmpty()) {
+ var error =
+ String.format(
+ "The session cluster has unmanaged jobs %s
that should be cancelled first",
+ unmanagedJobs.stream()
+ .map(JobID::toHexString)
+ .collect(Collectors.toList()));
+ LOG.warn(error);
+ if (eventRecorder.triggerEvent(
+ deployment,
+ EventRecorder.Type.Warning,
+ EventRecorder.Reason.CleanupFailed,
+ EventRecorder.Component.Operator,
+ error,
+ ctx.getKubernetesClient())) {
+ LOG.warn(error);
Review Comment:
you are logging the error twice, you don't need to log it at all as the
event triggering already logs it.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java:
##########
@@ -120,6 +126,61 @@ private void
recoverSession(FlinkResourceContext<FlinkDeployment> ctx) throws Ex
.setJobManagerDeploymentStatus(JobManagerDeploymentStatus.DEPLOYING);
}
+ /**
+ * Detects unmanaged jobs running in the session cluster. Unmanaged jobs
are jobs that exist in
+ * the Flink cluster but are not managed by FlinkSessionJob resources.
+ */
+ private Set<JobID> getUnmanagedJobs(
+ FlinkResourceContext<FlinkDeployment> ctx, Set<FlinkSessionJob>
sessionJobs) {
+ LOG.info(
+ "Starting unmanaged job detection for session cluster: {}",
+ ctx.getResource().getMetadata().getName());
+ try {
+ // Get all jobs running in the Flink cluster
+ var flinkService = ctx.getFlinkService();
+ var clusterClient =
flinkService.getClusterClient(ctx.getObserveConfig());
+ var allJobs =
+ clusterClient
+ .sendRequest(
+ JobsOverviewHeaders.getInstance(),
+ EmptyMessageParameters.getInstance(),
+ EmptyRequestBody.getInstance())
+ .get()
+ .getJobs();
+
+ // Get job IDs managed by FlinkSessionJob resources
+ Set<JobID> managedJobIds =
+ sessionJobs.stream()
+ .map(job ->
job.getStatus().getJobStatus().getJobId())
+ .filter(jobId -> jobId != null)
+ .map(JobID::fromHexString)
+ .collect(Collectors.toSet());
+
+ // Get job IDs that are not managed by FlinkSessionJob resources
and are currently
+ // running
+ // FAILED(TerminalState.GLOBALLY)
+ // CANCELED(TerminalState.GLOBALLY)
+ // FINISHED(TerminalState.GLOBALLY)
+ // Above terminal states are not considered since they are no
longer active jobs
Review Comment:
This comment feels a bit too verbose, enough to say only consider non
terminal jobs (feels a bit like AI generated)
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java:
##########
@@ -120,6 +126,61 @@ private void
recoverSession(FlinkResourceContext<FlinkDeployment> ctx) throws Ex
.setJobManagerDeploymentStatus(JobManagerDeploymentStatus.DEPLOYING);
}
+ /**
+ * Detects unmanaged jobs running in the session cluster. Unmanaged jobs
are jobs that exist in
+ * the Flink cluster but are not managed by FlinkSessionJob resources.
+ */
+ private Set<JobID> getUnmanagedJobs(
+ FlinkResourceContext<FlinkDeployment> ctx, Set<FlinkSessionJob>
sessionJobs) {
+ LOG.info(
+ "Starting unmanaged job detection for session cluster: {}",
+ ctx.getResource().getMetadata().getName());
Review Comment:
this should be on debug level, also no need to include resource name/info in
the log message. Its in the MDC already
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java:
##########
@@ -120,6 +126,61 @@ private void
recoverSession(FlinkResourceContext<FlinkDeployment> ctx) throws Ex
.setJobManagerDeploymentStatus(JobManagerDeploymentStatus.DEPLOYING);
}
+ /**
+ * Detects unmanaged jobs running in the session cluster. Unmanaged jobs
are jobs that exist in
+ * the Flink cluster but are not managed by FlinkSessionJob resources.
+ */
+ private Set<JobID> getUnmanagedJobs(
+ FlinkResourceContext<FlinkDeployment> ctx, Set<FlinkSessionJob>
sessionJobs) {
Review Comment:
We can simply replace this method with something like getNonTerminalJobIds()
or boolean anyNonTerminalJobs()
That would be enough for this feature.
##########
flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconcilerTest.java:
##########
@@ -139,4 +149,164 @@ public void testSetOwnerReference() throws Exception {
deployConfig.get(KubernetesConfigOptions.JOB_MANAGER_OWNER_REFERENCE);
Assertions.assertEquals(expectedOwnerReferences, or);
}
+
+ @Test
+ public void testGetUnmanagedJobs() throws Exception {
+ // Create a session cluster deployment with blocking enabled
+ FlinkDeployment deployment = TestUtils.buildSessionCluster();
+ deployment
+ .getSpec()
+ .getFlinkConfiguration()
+
.put(KubernetesOperatorConfigOptions.BLOCK_ON_UNMANAGED_JOBS.key(), "true");
+
+ assertEquals(
+ "true",
+ deployment
+ .getSpec()
+ .getFlinkConfiguration()
+
.get(KubernetesOperatorConfigOptions.BLOCK_ON_UNMANAGED_JOBS.key()));
+
+ // First, deploy the session cluster to set up proper state
+ reconciler.reconcile(deployment, flinkService.getContext());
+
+ // Verify deployment is in DEPLOYED state
+ assertEquals(
+ ReconciliationState.DEPLOYED,
+ deployment.getStatus().getReconciliationStatus().getState());
+
+ // Create different types of jobs
+ JobID managedJobId1 = new JobID();
+ JobID managedJobId2 = new JobID();
+ JobID unmanagedRunningJobId1 = new JobID();
+ JobID unmanagedTerminatedJobId = new JobID();
+ JobID unmanagedRunningJobId2 = new JobID();
+
+ // Add a mix of managed and unmanaged jobs to the testing service
+ flinkService
+ .listJobs()
+ .add(
+ Tuple3.of(
+ null,
+ new JobStatusMessage(
+ managedJobId1,
+ "managed-job-1",
+ JobStatus.RUNNING,
+ System.currentTimeMillis()),
+ new Configuration()));
+ flinkService
+ .listJobs()
+ .add(
+ Tuple3.of(
+ null,
+ new JobStatusMessage(
+ managedJobId2,
+ "managed-job-2",
+ JobStatus.RUNNING,
+ System.currentTimeMillis()),
+ new Configuration()));
+ flinkService
+ .listJobs()
+ .add(
+ Tuple3.of(
+ null,
+ new JobStatusMessage(
+ unmanagedRunningJobId1,
+ "unmanaged-running-job-1",
+ JobStatus.RUNNING,
+ System.currentTimeMillis()),
+ new Configuration()));
+ flinkService
+ .listJobs()
+ .add(
+ Tuple3.of(
+ null,
+ new JobStatusMessage(
+ unmanagedTerminatedJobId,
+ "unmanaged-terminated-job",
+ JobStatus.CANCELED,
+ System.currentTimeMillis()),
+ new Configuration()));
+ flinkService
+ .listJobs()
+ .add(
+ Tuple3.of(
+ null,
+ new JobStatusMessage(
+ unmanagedRunningJobId2,
+ "unmanaged-running-job-2",
+ JobStatus.RUNNING,
+ System.currentTimeMillis()),
+ new Configuration()));
+
+ // Create FlinkSessionJob resources for the managed jobs
+ FlinkSessionJob managedSessionJob1 = TestUtils.buildSessionJob();
+ managedSessionJob1.getMetadata().setName("managed-session-job-1");
+
managedSessionJob1.getStatus().getJobStatus().setJobId(managedJobId1.toHexString());
+ kubernetesClient.resource(managedSessionJob1).createOrReplace();
+
+ FlinkSessionJob managedSessionJob2 = TestUtils.buildSessionJob();
+ managedSessionJob2.getMetadata().setName("managed-session-job-2");
+
managedSessionJob2.getStatus().getJobStatus().setJobId(managedJobId2.toHexString());
+ kubernetesClient.resource(managedSessionJob2).createOrReplace();
+
+ Set<FlinkSessionJob> sessionJobs = new HashSet<>();
+ sessionJobs.add(managedSessionJob1);
+ sessionJobs.add(managedSessionJob2);
+
+ // Test with blocking enabled - should identify running unmanaged jobs
correctly
+ var context =
TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient);
+ var resourceContext = getResourceContext(deployment, context);
+
+ // Use reflection to access the private getUnmanagedJobs method
Review Comment:
We should not use reflection for this, we can make the method protected with
the @visiblefor testing annotation
--
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]
