[STRMCMP-1658] Enable Deploys to Fallback without State

[sethsaperstein-lyft]

overview

Add CRD parameter fallbackWithoutState to be enabled to allow deploys to occur without a savepoint or checkpoint in the case that the savepoint fails and there are no usable externalized checkpoints.

In the case that the savepoint and checkpoints are failing, this has a similar effect as savepointDisabled. The benefit of this configuration option is that it only exhibits this behavior when there is no other path forward and thus acts as a fallback without potential intervention. This is disabled by default.

The state graph does not change, but rather, the recovering state that attempts to find an externalized checkpoint forwards to submittingJob in the case that this is enabled and there is no externalized checkpoint

[sethsaperstein-lyft]

/PTAL @premsantosh @leoluoInSea @maghamravi

[sethsaperstein-lyft]

additional info

The original scope of this change was to handle 2 cases:

1. Savepoints and checkpoints are failing and we would like to start without state
2. Submitting job is failing and would otherwise succeed if not using a savepoint (ex: job graph change)

The original solution involved an extra state in the state graph for submitting without state. This state would be reached in the above two cases on upgrade as such:

1. savepointing -> recovering -> submittingJobWithoutState
2. savepointing -> submittingJob -> submittingJobWithoutState

After further review, it appears that case 2 can be solved by enabling the option allowNonRestoredState. This leaves only the first case to be handled in which the savepoint fails and no externalized checkpoints are available. In this case, we do not need a different state and instead can reuse the state submittingJob which will proceed without an empty savepointPath. This solves the issue while simplifying the solution. Consulted @maghamravi who agrees.

[sethsaperstein-lyft]

returning this back to 1.8.1 which corresponds to the image in dockerhub. Attempted to update this in the integration test PR but am reverting for now as not to deal with further configuration options that are non-trivial when dealing with 8GB memory total for github actions

[sethsaperstein-lyft]

Building the test app image as part of the integration test as opposed to relying on the remote image whose contents may not match the integ/operator-test-app contents

[anandswaminathan]

You are introducing a regression here when fallbackWithoutState is not set.

You are not failing deploy when there is an error

I see that you have handled the case below. Looks good

[anandswaminathan]

The logic looks good. Wondering if the duplication of lines can be avoided. Something like

failDeploy := false

if err !=nil  {
  failDeploy = true

} else path =="" {
  ...
  failDeploy = true
}

if failDeploy {
  if FallbackWithoutState {
  failDeploy = false
}
}

if failDeploy {
  s.deplpyFailed
}

submitjob

[premsantosh]

Is this comment incorrect because on line 115 you assert error is nil?

[premsantosh]

Oh! You can not verifying that the app is failing is it?

[sethsaperstein-lyft]

Lines have moved around since but if you are referring to s.Util.GetFlinkApplication this gets the k8s FlinkApplication CR and the err corresponds to if there were errors retrieving the CR object as opposed to anything related to the status of the actual job.

Ln 135 s.Util.FlinkAPIGet(newApp, endpoint) gets the actual flink job and corresponding status to show that the job itself is healthy

[sethsaperstein-lyft]

The logic looks good. Wondering if the duplication of lines can be avoided. Something like

failDeploy := false

if err !=nil  {
  failDeploy = true

} else path =="" {
  ...
  failDeploy = true
}

if failDeploy {
  if FallbackWithoutState {
  failDeploy = false
}
}

if failDeploy {
  s.deplpyFailed
}

submitjob

agreed. refactored