logicamp/flink-kube-operator
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(savepoint): periodically savepoint bug

Browse Source
This commit is contained in:
Mohamad Khani 2024-12-13 15:11:36 +03:30
parent 444cfc7061
commit 699cf12f72
1 changed files with 5 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
internal/managed_job/savepoint.go
View File

@ -43,8 +43,9 @@ func (job ManagedJob) trackSavepoint() error {
return v1alpha1.ErrNoSavepointTriggerId return v1alpha1.ErrNoSavepointTriggerId
} }
resp, err := job.client.TrackSavepoint(*job.def.Status.JobId, *job.def.Status.SavepointTriggerId) resp, err := job.client.TrackSavepoint(*job.def.Status.JobId, *job.def.Status.SavepointTriggerId)
pkg.Logger.Debug("[managed-job] [savepoint] track savepoint", pkg.Logger.Info("[managed-job] [savepoint] savepoint track result",
zap.Any("status.Id", resp.Status.Id), zap.Any("status.Id", resp.Status.Id),
zap.Any("path", resp.Operation.Location),
zap.Any("failureCause.stacktrace", resp.Operation.FailureCause.StackTrace), zap.Any("failureCause.stacktrace", resp.Operation.FailureCause.StackTrace),
zap.Any("failureCause.class", resp.Operation.FailureCause.Class), zap.Any("failureCause.class", resp.Operation.FailureCause.Class),
zap.Error(err), zap.Error(err),
@ -69,8 +70,9 @@ func (job ManagedJob) trackSavepoint() error {
if resp.Status.Id == api.SavepointStatusInCompleted { if resp.Status.Id == api.SavepointStatusInCompleted {
job.crd.Patch(job.def.UID, map[string]interface{}{ job.crd.Patch(job.def.UID, map[string]interface{}{
"status": map[string]interface{}{ "status": map[string]interface{}{
"lastSavepointPath": resp.Operation.Location, "savepointTriggerId": nil,
"lastSavepointDate": time.Now().Format(time.RFC3339), "lastSavepointPath": resp.Operation.Location,
"lastSavepointDate": time.Now().Format(time.RFC3339),
}, },
}) })
} }