You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "gaborgsomogyi (via GitHub)" <gi...@apache.org> on 2023/06/15 12:12:19 UTC
[GitHub] [flink-kubernetes-operator] gaborgsomogyi commented on a diff in pull request #614: [FLINK-32057] Support 1.18 rescale api for applying parallelism overrides
gaborgsomogyi commented on code in PR #614:
URL: https://github.com/apache/flink-kubernetes-operator/pull/614#discussion_r1230837995
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/NativeFlinkService.java:
##########
@@ -132,4 +165,153 @@ protected void deleteClusterInternal(
deleteHAData(namespace, clusterId, conf);
}
}
+
+ @Override
+ public boolean scale(FlinkResourceContext<?> ctx) throws Exception {
+ var resource = ctx.getResource();
+ var spec = resource.getSpec();
+
+ var observeConfig = ctx.getObserveConfig();
+
+ if (spec.getJob() == null
+ || !observeConfig.get(
+ KubernetesOperatorConfigOptions.JOB_UPGRADE_INPLACE_SCALING_ENABLED)) {
+ return false;
+ }
+
+ if (!observeConfig.get(FLINK_VERSION).isNewerVersionThan(FlinkVersion.v1_17)) {
+ LOG.debug("In-place rescaling is only available starting from Flink 1.18");
+ return false;
+ }
+
+ if (!observeConfig
+ .get(JobManagerOptions.SCHEDULER)
+ .equals(JobManagerOptions.SchedulerType.Adaptive)) {
+ LOG.debug("In-place rescaling is only available with the adaptive scheduler");
+ return false;
+ }
+
+ var status = resource.getStatus();
+ if (ReconciliationUtils.isJobInTerminalState(status)
+ || JobStatus.RECONCILING.name().equals(status.getJobStatus().getState())) {
+ LOG.info("Job in terminal or reconciling state cannot be scaled in-place");
Review Comment:
+1 to have either such log or comment. One can follow the logic better w/ such explanations.
##########
examples/autoscaling/Dockerfile:
##########
@@ -16,5 +16,5 @@
# limitations under the License.
################################################################################
-FROM flink:1.17
+FROM ghcr.io/apache/flink-docker:1.18-SNAPSHOT-scala_2.12-java11-debian
Review Comment:
I know there is no 1.18 release yet but using snapshot can make tests fail randomly.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractFlinkDeploymentObserver.java:
##########
@@ -233,42 +233,41 @@ private void onMissingDeployment(FlinkDeployment deployment) {
}
@Override
- protected void updateStatusToDeployedIfAlreadyUpgraded(
- FlinkResourceContext<FlinkDeployment> ctx) {
+ protected boolean checkIfAlreadyUpgraded(FlinkResourceContext<FlinkDeployment> ctx) {
var flinkDep = ctx.getResource();
var status = flinkDep.getStatus();
+
+ // We are performing a full upgrade
Optional<Deployment> depOpt = ctx.getJosdkContext().getSecondaryResource(Deployment.class);
- depOpt.ifPresent(
- deployment -> {
- Map<String, String> annotations = deployment.getMetadata().getAnnotations();
- if (annotations == null) {
- return;
- }
- Long deployedGeneration =
- Optional.ofNullable(annotations.get(FlinkUtils.CR_GENERATION_LABEL))
- .map(Long::valueOf)
- .orElse(-1L);
-
- Long upgradeTargetGeneration =
- ReconciliationUtils.getUpgradeTargetGeneration(flinkDep);
-
- if (deployedGeneration.equals(upgradeTargetGeneration)) {
- logger.info("Pending upgrade is already deployed, updating status.");
- ReconciliationUtils.updateStatusForAlreadyUpgraded(flinkDep);
- if (flinkDep.getSpec().getJob() != null) {
- status.getJobStatus()
- .setState(
- org.apache.flink.api.common.JobStatus.RECONCILING
- .name());
- }
- status.setJobManagerDeploymentStatus(JobManagerDeploymentStatus.DEPLOYING);
- } else {
- logger.warn(
- "Running deployment generation {} doesn't match upgrade target generation {}.",
- deployedGeneration,
- upgradeTargetGeneration);
- }
- });
+
+ if (!depOpt.isPresent()) {
+ return false;
Review Comment:
Other places of `return false/true` there is an explanation what happens. Maybe we can add something to all places. This stands not just here but all boolean exits.
##########
flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java:
##########
@@ -547,14 +553,30 @@ public Map<String, String> getClusterInfo(Configuration conf) {
}
@Override
- public boolean scale(ObjectMeta meta, JobSpec jobSpec, Configuration conf) {
- if (conf.get(JobManagerOptions.SCHEDULER_MODE) != SchedulerExecutionMode.REACTIVE
- && jobSpec != null) {
+ public boolean scale(FlinkResourceContext<?> ctx) {
+ boolean standalone = ctx.getDeploymentMode() == KubernetesDeploymentMode.STANDALONE;
+ boolean session = ctx.getResource().getSpec().getJob() == null;
+ if (!standalone) {
return false;
}
- desiredReplicas =
- conf.get(StandaloneKubernetesConfigOptionsInternal.KUBERNETES_TASKMANAGER_REPLICAS);
- return true;
+
+ if (session
+ || ctx.getObserveConfig().get(JobManagerOptions.SCHEDULER_MODE)
+ == SchedulerExecutionMode.REACTIVE) {
+ desiredReplicas =
+ ctx.getDeployConfig(ctx.getResource().getSpec())
+ .get(
+ StandaloneKubernetesConfigOptionsInternal
+ .KUBERNETES_TASKMANAGER_REPLICAS);
+ return true;
+ }
+
+ return false;
Review Comment:
Unless we want to add comments/logs it's better to have a single if statement for easier understanding.
```
if (ctx.getDeploymentMode() != KubernetesDeploymentMode.STANDALONE &&
(ctx.getResource().getSpec().getJob() == null || ctx.getObserveConfig().get(JobManagerOptions.SCHEDULER_MODE) == SchedulerExecutionMode.REACTIVE))
```
##########
flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java:
##########
@@ -112,13 +119,33 @@ public TestingFlinkDeploymentController(
public UpdateControl<FlinkDeployment> reconcile(
FlinkDeployment flinkDeployment, Context<FlinkDeployment> context) throws Exception {
FlinkDeployment cloned = ReconciliationUtils.clone(flinkDeployment);
+ updateGeneration(cloned);
statusUpdateCounter.setCurrent(flinkDeployment);
UpdateControl<FlinkDeployment> updateControl =
flinkDeploymentController.reconcile(cloned, context);
Assertions.assertTrue(updateControl.isNoUpdate());
return updateControl;
}
+ private void updateGeneration(FlinkDeployment resource) {
Review Comment:
Is there a diff compared to TestingFlinkSessionJobController.java or just copy/paste?
--
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