Package org.apache.flink.kubernetes.operator.reconciler

Class ReconciliationUtils

java.lang.Object

org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils

public class ReconciliationUtils
extends java.lang.Object

Reconciliation utilities.

Constructor Summary

Constructors

Constructor	Description
ReconciliationUtils()

Method Summary

Modifier and Type	Method	Description
static <SPEC extends org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec> boolean	applyValidationErrorAndResetSpec(FlinkResourceContext<? extends org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<SPEC,?>> ctx, java.lang.String validationError)	In case of validation errors we need to (temporarily) reset the old spec so that we can reconcile other outstanding changes, instead of simply blocking.
static void	checkAndUpdateStableSpec(org.apache.flink.kubernetes.operator.api.status.CommonStatus<?> status)	Checks the status and if the corresponding Flink job/application is in stable running state, it updates the last stable spec.
static void	clearLastReconciledSpecIfFirstDeploy(org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<?,?> resource)	Clear last reconciled spec if that corresponds to the first deployment.
static <T> T	clone(T object)
static <SPEC extends org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec> SPEC	getDeployedSpec(org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<SPEC,?> deployment)
static java.lang.Long	getUpgradeTargetGeneration(org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<?,?> resource)	Get spec generation for the current in progress upgrade.
static boolean	isJobInTerminalState(org.apache.flink.kubernetes.operator.api.status.CommonStatus<?> status)
static boolean	isJobRunning(org.apache.flink.kubernetes.operator.api.status.CommonStatus<?> status)
static boolean	isUpgradeModeChangedToLastStateAndHADisabledPreviously(org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<?,?> flinkApp, org.apache.flink.configuration.Configuration observeConfig)
static java.time.Duration	rescheduleAfter(org.apache.flink.kubernetes.operator.api.status.JobManagerDeploymentStatus status, org.apache.flink.kubernetes.operator.api.FlinkDeployment flinkDeployment, FlinkOperatorConfiguration operatorConfiguration)
static <STATUS extends org.apache.flink.kubernetes.operator.api.status.CommonStatus<?>,R extends org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<?,STATUS>> io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl<R>	toErrorStatusUpdateControl(FlinkResourceContext<R> ctx, java.lang.Exception e, StatusRecorder<R,STATUS> statusRecorder)	Update the resource error status and metrics when the operator encountered an exception during reconciliation.
static <SPEC extends org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec,STATUS extends org.apache.flink.kubernetes.operator.api.status.CommonStatus<SPEC>,R extends io.fabric8.kubernetes.client.CustomResource<SPEC,STATUS>> io.javaoperatorsdk.operator.api.reconciler.UpdateControl<R>	toUpdateControl(FlinkOperatorConfiguration operatorConfiguration, R current, R previous, boolean reschedule)
static <SPEC extends org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec> void	updateAfterScaleUp(org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<SPEC,?> target, org.apache.flink.configuration.Configuration deployConfig, java.time.Clock clock, FlinkService.ScalingResult scalingResult)
static void	updateForReconciliationError(FlinkResourceContext ctx, java.lang.Throwable error)
static <SPEC extends org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec> void	updateLastReconciledSnapshotTriggerNonce(org.apache.flink.kubernetes.operator.api.status.SnapshotInfo snapshotInfo, org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<SPEC,?> target, SnapshotType snapshotType)
static <SPEC extends org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec> void	updateReconciliationMetadata(org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<SPEC,?> resource)
static <SPEC extends org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec> void	updateStatusBeforeDeploymentAttempt(org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<SPEC,?> target, org.apache.flink.configuration.Configuration conf)
static <SPEC extends org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec> void	updateStatusBeforeDeploymentAttempt(org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<SPEC,?> target, org.apache.flink.configuration.Configuration conf, java.time.Clock clock)	Update status before deployment attempt of a new resource spec.
static void	updateStatusForAlreadyUpgraded(org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<?,?> resource)	Updates status in cases where a previously successful deployment wasn't recorded for any reason.
static <SPEC extends org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec> void	updateStatusForDeployedSpec(org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<SPEC,?> target, org.apache.flink.configuration.Configuration conf)
static <SPEC extends org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec> void	updateStatusForDeployedSpec(org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<SPEC,?> target, org.apache.flink.configuration.Configuration conf, java.time.Clock clock)	Update status after successful deployment of a new resource spec.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

ReconciliationUtils

public ReconciliationUtils()

Method Detail

updateStatusForDeployedSpec

public static <SPEC extends org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec> void updateStatusForDeployedSpec(org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<SPEC,?> target,
                                                                                                                              org.apache.flink.configuration.Configuration conf,
                                                                                                                              java.time.Clock clock)

Update status after successful deployment of a new resource spec. Existing reconciliation errors will be cleared, lastReconciled spec will be updated and for suspended jobs it will also be marked stable.

For Application deployments TaskManager info will also be updated.

Type Parameters:

SPEC - Spec type.

Parameters:

target - Target Flink resource.

conf - Deployment configuration.

clock - Clock for getting time.

updateStatusForDeployedSpec

@VisibleForTesting
public static <SPEC extends org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec> void updateStatusForDeployedSpec(org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<SPEC,?> target,
                                                                                                                              org.apache.flink.configuration.Configuration conf)

updateStatusBeforeDeploymentAttempt

public static <SPEC extends org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec> void updateStatusBeforeDeploymentAttempt(org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<SPEC,?> target,
                                                                                                                                      org.apache.flink.configuration.Configuration conf,
                                                                                                                                      java.time.Clock clock)

Update status before deployment attempt of a new resource spec. Existing reconciliation errors will be cleared, lastReconciled spec will be updated and reconciliation status marked UPGRADING.

