[GitHub] [flink-kubernetes-operator] gyfora commented on a diff in pull request #278: [WIP] Add standalone mode support

[GitBox <gi...@apache.org>]

gyfora commented on code in PR #278:
URL: https://github.com/apache/flink-kubernetes-operator/pull/278#discussion_r915920165


##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java:
##########
@@ -494,4 +500,42 @@ public static <SPEC extends AbstractFlinkSpec> void updateStatusForAlreadyUpgrad
         reconciliationStatus.setLastReconciledSpec(
                 ReconciliationUtils.writeSpecWithMeta(lastSpecWithMeta.f0, lastSpecWithMeta.f1));
     }
+
+    public static boolean shouldRollBackDeployment(
+            ReconciliationStatus<FlinkDeploymentSpec> reconciliationStatus,
+            Configuration configuration,
+            FlinkService flinkService) {
+
+        if (reconciliationStatus.getState() == ReconciliationState.ROLLING_BACK) {
+            return true;
+        }
+
+        if (!configuration.get(KubernetesOperatorConfigOptions.DEPLOYMENT_ROLLBACK_ENABLED)
+                || reconciliationStatus.getState() == ReconciliationState.ROLLED_BACK
+                || reconciliationStatus.isLastReconciledSpecStable()) {
+            return false;
+        }
+
+        var lastStableSpec = reconciliationStatus.deserializeLastStableSpec();
+        if (lastStableSpec != null
+                && lastStableSpec.getJob() != null
+                && lastStableSpec.getJob().getState() == JobState.SUSPENDED) {
+            // Should not roll back to suspended state
+            return false;
+        }
+
+        Duration readinessTimeout =
+                configuration.get(KubernetesOperatorConfigOptions.DEPLOYMENT_READINESS_TIMEOUT);
+        if (!Instant.now()

Review Comment:
   This method/logic has been moved to the `AbstractFlinkResourceReconciler`



##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java:
##########
@@ -216,30 +229,94 @@ protected FlinkConfigBuilder applyTaskManagerSpec() throws IOException {
                     spec.getTaskManager().getPodTemplate(),
                     effectiveConfig,
                     false);
+
+            if (spec.getTaskManager().getReplicas() != null
+                    && spec.getTaskManager().getReplicas() > 0) {
+                effectiveConfig.set(
+                        StandaloneKubernetesConfigOptionsInternal.KUBERNETES_TASKMANAGER_REPLICAS,
+                        spec.getTaskManager().getReplicas());
+            }
+        }
+
+        if (spec.getJob() != null
+                && KubernetesDeploymentMode.getDeploymentMode(spec)
+                        == KubernetesDeploymentMode.STANDALONE) {
+            final int tmTaskSlots = effectiveConfig.get(TaskManagerOptions.NUM_TASK_SLOTS);
+            final int jobParallelism = spec.getJob().getParallelism();
+            if (!effectiveConfig.contains(
+                            StandaloneKubernetesConfigOptionsInternal
+                                    .KUBERNETES_TASKMANAGER_REPLICAS)
+                    && tmTaskSlots > 0
+                    && jobParallelism > 0) {
+                effectiveConfig.set(
+                        StandaloneKubernetesConfigOptionsInternal.KUBERNETES_TASKMANAGER_REPLICAS,
+                        (int) Math.ceil((double) jobParallelism / tmTaskSlots));
+            }

Review Comment:
   I think this logic could be simplified by a combination of the `getParallelism()` and `FlinkUtils#getNumTaskManagers() ` . At the moment it seems a little overly complex :) 
   
   



##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java:
##########
@@ -206,12 +204,6 @@ protected void rollback(CR resource, Context context, Configuration observeConfi
         reconciliationStatus.setState(ReconciliationState.ROLLED_BACK);
     }
 
-    @Override
-    public boolean reconcileOtherChanges(CR resource, Configuration observeConfig)
-            throws Exception {
-        return SavepointUtils.triggerSavepointIfNeeded(flinkService, resource, observeConfig);
-    }

Review Comment:
   Why did you remove it from here? It applies for both sessionjobs and applications



##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/InternalKubernetesOperatorConfigOptions.java:
##########
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.operator.config;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.kubernetes.operator.crd.spec.KubernetesDeploymentMode;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/** This class holds internal configuration constants used by the flink operator. */
+public class InternalKubernetesOperatorConfigOptions {
+
+    public static final ConfigOption<KubernetesDeploymentMode> KUBERNETES_DEPLOYMENT_MODE =
+            key("kubernetes.operator.internal.deployment-mode")

Review Comment:
   can we prefix with `$internal.` please?



##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java:
##########
@@ -288,41 +282,8 @@ private boolean checkNewSpecAlreadyDeployed(CR resource, Configuration deployCon
      * @param configuration Flink cluster configuration.
      * @return True if the resource should be rolled back.
      */
-    private boolean shouldRollBack(
-            ReconciliationStatus<SPEC> reconciliationStatus, Configuration configuration) {
-
-        if (reconciliationStatus.getState() == ReconciliationState.ROLLING_BACK) {
-            return true;
-        }
-
-        if (!configuration.get(KubernetesOperatorConfigOptions.DEPLOYMENT_ROLLBACK_ENABLED)
-                || reconciliationStatus.getState() == ReconciliationState.ROLLED_BACK
-                || reconciliationStatus.isLastReconciledSpecStable()) {
-            return false;
-        }

Review Comment:
   Why did you remove this?



-- 
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org