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

feat: allow restore jobs that has not any taken save point

Browse Source
This commit is contained in:
Mohamad Khani 2024-12-13 02:50:43 +03:30
parent c066a9fd90
commit b2d2295d07
2 changed files with 10 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
internal/managed_job/cycle.go
View File

@ -102,6 +102,10 @@ func (job *ManagedJob) Cycle() {
job.restore() job.restore()
return return
} }
if job.def.Status.JobStatus == v1alpha1.JobStatusFailed && job.def.Status.LastSavepointPath == nil {
job.restore()
return
}
pkg.Logger.Warn("[managed-job] [cycle]", zap.String("unhanded job status", string(job.def.Status.JobStatus))) pkg.Logger.Warn("[managed-job] [cycle]", zap.String("unhanded job status", string(job.def.Status.JobStatus)))
} }

9
internal/managed_job/restore.go
View File

@ -14,23 +14,26 @@ import (
// restore the job from savepoint and jarId in managedJob // restore the job from savepoint and jarId in managedJob
func (job *ManagedJob) restore() error { func (job *ManagedJob) restore() error {
var savepointPath string
if job.def.Status.LastSavepointPath == nil { if job.def.Status.LastSavepointPath == nil {
pkg.Logger.Error("[managed-job] [restore]", zap.Error(v1alpha1.ErrNoSavepointPath)) pkg.Logger.Error("[managed-job] [restore]", zap.Error(v1alpha1.ErrNoSavepointPath))
return v1alpha1.ErrNoSavepointPath //return v1alpha1.ErrNoSavepointPath
} else {
savepointPath = *job.def.Status.LastSavepointPath
} }
if job.def.Status.JarId == nil { if job.def.Status.JarId == nil {
err := errors.New("missing jar id") err := errors.New("missing jar id")
pkg.Logger.Error("[managed-job] [restore]", zap.Error(err)) pkg.Logger.Error("[managed-job] [restore]", zap.Error(err))
return err return err
} }
pkg.Logger.Info("[managed-job] [restore] restoring job", zap.String("name", job.def.GetName()), zap.String("savepointPath", *job.def.Status.LastSavepointPath)) pkg.Logger.Info("[managed-job] [restore] restoring job", zap.String("name", job.def.GetName()), zap.String("savepointPath", savepointPath))
var jobId *string var jobId *string
for { for {
runJarResp, err := job.client.RunJar(api.RunOpts{ runJarResp, err := job.client.RunJar(api.RunOpts{
JarID: *job.def.Status.JarId, JarID: *job.def.Status.JarId,
AllowNonRestoredState: true, AllowNonRestoredState: true,
EntryClass: job.def.Spec.EntryClass, EntryClass: job.def.Spec.EntryClass,
SavepointPath: *job.def.Status.LastSavepointPath, SavepointPath: savepointPath,
}) })
if err != nil { if err != nil {
if strings.ContainsAny(err.Error(), ".jar does not exist") { if strings.ContainsAny(err.Error(), ".jar does not exist") {