For Application deployments TaskManager info will also be updated.

Type Parameters:

SPEC - Spec type.

Parameters:

target - Target Flink resource.

conf - Deployment configuration.

clock - Clock for getting the current time.

updateStatusBeforeDeploymentAttempt

@VisibleForTesting
public static <SPEC extends org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec> void updateStatusBeforeDeploymentAttempt(org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<SPEC,?> target,
                                                                                                                                      org.apache.flink.configuration.Configuration conf)

updateAfterScaleUp

public static <SPEC extends org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec> void updateAfterScaleUp(org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<SPEC,?> target,
                                                                                                                     org.apache.flink.configuration.Configuration deployConfig,
                                                                                                                     java.time.Clock clock,
                                                                                                                     FlinkService.ScalingResult scalingResult)

updateLastReconciledSnapshotTriggerNonce

public static <SPEC extends org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec> void updateLastReconciledSnapshotTriggerNonce(org.apache.flink.kubernetes.operator.api.status.SnapshotInfo snapshotInfo,
                                                                                                                                           org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<SPEC,?> target,
                                                                                                                                           SnapshotType snapshotType)

updateForReconciliationError

public static void updateForReconciliationError(FlinkResourceContext ctx,
                                                java.lang.Throwable error)

clone

public static <T> T clone(T object)

toUpdateControl

public static <SPEC extends org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec,STATUS extends org.apache.flink.kubernetes.operator.api.status.CommonStatus<SPEC>,R extends io.fabric8.kubernetes.client.CustomResource<SPEC,STATUS>> io.javaoperatorsdk.operator.api.reconciler.UpdateControl<R> toUpdateControl(FlinkOperatorConfiguration operatorConfiguration,
                                                                                                                                                                                                                                                                                                                                                R current,
                                                                                                                                                                                                                                                                                                                                                R previous,
                                                                                                                                                                                                                                                                                                                                                boolean reschedule)

rescheduleAfter

public static java.time.Duration rescheduleAfter(org.apache.flink.kubernetes.operator.api.status.JobManagerDeploymentStatus status,
                                                 org.apache.flink.kubernetes.operator.api.FlinkDeployment flinkDeployment,
                                                 FlinkOperatorConfiguration operatorConfiguration)

isUpgradeModeChangedToLastStateAndHADisabledPreviously

public static boolean isUpgradeModeChangedToLastStateAndHADisabledPreviously(org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<?,?> flinkApp,
                                                                             org.apache.flink.configuration.Configuration observeConfig)

getDeployedSpec

public static <SPEC extends org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec> SPEC getDeployedSpec(org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<SPEC,?> deployment)

isJobInTerminalState

public static boolean isJobInTerminalState(org.apache.flink.kubernetes.operator.api.status.CommonStatus<?> status)

isJobRunning

public static boolean isJobRunning(org.apache.flink.kubernetes.operator.api.status.CommonStatus<?> status)

applyValidationErrorAndResetSpec

public static <SPEC extends org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec> boolean applyValidationErrorAndResetSpec(FlinkResourceContext<? extends org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<SPEC,?>> ctx,
                                                                                                                                      java.lang.String validationError)

In case of validation errors we need to (temporarily) reset the old spec so that we can reconcile other outstanding changes, instead of simply blocking.

This is only possible if we have a previously reconciled spec.

For in-flight application upgrades we need extra logic to set the desired job state to running

Type Parameters:

SPEC - Spec type.

Parameters:

ctx - The current deployment context

validationError - Validation error encountered for the current spec

Returns:

True if the spec was reset and reconciliation can continue. False if nothing to reconcile.

toErrorStatusUpdateControl

public static <STATUS extends org.apache.flink.kubernetes.operator.api.status.CommonStatus<?>,R extends org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<?,STATUS>> io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl<R> toErrorStatusUpdateControl(FlinkResourceContext<R> ctx,
                                                                                                                                                                                                                                                                                                java.lang.Exception e,
                                                                                                                                                                                                                                                                                                StatusRecorder<R,STATUS> statusRecorder)

Update the resource error status and metrics when the operator encountered an exception during reconciliation.

Type Parameters:

STATUS - Status type.

R - Resource type.

Parameters:

ctx - Flink Resource context

e - Exception that caused the retry

statusRecorder - statusRecorder object for patching status

Returns:

This always returns Empty optional currently, due to the status update logic

getUpgradeTargetGeneration

public static java.lang.Long getUpgradeTargetGeneration(org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<?,?> resource)

Get spec generation for the current in progress upgrade.

Parameters:

resource - Flink resource.

Returns:

The spec generation for the upgrade.

clearLastReconciledSpecIfFirstDeploy

public static void clearLastReconciledSpecIfFirstDeploy(org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<?,?> resource)

Clear last reconciled spec if that corresponds to the first deployment. This is important in cases where the first deployment fails.

Parameters:

resource - Flink resource.

checkAndUpdateStableSpec

public static void checkAndUpdateStableSpec(org.apache.flink.kubernetes.operator.api.status.CommonStatus<?> status)

Checks the status and if the corresponding Flink job/application is in stable running state, it updates the last stable spec.

Parameters:

status - Status to be updated.

updateStatusForAlreadyUpgraded

public static void updateStatusForAlreadyUpgraded(org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<?,?> resource)

Updates status in cases where a previously successful deployment wasn't recorded for any reason. We simply change the job status from SUSPENDED to RUNNING and ReconciliationState to DEPLOYED while keeping the metadata.

Parameters:

resource - Flink resource to be updated.

updateReconciliationMetadata

public static <SPEC extends org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec> void updateReconciliationMetadata(org.apache.flink.kubernetes.operator.api.AbstractFlinkResource<SPEC,?> resource)