gyfora commented on code in PR #357:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/357#discussion_r958388232
##########
examples/basic-ingress.yaml:
##########
@@ -39,3 +39,4 @@ spec:
job:
jarURI: local:///opt/flink/examples/streaming/StateMachineExample.jar
parallelism: 2
+ mode: native
Review Comment:
missing newline at end
##########
examples/custom-logging.yaml:
##########
@@ -63,3 +63,4 @@ spec:
logger.zookeeper.level = INFO
logger.netty.name =
org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline
logger.netty.level = OFF
+ mode: native
Review Comment:
missing newline at end
##########
examples/basic-session-deployment-only.yaml:
##########
@@ -34,3 +34,4 @@ spec:
resource:
memory: "2048m"
cpu: 1
+ mode: native
Review Comment:
missing newline at end
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java:
##########
@@ -417,6 +421,33 @@ private static String createTempFile(Pod podTemplate)
throws IOException {
return tmp.getAbsolutePath();
}
+ private static void setPvcTemplate(
+ List<PersistentVolumeClaim> persistentVolumeClaims,
+ Configuration effectiveConfig,
+ boolean isJm)
+ throws IOException {
+ if (persistentVolumeClaims != null) {
+ final ConfigOption<String> pvcConfigOptions =
+ isJm
+ ?
StandaloneKubernetesConfigOptionsInternal.JOB_MANAGER_PVC_TEMPLATE
+ :
StandaloneKubernetesConfigOptionsInternal.TASK_MANAGER_PVC_TEMPLATE;
+ effectiveConfig.setString(
+ pvcConfigOptions, String.join(",",
createPvcTempFile(persistentVolumeClaims)));
+ }
+ }
+
+ private static List<String> createPvcTempFile(List<PersistentVolumeClaim>
pvcTemplates)
Review Comment:
We should add the newly created files to the list of cleaned up files under
`cleanupTmpFiles` in this class
##########
examples/standalone_pvc.yaml:
##########
@@ -0,0 +1,81 @@
+################################################################################
+# 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.
+################################################################################
+
+kind: FlinkDeployment
+metadata:
+ namespace: default
+ name: basic-example
+spec:
+ image: flink:1.15.1
+ flinkVersion: v1_15
+ flinkConfiguration:
+ taskmanager.numberOfTaskSlots: "2"
+ serviceAccount: flink
+ jobManager:
+ replicas: 1
+ resource:
+ memory: "2048m"
+ cpu: 1
+ volumeClaimTemplates:
+ - metadata:
+ name: log
+ spec:
+ accessModes: [ "ReadWriteOnce" ]
+ ### Please override with custom supported storageClassName
+ storageClassName: "alicloud-local-lvm"
+ resources:
+ requests:
+ storage: 10Gi
+ podTemplate:
+ apiVersion: v1
+ kind: Pod
+ metadata:
+ name: job-manager-pod-template
+ spec:
+ containers:
+ - name: flink-main-container
+ volumeMounts:
+ - name: log
+ mountPath: /opt/flink/log
+ taskManager:
+ replicas: 1
+ resource:
+ memory: "2048m"
+ cpu: 1
+ volumeClaimTemplates:
+ - metadata:
+ name: log
+ spec:
+ accessModes: [ "ReadWriteOnce" ]
+ ### Please override with custom supported storageClassName
+ storageClassName: "alicloud-local-lvm"
+ resources:
+ requests:
+ storage: 10Gi
+ podTemplate:
+ apiVersion: v1
+ kind: Pod
+ metadata:
+ name: task-manager-pod-template
+ spec:
+ containers:
+ - name: flink-main-container
+ volumeMounts:
+ - name: log
+ mountPath: /opt/flink/log
+ mode: standalone
Review Comment:
missing newline at end
##########
examples/basic-checkpoint-ha.yaml:
##########
@@ -58,3 +58,4 @@ spec:
upgradeMode: savepoint
state: running
savepointTriggerNonce: 0
+ mode: native
Review Comment:
Can you add the missing newlines for the modified examples please?
##########
examples/standalone_pvc.yaml:
##########
@@ -0,0 +1,81 @@
+################################################################################
+# 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.
+################################################################################
+
+kind: FlinkDeployment
+metadata:
+ namespace: default
+ name: basic-example
+spec:
+ image: flink:1.15.1
+ flinkVersion: v1_15
+ flinkConfiguration:
+ taskmanager.numberOfTaskSlots: "2"
+ serviceAccount: flink
+ jobManager:
+ replicas: 1
+ resource:
+ memory: "2048m"
+ cpu: 1
+ volumeClaimTemplates:
+ - metadata:
+ name: log
+ spec:
+ accessModes: [ "ReadWriteOnce" ]
+ ### Please override with custom supported storageClassName
+ storageClassName: "alicloud-local-lvm"
Review Comment:
Can we remove the reference to `alicloud` ?
##########
examples/pod-template.yaml:
##########
@@ -82,3 +82,4 @@ spec:
jarURI: local:///opt/flink/downloads/flink-examples-streaming.jar
entryClass:
org.apache.flink.streaming.examples.statemachine.StateMachineExample
parallelism: 2
+ mode: native
Review Comment:
missing newline at end
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkUtils.java:
##########
@@ -187,6 +187,18 @@ public static void setGenerationAnnotation(Configuration
conf, Long generation)
conf.set(KubernetesConfigOptions.JOB_MANAGER_ANNOTATIONS, labels);
}
+ public static void setTaskmanagerGenerationAnnotation(Configuration conf,
Long generation) {
+ if (generation == null) {
+ return;
+ }
+ var labels =
+ new HashMap<>(
+
conf.getOptional(KubernetesConfigOptions.TASK_MANAGER_ANNOTATIONS)
+ .orElse(Collections.emptyMap()));
+ labels.put(CR_GENERATION_LABEL, generation.toString());
+ conf.set(KubernetesConfigOptions.TASK_MANAGER_ANNOTATIONS, labels);
+ }
Review Comment:
I feel this method duplicates too much code, we could simply have a generic
method that takes the TASK_MANAGER/JOBMANAGER_ANNOTATIONS config option as
param to factor out the logic
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java:
##########
@@ -141,41 +148,19 @@ protected void observeJmDeployment(
return;
}
- Optional<Deployment> deployment =
context.getSecondaryResource(Deployment.class);
- if (deployment.isPresent()) {
- DeploymentStatus status = deployment.get().getStatus();
- DeploymentSpec spec = deployment.get().getSpec();
- if (status != null
- && status.getAvailableReplicas() != null
- && spec.getReplicas().intValue() == status.getReplicas()
- && spec.getReplicas().intValue() ==
status.getAvailableReplicas()
- && flinkService.isJobManagerPortReady(effectiveConfig)) {
-
- // typically it takes a few seconds for the REST server to be
ready
- logger.info(
- "JobManager deployment port is ready, waiting for the
Flink REST API...");
- deploymentStatus.setJobManagerDeploymentStatus(
- JobManagerDeploymentStatus.DEPLOYED_NOT_READY);
+ if
(KubernetesDeploymentMode.NATIVE.equals(flinkApp.getSpec().getMode())) {
+ Optional<Deployment> deployment =
context.getSecondaryResource(Deployment.class);
+ if (deployment.isPresent()) {
+ jmDeploymentObserver(deployment.get(), effectiveConfig,
deploymentStatus, flinkApp);
return;
Review Comment:
Why do we use StatefulSets for JobManager also? Seems to be an unnecessary
complication without much benefits.
--
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]
