Skip to content

Commit

Permalink
PR review
Browse files Browse the repository at this point in the history
  • Loading branch information
avneet15 committed Dec 18, 2023
1 parent 614068a commit 46ab312
Show file tree
Hide file tree
Showing 2 changed files with 5 additions and 4 deletions.
1 change: 1 addition & 0 deletions pkg/controller/flinkapplication/flink_state_machine.go
Original file line number Diff line number Diff line change
Expand Up @@ -784,6 +784,7 @@ func (s *FlinkStateMachine) handleSubmittingJob(ctx context.Context, app *v1beta
if err != nil {
logger.Info(ctx, "Job monitoring failed with error: %v", err)
s.flinkController.LogEvent(ctx, app, corev1.EventTypeWarning, "JobMonitoringFailed", err.Error())
s.flinkController.UpdateLatestJobID(ctx, app, "")
s.updateApplicationPhase(app, v1beta1.FlinkApplicationRollingBackJob)
return statusChanged, err
}
Expand Down
8 changes: 4 additions & 4 deletions pkg/controller/flinkapplication/flink_state_machine_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -759,11 +759,11 @@ func TestSubmittingVertexFailsToStart(t *testing.T) {
mockK8Cluster.UpdateStatusFunc = func(ctx context.Context, object k8sclient.Object) error {
if statusUpdateCount == 1 {
application := object.(*v1beta1.FlinkApplication)
assert.Equal(t, jobID, mockFlinkController.GetLatestJobID(ctx, application))
assert.Equal(t, "", mockFlinkController.GetLatestJobID(ctx, application))
assert.Equal(t, v1beta1.FlinkApplicationRollingBackJob, application.Status.Phase)
} else if statusUpdateCount == 2 {
application := object.(*v1beta1.FlinkApplication)
assert.Equal(t, jobID, mockFlinkController.GetLatestJobID(ctx, application))
assert.Equal(t, "", mockFlinkController.GetLatestJobID(ctx, application))
assert.Equal(t, v1beta1.FlinkApplicationDeployFailed, application.Status.Phase)
} else if statusUpdateCount == 3 {
application := object.(*v1beta1.FlinkApplication)
Expand Down Expand Up @@ -928,11 +928,11 @@ func TestSubmittingVertexStartTimeout(t *testing.T) {
mockK8Cluster.UpdateStatusFunc = func(ctx context.Context, object k8sclient.Object) error {
if statusUpdateCount == 1 {
application := object.(*v1beta1.FlinkApplication)
assert.Equal(t, jobID, mockFlinkController.GetLatestJobID(ctx, application))
assert.Equal(t, "", mockFlinkController.GetLatestJobID(ctx, application))
assert.Equal(t, v1beta1.FlinkApplicationRollingBackJob, application.Status.Phase)
} else if statusUpdateCount == 2 {
application := object.(*v1beta1.FlinkApplication)
assert.Equal(t, jobID, mockFlinkController.GetLatestJobID(ctx, application))
assert.Equal(t, "", mockFlinkController.GetLatestJobID(ctx, application))
assert.Equal(t, v1beta1.FlinkApplicationDeployFailed, application.Status.Phase)
} else if statusUpdateCount == 3 {
application := object.(*v1beta1.FlinkApplication)
Expand Down

0 comments on commit 46ab312

Please sign in to comment.