mielientiev commented on code in PR #1065:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/1065#discussion_r3120082868
##########
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 checked this and for ignore/non-upgrade changes (labels/operator configs)
I see that observedGeneration is updated as well.
--
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]
