[FLINK-33634] Add Conditions to Flink CRD's Status field #957

lajith2006 · 2025-03-20T05:15:17Z

What is the purpose of the change

This PR address the FLIP to add conditions to flink deployment status related to issue https://issues.apache.org/jira/browse/FLINK-33634

Brief change log

Added list of conditions to FlinkDeploymentStatus
A new class ConditionUtils introduced as utility class to build condition.

Verifying this change

Deployed and validated session job and application custer deployment.
Validated the status transitions.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changes to the CustomResourceDescriptors: yes
Core observer or reconciler logic that is regularly executed: (yes / no)

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

davidradl · 2025-03-21T15:51:36Z

@lajith2006 please could you fill in a descriptive title next to the Jira number

davidradl

removed comment

davidradl · 2025-03-21T15:56:23Z

@lajith2006 this implementation does not match the Flip words and has not been discussed in the slack thread, please can you update the discussion thread to ensure that you get consensus, and confirm that the Flip vote still is valid after this change.

davidradl · 2025-03-21T16:01:34Z

...api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkDeploymentStatus.java

+        } else if (getJobStatus() != null && getJobStatus().getState() != null) {
+            switch (getJobStatus().getState()) {
+                case RECONCILING:
+                    phase = "Pending";


why is this one not RECONCILING like the pattern the others follow others? I suggest a comment, also a constant is better then a an inline literal.

Right, as mentioned here , phase which is intended to keep the current status of FlinkDeployment especially useful in Openshift environment, as Openshift UI can render the value from status.phase and populate the current status of deployment.

As when FlinkDeployment applied in Application Mode, in the initial phase, as the JM brining up , job state would be in RECONCILING state, so status.phase is kept as "Pending".

davidradl · 2025-03-21T16:04:20Z

...api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkDeploymentStatus.java

+                    updateCondition(
+                            conditions,
+                            ConditionUtils.runningTrue(
+                                    "JobManager is running and ready to receive REST API call",


nit: call -> calls

davidradl · 2025-03-21T16:04:31Z

...api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkDeploymentStatus.java

+        if (reconciliationStatus != null
+                && reconciliationStatus.deserializeLastReconciledSpec() != null
+                && reconciliationStatus.deserializeLastReconciledSpec().getJob() == null) {
+            switch (jobManagerDeploymentStatus) {


can we have a datastructure, keyed off the jobManagerDeploymentStatus, that we interrogate to get the inserts for the updateConditions. Maybe a map with the key of the status and the value of an object RunningCondition, that has 2 fields the boolean and the description. Something like :

Map<JobManagerDeploymentStatus,String> jobmanagerDeploymentStatusMap = new HashMap<String,String>() {{ put(READY, new RunningCondition(true, "JobManager is running and ready to receive REST API calls")), .... }};

then the code just loops through the map updating conditions using the map values. Similar for jobstatus

Wondering will that add any value?. Any way to build the Map , we have to call ConditionUtils to build the Condition, so rather than have a Prebuild Map , we can directly call ConditionUtils to build them right. Your thoughts?.

I think you have pre built the map now - with map.of - looks good - thanks.

davidradl · 2025-03-21T16:11:59Z

...api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkDeploymentStatus.java

+                            conditions,
+                            ConditionUtils.runningFalse(
+                                    JobStatus.RESTARTING.name(),
+                                    "The job is currently undergoing a restarting"));


nit: "The job is currently undergoing a restarting" -> "The job is currently restarting"

davidradl · 2025-03-21T16:13:27Z

...api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkDeploymentStatus.java

+    private String phase;
+
+    public List<Condition> getConditions() {
+        if (reconciliationStatus != null


I am curious which parts of the code refer to application mode and which are session cluster. It would be good to have some comments to detail this and maybe use the mode name in method name or variable names to make this more intuitive to read.

I will add comments in the respective code.

davidradl · 2025-03-21T16:24:12Z

...api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkDeploymentStatus.java

+                            conditions,
+                            ConditionUtils.runningFalse(
+                                    "JobManager process is starting up",
+                                    "JobManager process is starting up"));


I am curious why the message and reaon is the same . I would expect them to be different. In this case I would expect the reason to be "A New JVM deployment exists and is being created" - i.e. the word on the arrow in the UML.

As per description here which says Reason is intended to be a one-word, CamelCase representation of the category of cause of the current status, and Message is intended to be a human-readable phrase or sentence, I would use the existing enum https://github.com/apache/flink-kubernetes-operator/blob/main/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/JobManagerDeploymentStatus.java, to use as reason and respective explanation as message

I see https://github.com/apache/flink-kubernetes-operator/pull/961/files, gives examples of camelcase single words reasons. We should follow that style for the cases that PR does not cover.

Done a refactoring for the conditions.

davidradl · 2025-03-21T16:26:48Z

...test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java

@@ -448,6 +465,15 @@ public void verifyUpgradeFromSavepointLegacyMode(FlinkVersion flinkVersion) thro
        assertEquals(
                "savepoint_1", appCluster.getStatus().getJobStatus().getUpgradeSavepointPath());

+        // Validate status conditions


can we use a java parameterized test (or more than one as appropriate) to cover all the permutations of the tests.

davidradl · 2025-03-21T16:30:57Z

...perator-api/src/main/java/org/apache/flink/kubernetes/operator/api/utils/ConditionUtils.java

+import java.util.Date;
+
+/**
+ * Creates a condition object with the specified parameters.


nit: with the specified parameters -> whether this is running , the message and the reason.
then for return just say @return A condition

davidradl · 2025-03-21T16:32:00Z

...perator-api/src/main/java/org/apache/flink/kubernetes/operator/api/utils/ConditionUtils.java

+ */
+public class ConditionUtils {
+    public static Condition runningTrue(final String message, final String reason) {
+        return crCondition("Running", "True", message, reason);


no need to pass "Running" as a parameter, we can hard code that in the method,

nit: rename crCondition -> crRunningCondition

I would like to keep method name as crCondition rather crRunningCondition , just because , crCondition will keep it as more generic , so that if in case, if we need to address any other condition with type other than Running , we just need to pass parameter to build respective condition.

I suggest doing that refactor when we need to, in the spirit of keeping the code lean.

@davidradl There were some refactoring done on this class.

davidradl · 2025-03-21T16:32:39Z

...perator-api/src/main/java/org/apache/flink/kubernetes/operator/api/utils/ConditionUtils.java

+    }
+
+    public static Condition runningFalse(final String message, final String reason) {
+        return crCondition("Running", "False", message, reason);


are there constants / enums for "False" and "True" we can use?
Os there an existing constant for "Running" we can use from OCP or the like. If not we should define this as a constant. We use this literal a few times in this change.

I haven't see any existing enum/constants for "False" / "True , instead looks like used as string itself in wherever required in existing code.

lajith2006 · 2025-04-01T07:38:53Z

As mentioned in the FLIP , we will have status.conditions, which holds the list of conditions that contains the status of Job transitions in each stage.

When the FlinkOperator is installed in Openshift and manages FlinkDeployment , with status.conditions in the CR , FlinkDeployment status in the Openshift UI will provide some misleading information.

Openshift UI works in such a way that, if there are multiple conditions , and at any point one of the condition had a flag of status with true and type as Running , irrespective of latest condition status has a flag as of false and type Running , In Openshift UI , status says that condition of deployment/job is Running. Which is misleading information, if you consider the scenario of Job was running and then later point job was suspended.

For example as below when a running Job is stopped at some point , we will have a condition as below.

conditions:
  - lastTransitionTime: '2025-04-01T06:17:08Z'
    message: Reconciling
    reason: Job is currently reconciling
    status: 'False'
    type: Running
  - lastTransitionTime: '2025-04-01T07:07:23Z'
    message: JobRunning
    reason: Job is running
    status: 'True'
    type: Running
  - lastTransitionTime: '2025-04-01T07:28:37Z'
    message: JobFinished
    reason: Job's tasks have successfully finished
    status: 'False'
    type: Running

and Openshift UI will have status as

So to overcome those problems, we will have a status.phase(same as other k8s resources like pod) , which will hold the current status of deployment/job. Openshift UI will take precedence of status.phase over status.condition to display the status of FlinkDeployment resource.

csviri · 2025-04-01T12:32:19Z

Would it be useful also to set also observedGeneration on the conditions, I believe it is in general a good practice.

csviri · 2025-04-01T12:51:16Z

...api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkDeploymentStatus.java

+                && existingCondition.getMessage().equals(condition.getMessage())) {
+            return;
+        }
+        conditions.add(condition);


It's maybe just me, just don't get how this list is truncated eventually, in other words, how is it prevented from growing indefinitely?

Thanks for review @csviri . Right now it's not preventing from growing indefinitely. As conditions reflects the different transitions state of Job/Deployment , thinking what could be the use case where can grow indefinitely?.

I think there should only ever be a single condition of type Running in the list that we return. Since we only have a single condition type right now, then the list should only have a single element. The latestTransition timestamp needs to represent when running changed from true->false or false->true. We can however keep updating the message if we want.

Thanks @gyfora for review , If I read correctly , you meant that we need to have only condition in the list at right now as we have only one type Running instead of multiple conditions of same type Running in the list as currently this PR having. And the lastTransitionTime in the condition must represent when the Running type changed its status from true>false or false > true. Is that correct?.

@lajith2006 you say the lastTransitionTime in the condition must represent when the Running type changed its status from true>false or false > true. Is that correct?. So I am curious what happens in the history if we change the reason Text? Can you check that if we change the reason text and not the running flag, we see all of the entries in the history if all the historical conditions have the same lastTransitionTime with changing reasons.

Yes @lajith2006 , there should be a single condition of type Running and lastTransitionTime when that changed

Thanks for clarification @gyfora

So just to be align with what you have mentioned , for an example , when we do an application cluster deployment, as Job cycles goes through CREATED>RUNNING

initially conditions will have as below

status conditions: - type: Running status: "False" reason: Job is created message: "Job is newly created, no task has started to run" lastTransitionTime: 2025-04-030T06:17:08Z

And then when the Job starts running at 2025-04-030T06:19:01Z , condition will have as below

status conditions: - type: Running status: "True" reason: Job is running message: "Job is running" lastTransitionTime: 2025-04-030T06:19:01Z

and then later at 2025-04-030T07:00:01Z. when job gets finished , condition will have as below.

status conditions: - type: Running status: "False" reason: Job's tasks have successfully finished message: Job's tasks have successfully finished lastTransitionTime: 2025-04-030T07:00:01Z

which means ideally we will not have an history in the condition when the Job goes through different job lifecycle , instead only single condition and lastTransitionTime will reflect when the status was changed.

Yes, I think this makes sense for a Running condition

davidradl · 2025-04-10T10:41:44Z

...api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkDeploymentStatus.java

+                && reconciliationStatus.deserializeLastReconciledSpec().getJob() == null) {
+            // Populate conditions for SessionMode deployment
+            switch (jobManagerDeploymentStatus) {
+                case READY:


instead of this switch can we just issue

updateCondition( conditions, ConditionUtils.crCondition( ConditionUtils.SESSION_MODE_CONDITION.get( jobManagerDeploymentStatus.name())));

@davidradl this has been revisited and made it simple by removing switch.

davidradl · 2025-04-10T10:43:03Z

...api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkDeploymentStatus.java

+            }
+        } else if (getJobStatus() != null && getJobStatus().getState() != null) {
+            // Populate conditions for ApplicationMode deployment
+            switch (getJobStatus().getState()) {


can this switch as-is be replaced with

updateCondition( conditions, ConditionUtils.crCondition( ConditionUtils.APPLICATION_MODE_CONDITION.get( getJobStatus().getState().name())));

I was thinking exactly the same here @lajith2006

Yes, @gyfora and @davidradl this has been revisited and addressed by removing switch and made it simple.

davidradl · 2025-04-10T10:45:35Z

...perator-api/src/main/java/org/apache/flink/kubernetes/operator/api/utils/ConditionUtils.java

+                            .withType("Running")
+                            .withStatus("True")
+                            .withMessage("Ready")
+                            .withReason("JobManager is running and ready to receive REST API calls")


isn't the reason - how it got to this state, not a description of the state itself?

https://kubernetes.io/docs/concepts/workloads/pods/pod-lifecycle/#pod-conditions

it says:
reason | Machine-readable, UpperCamelCase text indicating the reason for the condition's last transition.

@davidradl this has been updated and reason as is made as one-word, CamelCase representation of current status.

gyfora · 2025-05-12T05:30:38Z

...api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkDeploymentStatus.java

+                && reconciliationStatus.deserializeLastReconciledSpec() != null
+                && reconciliationStatus.deserializeLastReconciledSpec().getJob() == null) {


We should avoid deserializing twice here, could we simply check the job status instead?

Sure will use JobStatus.

gyfora · 2025-05-13T20:13:02Z

@lajith2006 , I think we should try to somehow try to wrap this PR up. We have been going back and forth on this for months, a lot of reviews, different people involved (many unaddressed comments from @davidradl). This is a significant time investment from all involved.

Overall I feel that you need to iterate over the changes once more and try to eliminate the parts that were added with a much more complex design in mind, and just simplify the methods, eliminate the large switch statements etc.

I would love to include this in the upcoming release but for that we need to merge it by the end of next week.

davidradl · 2025-05-14T08:53:10Z

...api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkDeploymentStatus.java

+                        ConditionUtils.crCondition(
+                                ConditionUtils.SESSION_MODE_CONDITION.get(
+                                        jobManagerDeploymentStatus.name())));
+            } else if (jobStatus != null) {


nit: } else if (jobStatus != null) { -> else {

davidradl · 2025-05-14T08:55:58Z

...api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkDeploymentStatus.java

+            if (existingCondition.isPresent()) {
+                return;
+            }
+            // Remove existing Condition with type running and then add a new condition that


nit: the comment is not quite right I think.
I suggest this comment be
// Remove existing Condition with type running if not the same as the new condition

then add a new condition that reflects the current state. should be after the if

It's been refactored..

davidradl · 2025-05-14T09:01:29Z

...api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkDeploymentStatus.java

+            conditions.removeIf(
+                    c ->
+                            c.getType().equals(CONDITION_TYPE_RUNNING)
+                                    && !c.getMessage().equals(newCondition.getMessage())


we have got here because the new condition is not the same as the existing condition. So we should just remove the existing condition without any conditions.

As-is the condition will not be removed if existing condition message or reason matches the new condition; which is not correct.

So we should just remove the existing condition without any conditions.

Right its fine for now, But I think, if in later point of time, if we are adding new conditions other than Running, just removing the existing condition without of any check will cause an issue.

As-is the condition will not be removed if existing condition message or reason matches the new condition; which is not correct.

That's right, but for Running type , we don't have any such case were reason or message can be matched with existing condition.

gyfora · 2025-05-14T18:15:32Z

...api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkDeploymentStatus.java

+        if (getJobStatus() != null) {
+            JobStatus jobStatus = getJobStatus().getState();
+            if (jobStatus == null) {
+                // Populate conditions for SessionMode deployment
+                updateCondition(
+                        conditions,
+                        ConditionUtils.crCondition(
+                                ConditionUtils.SESSION_MODE_CONDITION.get(
+                                        jobManagerDeploymentStatus.name())));
+            } else if (jobStatus != null) {
+                // Populate conditions for ApplicationMode deployment
+                updateCondition(
+                        conditions,
+                        ConditionUtils.crCondition(
+                                ConditionUtils.APPLICATION_MODE_CONDITION.get(jobStatus.name())));
+            }
+        }


I think this entire code block should be part of the ConditionUtils probably encapsulated into a single:

ConditionUtils. updateCondition(status)
method.

Thanks @gyfora . I believe , If we have them in ConditionUtils block, we will not be able to utilise the jobManagerDeploymentStatus ConditionUtils. Please Correct me if I am wrong.

This could be implemented as a static method that operates on the status

This could be implemented as a static method that operates on the status

@gyfora Yes - we were thinking changing to something like

if (getJobStatus() != null) { JobStatus jobStatus = getJobStatus().getState(); Condition conditionToAdd = null; if (jobStatus == null) { // Populate conditions for SessionMode deployment conditionToAdd = ConditionUtils.SESSION_MODE_CONDITION.get( jobManagerDeploymentStatus.name()); } else { // Populate conditions for ApplicationMode deployment conditionToAdd = ConditionUtils.APPLICATION_MODE_CONDITION.get(jobStatus.name()); } conditions = ConditionsUtils.createListOfConditions(conditionToAdd); }

Can you see a slicker way of doing this given we need jobManagerDeploymentStatus.name() for session and jobStatus.name() for application mode - which we know at this level. We could add a ternary if, I am not too worried about this.

I just wanted to factor out this condition logic from the FlinkDeploymentStatus class, it doesn't seem to belong there. I don't have an issue with he logic itself

Moved logic to ConditionUtils

gyfora · 2025-05-14T18:18:42Z

...api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkDeploymentStatus.java

+        if (newCondition.getType().equals(CONDITION_TYPE_RUNNING)) {
+            Optional<Condition> existingCondition =
+                    conditions.stream()
+                            .filter(
+                                    c ->
+                                            c.getType().equals(CONDITION_TYPE_RUNNING)
+                                                    && c.getReason()
+                                                            .equals(newCondition.getReason())
+                                                    && c.getMessage()
+                                                            .equals(newCondition.getMessage()))
+                            .findFirst();
+            // Until there is a condition change which reflects the latest state, no need to add
+            // condition to list.
+            if (existingCondition.isPresent()) {
+                return;
+            }
+            // Remove existing Condition with type running and then add a new condition that


Honestly I don't get this whole logic here.
We should have a single condition at the moment of type RUNNING. So a at any point in time we either have a SingletonList or empty list.

Can we simply always update the list to List.of(newCondition) + update the transition timestamp if the value changed from true->false or vice versa.

@davidradl @lajith2006 am I missing something ?

@gyfora I talked with @lajith2006. His code originally was wanting to handle more than one condition and manage the existing list to prevent 2 running conditions being in the list.

We agreed to keep it simple and amend this code as you suggest i.e. just add a new list with one running condition. We thought the transition time should also be updated if the message changes, as there are cases in the flow diagram in the Flip where the running condition remains false, but the message changes, in this case we think that we should update the transition time WDYT?

lastTransitionTime time should refer to the time when the status changes (true -> false or false->true) at least this is how it is used everywhere I know and everywhere I find online. Changed message alone should not update the timestamp as it would break the semantics and makes it impossible to tell when the job actually stopped running for example.

Thanks @gyfora , modified in such a way that lastTransitionTimeStamp gets updated when status changes from true> false or vice versa.

gyfora · 2025-05-14T18:25:01Z

...perator-api/src/main/java/org/apache/flink/kubernetes/operator/api/utils/ConditionUtils.java

+                            .withMessage("JobManager deployment failed")
+                            .build());
+
+    public static final Map<String, Condition> APPLICATION_MODE_CONDITION =


To me this entire mapping feels like that it should be replaced with a method,

return new ConditionBuilder() .withType(CONDITION_TYPE_RUNNING) .withStatus(status == RUNNING ? "True" : "False") .withReason(status.name()) .withMessage("Job state " + status.name) .build()

Thanks, I have a them in method. As reason is expected to be in camelcase, I have added a method to make it as camecase from status.name().

gyfora · 2025-05-14T18:25:57Z

...perator-api/src/main/java/org/apache/flink/kubernetes/operator/api/utils/ConditionUtils.java

+                            .withStatus("True")
+                            .withReason("JobManagerReady")
+                            .withMessage(
+                                    "JobManager is running and ready to receive REST API calls")


May be a better design to add this fields into the enum directly then simply get it from it.

Thanks, Sure, have addressed using enum .

davidradl · 2025-05-16T10:53:16Z

...perator-api/src/main/java/org/apache/flink/kubernetes/operator/api/utils/ConditionUtils.java

+                            .withMessage(JobManagerDeploymentConditionStatus.ERROR.getMessage())
+                            .build());
+
+    public static Condition getCondition(FlinkDeploymentStatus flinkDeploymentStatus) {


Only the parameters that are required should be passed not the complete FlinkDeploymentStatus.

I think from here @gyfora meant by operate on status on that util method .

I think from here @gyfora meant by operate on status on that util method .

OK I see

davidradl · 2025-05-16T10:53:56Z

...ava/org/apache/flink/kubernetes/operator/api/status/JobManagerDeploymentConditionStatus.java

+public enum JobManagerDeploymentConditionStatus {
+    READY("True", "JobManagerReady", "JobManager is running and ready to receive REST API calls"),
+    MISSING("False", "JobManagerDeploymentMissing", "JobManager deployment not found"),
+    DEPLOYING("False", "JobManagerIsDeploying", "JobManager process is starting up"),


is deploying a starting up the same thing?

I have referenced from here https://github.com/apache/flink-kubernetes-operator/blob/main/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/JobManagerDeploymentStatus.java#L29. I would have as JobManager deployment is in progress. Happy to make change , WDYT?.

OK that makes sense. Maybe link to that class in a comment to say where you got the descriptions from.

davidradl · 2025-05-16T10:54:19Z

...ava/org/apache/flink/kubernetes/operator/api/status/JobManagerDeploymentConditionStatus.java

+    DEPLOYED_NOT_READY(
+            "False",
+            "DeployedNotReady",
+            "JobManager is running but not ready yet to receive REST API calls"),


nit ready yet -> yet ready

Thanks, I will change that .

davidradl · 2025-05-16T10:54:38Z

...perator-api/src/main/java/org/apache/flink/kubernetes/operator/api/utils/ConditionUtils.java

+                .build();
+    }
+
+    private static String toCameCase(String reason) {


I think you mean toCamelCase

Thank you, my bad.

davidradl · 2025-05-16T10:57:20Z

...perator-api/src/main/java/org/apache/flink/kubernetes/operator/api/utils/ConditionUtils.java

+
+    private static String toCameCase(String reason) {
+        reason = reason.toLowerCase();
+        return reason.substring(0, 1).toUpperCase() + reason.substring(1);


Camel case is not just upper casing the first letter. We may need to upper case in the string as well. I suggest mapping the lower case to the appropriate camel cased reason.

Right, but not as of now as per here. If we go with mapping, we end up with new method, do we required them?.

davidradl · 2025-05-16T11:01:07Z

...perator-api/src/main/java/org/apache/flink/kubernetes/operator/api/utils/ConditionUtils.java

+        if (condition == null) {
+            return;
+        }
+        if (isLastTransactionTimeStampUpdateRequired(conditions, condition)) {


would it not be simpler to have an existing condition that is null or a value, then we do not need to get the first element of the list twice.

davidradl · 2025-05-16T11:04:43Z

...perator-api/src/main/java/org/apache/flink/kubernetes/operator/api/utils/ConditionUtils.java

+            return;
+        }
+        if (isLastTransactionTimeStampUpdateRequired(conditions, condition)) {
+            condition.setLastTransitionTime(


I am wondering whether we should use https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-release-1.11/api/java/org/apache/flink/autoscaler/utils/DateTimeUtils.html#readable(java.time.Instant,java.time.ZoneId)
here

Above is from flink-autoscaler , but flink-kubernetes-operator-api , doesn't have dependency to flink-autoscaler , so we can't utilise it.

davidradl · 2025-05-16T11:05:31Z

...perator-api/src/main/java/org/apache/flink/kubernetes/operator/api/utils/ConditionUtils.java

+                .withType(CONDITION_TYPE_RUNNING)
+                .withStatus(jobStatus == RUNNING ? "True" : "False")
+                .withReason(toCameCase(jobStatus.name()))
+                .withMessage("Job state " + jobStatus.name())


should the message say Job status - as that is what we are reading?

davidradl · 2025-05-16T11:08:58Z

...perator-api/src/main/java/org/apache/flink/kubernetes/operator/api/utils/ConditionUtils.java

+/** Creates a condition object with the type, status, message and reason. */
+public class ConditionUtils {
+    public static final String CONDITION_TYPE_RUNNING = "Running";
+    private static final Map<String, Condition> SESSION_MODE_CONDITION =


@gyfora are you happy with this map in the utils class - it might be cleaner to be in its own appropriately named class, same for the application mode map

To be completely honest, I am not very happy about it... :)

This is again a copy paste, logic that could be simply replaced with:

public Condition from(JobManagerDeploymentStatus jmStatus) { return new ConditionBuilder() .withType(jmStatus == DEPLOYED ? CONDITION_TYPE_RUNNING : NOT_RUNNING) .withStatus(jmStatus.getStatus()) .withReason(jmStatus.getReason()) .withMessage(jmStatus.getMessage()) .build() }

I have already given the same exact feedback to the JobStatus, not sure why it wasn't followed here too.

It has been moved to method and moved reason and messages to enum and getting from there.

davidradl · 2025-05-16T11:17:20Z

...perator-api/src/main/java/org/apache/flink/kubernetes/operator/api/utils/ConditionUtils.java

+    private static final Map<String, Condition> SESSION_MODE_CONDITION =
+            Map.of(
+                    JobManagerDeploymentStatus.READY.name(),
+                    new ConditionBuilder()


nit: you could refactor to a method and pass JobManagerDeploymentConditionStatus.READY, as the rest of the condition builder is the same apart from the status .
So

JobManagerDeploymentStatus.READY.name(), createRunningConditionBuilder(JobManagerDeploymentConditionStatus.READY), JobManagerDeploymentStatus.MISSING.name(), createRunningConditionBuilder(JobManagerDeploymentConditionStatus.MISSING), ...

It has been moved to method and reason and messages are getting from enum.

davidradl · 2025-05-22T10:18:38Z

...perator-api/src/main/java/org/apache/flink/kubernetes/operator/api/utils/ConditionUtils.java

+                                    flinkDeploymentStatus.getJobManagerDeploymentStatus())
+                            : getApplicationModeCondition(jobStatus);
+        }
+


can we not update the last transition time here?

davidradl · 2025-05-22T17:28:34Z

Would it be useful also to set also observedGeneration on the conditions, I believe it is in general a good practice.

It seems that this would be useful if the condition.observedGeneration does not matches the metadata.generation value.
@gyfora WDYT? We could add this is a following on pr if you think this is a reasonable addition.

lajith2006 · 2025-05-23T15:46:41Z

Would it be useful also to set also observedGeneration on the conditions, I believe it is in general a good practice.

by looking into some docs, It seems like condition offers a meaningful status info if the cond.observedGeneration is same as metadata.generation, provided condition is calculated based on the last configuration of the object. So in our case here , as condition represents the different status of job, not sure if we have cond.observedGeneration, that will matches with metadata.generation. WDYT?

lajith2006 added 4 commits March 12, 2025 14:49

[FLINK-33634] Add Conditions to Flink CRD's Status field

afdf40b

[FLINK-33634] Add Conditions to Flink CRD's Status field

1d84164

[FLINK-33634] Add Conditions to Flink CRD's Status field

727bb4b

[FLINK-33634] Add Conditions to Flink CRD's Status field

c66fd1f

davidradl suggested changes Mar 21, 2025

View reviewed changes

davidradl reviewed Mar 21, 2025

View reviewed changes

gyfora changed the title ~~FLINK-33634~~ [FLINK-33634] Add Conditions to Flink CRD's Status field Mar 23, 2025

lajith2006 added 3 commits March 26, 2025 17:09

[FLINK-33634] Add Conditions to Flink CRD's Status field

c623f33

[FLINK-33634] Add Conditions to Flink CRD's Status field

196c494

[FLINK-33634] Add Conditions to Flink CRD's Status field

32294f4

csviri reviewed Apr 1, 2025

View reviewed changes

davidradl reviewed Apr 10, 2025

View reviewed changes

[FLINK-33634] Add Conditions to Flink CRD's Status field

7b6d6a1

gyfora reviewed May 12, 2025

View reviewed changes

[FLINK-33634] Add Conditions to Flink CRD's Status field

4f6679b

lajith2006 added 2 commits May 14, 2025 12:17

[FLINK-33634] Add Conditions to Flink CRD's Status field

93be98a

[FLINK-33634] Add Conditions to Flink CRD's Status field

9faab16

davidradl reviewed May 14, 2025

View reviewed changes

gyfora reviewed May 14, 2025

View reviewed changes

[FLINK-33634] Add Conditions to Flink CRD's Status field

b40dcf3

davidradl reviewed May 16, 2025

View reviewed changes

[FLINK-33634] Add Conditions to Flink CRD's Status field

70030db

davidradl reviewed May 16, 2025

View reviewed changes

[FLINK-33634] Add Conditions to Flink CRD's Status field

9311031

davidradl reviewed May 22, 2025

View reviewed changes

		&& reconciliationStatus.deserializeLastReconciledSpec() != null
		&& reconciliationStatus.deserializeLastReconciledSpec().getJob() == null) {

[FLINK-33634] Add Conditions to Flink CRD's Status field #957

Are you sure you want to change the base?

[FLINK-33634] Add Conditions to Flink CRD's Status field #957

Uh oh!

Conversation

lajith2006 commented Mar 20, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

davidradl commented Mar 21, 2025

Uh oh!

davidradl left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

davidradl commented Mar 21, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

davidradl Mar 21, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lajith2006 Mar 26, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lajith2006 Mar 26, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

davidradl Mar 21, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lajith2006 commented Apr 1, 2025

Uh oh!

csviri commented Apr 1, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

lajith2006 commented Mar 20, 2025 •

edited

Loading

davidradl left a comment •

edited

Loading

davidradl Mar 21, 2025 •

edited

Loading

lajith2006 Mar 26, 2025 •

edited

Loading

lajith2006 Mar 26, 2025 •

edited

Loading

davidradl Mar 21, 2025 •

edited

Loading

lajith2006 Apr 10, 2025 •

edited

Loading

davidradl Apr 10, 2025 •

edited

Loading

davidradl Apr 10, 2025 •

edited

Loading

davidradl Apr 10, 2025 •

edited

Loading