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

feat(config): add savepoint interval config

Browse Source
This commit is contained in:
Mohamad Khani 2024-11-30 02:18:44 +03:30
parent 8f4cb093b1
commit e95634c942
3 changed files with 9 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
config.yaml
View File

@ -3,3 +3,4 @@ jobs:
name: Price Processor name: Price Processor
entryClass: top.bazargam.Main entryClass: top.bazargam.Main
jarURI: http://price-processor.bz2/price-processor-v0.0.1.jar jarURI: http://price-processor.bz2/price-processor-v0.0.1.jar
savepointInterval: 3m

3
internal/config/config.type.go
View File

@ -1,10 +1,13 @@
package config package config
import "time"
type JobDef struct { type JobDef struct {
Key string `yaml:"key"` Key string `yaml:"key"`
Name string `yaml:"name"` Name string `yaml:"name"`
EntryClass string `yaml:"entryClass"` EntryClass string `yaml:"entryClass"`
JarURI string `yaml:"jarURI"` JarURI string `yaml:"jarURI"`
SavepointInterval time.Duration `yaml:"savepointInterval"`
} }
type Config struct { type Config struct {

2
internal/managed_job/cycle.go
View File

@ -58,7 +58,7 @@ func (job *ManagedJob) cycle() {
if errors.Is(err, ErrNoJobId) { if errors.Is(err, ErrNoJobId) {
job.state = nil job.state = nil
} }
if job.state.LastSavepointDate == nil || time.Now().Add(-time.Minute*3).After(*job.state.LastSavepointDate) { if job.state.LastSavepointDate == nil || time.Now().Add(-job.def.SavepointInterval).After(*job.state.LastSavepointDate) {
if job.state.SavepointTriggerId == nil { if job.state.SavepointTriggerId == nil {
job.createSavepoint() job.createSavepoint()
} else { } else {