feat: update flink http client library

Dockerfile

@@ -1,4 +1,4 @@
-FROM public.ecr.aws/docker/library/golang:1.23.4-bookworm AS build
+FROM public.ecr.aws/docker/library/golang:1.24.1-bookworm AS build
 
 ARG upx_version=4.2.4
 

go.mod

@@ -5,7 +5,7 @@ go 1.23.2
 require (
 	github.com/danielgtaylor/huma/v2 v2.27.0
 	github.com/gofiber/fiber/v2 v2.52.6
-	github.com/logi-camp/go-flink-client v0.2.0
+	github.com/logi-camp/go-flink-client v0.2.1
 	github.com/samber/lo v1.47.0
 	go.uber.org/zap v1.27.0
 	k8s.io/apimachinery v0.31.3

go.sum

@@ -64,6 +64,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/logi-camp/go-flink-client v0.2.0 h1:PIyfJq7FjW28bnvemReCicIuQD7JzVgJDk2xPTZUS2s=
 github.com/logi-camp/go-flink-client v0.2.0/go.mod h1:A79abedX6wGQI0FoICdZI7SRoGHj15QwMwWowgsKYFI=
+github.com/logi-camp/go-flink-client v0.2.1 h1:STfKamFm9+2SxxfZO3ysdFsb5MViQdThB4UHbnkUOE8=
+github.com/logi-camp/go-flink-client v0.2.1/go.mod h1:A79abedX6wGQI0FoICdZI7SRoGHj15QwMwWowgsKYFI=
 github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0=
 github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc=
 github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=

helm/chart/values.yaml

@@ -139,9 +139,9 @@ flink:
 
 
   taskManager:
-    numberOfTaskSlots: 12  # Number of task slots for TaskManager
+    numberOfTaskSlots: 12  # Number of task slots for task manager
     processMemory: 4096m # Size of task manager process memory
-    replicas: 1
+    replicas: 1 # Number of task manager replicas
     storage:
       rocksDb:
         size: 4Gi

internal/managed_job/cycle.go

@@ -13,7 +13,7 @@ func (job *ManagedJob) Cycle() {
 	// pkg.Logger.Debug("[managed-job] [new] check cycle", zap.String("jobName", job.def.GetName()))
 
 	// Init job
-	if job.def.Status.LifeCycleStatus == "" && job.def.Status.JobStatus == "" {
+	if job.def.Status.LifeCycleStatus == "" && (job.def.Status.JobStatus == "" || job.def.Status.JobStatus == v1alpha1.JobStatusFinished) {
 		job.Run(false)
 		return
 	}

internal/managed_job/manager.go

@@ -117,14 +117,15 @@ func (mgr *Manager) cycle(client *api.Client, crdInstance *crd.Crd) {
 				"status": patchStatusObj,
 			})
 		} else {
-			patchStatusObj := map[string]interface{}{
-				"jobStatus":       "",
-				"lifeCycleStatus": string(v1alpha1.LifeCycleStatusFailed),
-			}
+			// TODO handle job not found status
+			// patchStatusObj := map[string]interface{}{
+			// 	"jobStatus":       "",
+			// 	"lifeCycleStatus": string(v1alpha1.LifeCycleStatusFailed),
+			// }
 
-			crdInstance.Patch(uid, map[string]interface{}{
-				"status": patchStatusObj,
-			})
+			// crdInstance.Patch(uid, map[string]interface{}{
+			// 	"status": patchStatusObj,
+			// })
 		}
 
 		managedJob.Cycle()

internal/managed_job/run.go

@@ -43,7 +43,7 @@ func (job *ManagedJob) Run(restoreMode bool) error {
 				EntryClass:            job.def.Spec.EntryClass,
 				SavepointPath:         savepointPath,
 				Parallelism:           job.def.Spec.Parallelism,
-				ProgramArg:            job.def.Spec.Args,
+				ProgramArgsList:       job.def.Spec.Args,
 			})
 			if err == nil {
 				pkg.Logger.Info("[managed-job] [run] jar successfully ran", zap.Any("run-jar-resp", runJarResp))