Review comments

Author: gyfora

docs/layouts/shortcodes/generated/dynamic_section.html

@@ -126,7 +126,7 @@
             <td><h5>kubernetes.operator.job.upgrade.last-state.job-cancel.enabled</h5></td>
             <td style="word-wrap: break-word;">false</td>
             <td>Boolean</td>
-            <td>Cancel jobs during last-state upgrade.</td>
+            <td>Cancel jobs during last-state upgrade. This config is ignored for session jobs where cancel is the only mechanism to perform this type of upgrade.</td>
         </tr>
         <tr>
             <td><h5>kubernetes.operator.job.upgrade.last-state.max.allowed.checkpoint.age</h5></td>

docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html

@@ -216,7 +216,7 @@
             <td><h5>kubernetes.operator.job.upgrade.last-state.job-cancel.enabled</h5></td>
             <td style="word-wrap: break-word;">false</td>
             <td>Boolean</td>
-            <td>Cancel jobs during last-state upgrade.</td>
+            <td>Cancel jobs during last-state upgrade. This config is ignored for session jobs where cancel is the only mechanism to perform this type of upgrade.</td>
         </tr>
         <tr>
             <td><h5>kubernetes.operator.job.upgrade.last-state.max.allowed.checkpoint.age</h5></td>

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java

@@ -470,7 +470,8 @@ public static String operatorConfigKey(String key) {
             operatorConfig("job.upgrade.last-state.job-cancel.enabled")
                     .booleanType()
                     .defaultValue(false)
-                    .withDescription("Cancel jobs during last-state upgrade.");
+                    .withDescription(
+                            "Cancel jobs during last-state upgrade. This config is ignored for session jobs where cancel is the only mechanism to perform this type of upgrade.");
 
     @Documentation.Section(SECTION_ADVANCED)
     public static final ConfigOption<Boolean> OPERATOR_HEALTH_PROBE_ENABLED =

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java

@@ -122,7 +122,7 @@ protected void onTargetJobNotFound(FlinkResourceContext<R> ctx) {
             // upgrading state and retry the upgrade (if possible)
             resource.getStatus().getReconciliationStatus().setState(ReconciliationState.DEPLOYED);
         }
-        jobStatus.setState(org.apache.flink.api.common.JobStatus.RECONCILING.name());
+        jobStatus.setState(JobStatus.RECONCILING.name());
         resource.getStatus().setError(JOB_NOT_FOUND_ERR);
     }
 

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java

@@ -26,7 +26,6 @@
 import org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec;
 import org.apache.flink.kubernetes.operator.api.spec.JobSpec;
 import org.apache.flink.kubernetes.operator.api.spec.JobState;
-import org.apache.flink.kubernetes.operator.api.spec.UpgradeMode;
 import org.apache.flink.kubernetes.operator.api.status.CommonStatus;
 import org.apache.flink.kubernetes.operator.api.status.FlinkDeploymentStatus;
 import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus;
@@ -43,7 +42,6 @@
 import org.apache.flink.kubernetes.operator.exception.ValidationException;
 import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
 import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
-import org.apache.flink.runtime.jobmanager.HighAvailabilityMode;
 
 import io.fabric8.kubernetes.client.CustomResource;
 import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl;
@@ -296,6 +294,7 @@ UpdateControl<R> toUpdateControl(
         }
     }
 
+    @VisibleForTesting
     public static Duration rescheduleAfter(
             JobManagerDeploymentStatus status,
             FlinkDeployment flinkDeployment,
@@ -308,7 +307,6 @@ public static Duration rescheduleAfter(
             case READY:
                 rescheduleAfter =
                         savepointInProgress(flinkDeployment.getStatus().getJobStatus())
-                                        || isJobCancelling(flinkDeployment.getStatus())
                                 ? operatorConfiguration.getProgressCheckInterval()
                                 : operatorConfiguration.getReconcileInterval();
                 break;
@@ -329,18 +327,6 @@ private static boolean savepointInProgress(JobStatus jobStatus) {
         return StringUtils.isNotEmpty(jobStatus.getSavepointInfo().getTriggerId());
     }
 
-    public static boolean isUpgradeModeChangedToLastStateAndHADisabledPreviously(
-            AbstractFlinkResource<?, ?> flinkApp, Configuration observeConfig) {
-
-        var deployedSpec = getDeployedSpec(flinkApp);
-        UpgradeMode previousUpgradeMode = deployedSpec.getJob().getUpgradeMode();
-        UpgradeMode currentUpgradeMode = flinkApp.getSpec().getJob().getUpgradeMode();
-
-        return previousUpgradeMode != UpgradeMode.LAST_STATE
-                && currentUpgradeMode == UpgradeMode.LAST_STATE
-                && !HighAvailabilityMode.isHighAvailabilityModeActivated(observeConfig);
-    }
-
     public static <SPEC extends AbstractFlinkSpec> SPEC getDeployedSpec(
             AbstractFlinkResource<SPEC, ?> deployment) {
         var reconciliationStatus = deployment.getStatus().getReconciliationStatus();

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java

@@ -39,8 +39,6 @@
 import io.fabric8.kubernetes.client.KubernetesClient;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nullable;
 
@@ -51,8 +49,6 @@
 /** Service for submitting and interacting with Flink clusters and jobs. */
 public interface FlinkService {
 
-    Logger LOG = LoggerFactory.getLogger(FlinkService.class);
-
     KubernetesClient getKubernetesClient();
 
     void submitApplicationCluster(JobSpec jobSpec, Configuration conf, boolean requireHaMetadata)