gyfora commented on code in PR #1065:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/1065#discussion_r3109096040
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconciler.java:
##########
@@ -178,20 +181,59 @@ public DeleteControl
cleanupInternal(FlinkResourceContext<FlinkSessionJob> ctx)
}
public static boolean sessionClusterReady(Optional<FlinkDeployment>
flinkDeploymentOpt) {
- if (flinkDeploymentOpt.isPresent()) {
- var flinkdep = flinkDeploymentOpt.get();
- var jobmanagerDeploymentStatus =
flinkdep.getStatus().getJobManagerDeploymentStatus();
- if (jobmanagerDeploymentStatus !=
JobManagerDeploymentStatus.READY) {
- LOG.info(
- "Session cluster deployment is in {} status, not ready
for serve",
- jobmanagerDeploymentStatus);
- return false;
- } else {
- return true;
- }
- } else {
+ if (flinkDeploymentOpt.isEmpty()) {
LOG.warn("Session cluster deployment is not found");
return false;
}
+ var flinkdep = flinkDeploymentOpt.get();
+ var jobmanagerDeploymentStatus =
flinkdep.getStatus().getJobManagerDeploymentStatus();
+ if (jobmanagerDeploymentStatus != JobManagerDeploymentStatus.READY) {
+ LOG.info(
+ "Session cluster deployment is in {} status, not ready for
serve",
+ jobmanagerDeploymentStatus);
+ return false;
+ }
+
+ // Block while FlinkDeployment is in a transitional reconciliation
state
+ var reconciliationState =
flinkdep.getStatus().getReconciliationStatus().getState();
+ if (reconciliationState != ReconciliationState.DEPLOYED
+ && reconciliationState != ReconciliationState.ROLLED_BACK) {
+ LOG.info(
+ "Session cluster deployment reconciliation state is {},
not ready for serve",
+ reconciliationState);
+ return false;
+ }
+
+ return true;
+ }
+
+ /**
+ * Checks that the FlinkDeployment has no pending spec changes
+ *
+ * <p>When they differ, the deployment spec was changed but the operator
hasn't acted yet. The
+ * cluster may still appear healthy (JM READY, state DEPLOYED), but it is
about to be updated
+ * (e.g. deleted and recreated). Allowing a session job to start upgrading
in this window would
+ * risk the savepoint being destroyed during the cluster rebuild.
+ *
+ * <p>This check is only used in {@link #readyToReconcile}, not in {@link
#sessionClusterReady},
+ * so it does not block cleanup or FlinkService creation.
+ */
+ private static boolean noDeploymentChangesPending(
+ Optional<FlinkDeployment> flinkDeploymentOpt) {
+ if (flinkDeploymentOpt.isEmpty()) {
+ return false;
+ }
+ var flinkdep = flinkDeploymentOpt.get();
+ if (!Objects.equals(
+ flinkdep.getMetadata().getGeneration(),
+ flinkdep.getStatus().getObservedGeneration())) {
Review Comment:
I think it would be worth double checking the the observedGeneration is
always correctly updated (even for ignore/non-upgrade changes such as
labels/operator configs etc). Otherwise we could easily block session job
deletion
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java:
##########
@@ -64,9 +66,38 @@ public SessionReconciler(
@Override
protected boolean readyToReconcile(FlinkResourceContext<FlinkDeployment>
ctx) {
+ var deployment = ctx.getResource();
+ // Only check when there is a pending spec change on the deployment
itself
+ if (!Objects.equals(
+ deployment.getMetadata().getGeneration(),
+ deployment.getStatus().getObservedGeneration())) {
+ var deploymentName = deployment.getMetadata().getName();
+ boolean hasTransitioningSessionJob =
+
ctx.getJosdkContext().getSecondaryResources(FlinkSessionJob.class).stream()
+ .filter(job ->
deploymentName.equals(job.getSpec().getDeploymentName()))
+
.anyMatch(SessionReconciler::isSessionJobTransitioning);
+ if (hasTransitioningSessionJob) {
+ LOG.info(
+ "Deferring session cluster upgrade: associated session
jobs are being upgraded");
+ return false;
+ }
+ }
return true;
}
+ private static boolean isSessionJobTransitioning(FlinkSessionJob job) {
+ var reconStatus = job.getStatus().getReconciliationStatus();
+ // New session jobs state is UPGRADING by default before first
deployment. Treating them as
+ // transitioning would deadlock: the deployment can't
+ // start because it waits for the job, and the job can't start because
it waits for the
+ // cluster.
+ if (reconStatus.isBeforeFirstDeployment()) {
+ return false;
+ }
+ var state = reconStatus.getState();
+ return state == ReconciliationState.UPGRADING || state ==
ReconciliationState.ROLLING_BACK;
Review Comment:
Is there any case where a session job may get stuck in
upgrading/rolling_back state? This check is quite tricky as it may cause a
"deadlock" situation with the session upgrade.
--
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]
