diff --git a/crds.yaml b/crds.yaml
index 7cb09eb..c4572ad 100644
--- a/crds.yaml
+++ b/crds.yaml
@@ -44,9 +44,8 @@ spec:
                 jarURIBasicAuthPassword:
                   type: string
                 args:
-                  type: array
-                  items:
-                    type: string
+                  description: "Program arguments. Accepts a list of strings (--key, value, ...) or a map of key-value pairs."
+                  x-kubernetes-preserve-unknown-fields: true
                 savepointInterval:
                   type: string
                   format: duration
diff --git a/example-job.yaml b/example-job.yaml
index 8cc8e5e..6537f48 100644
--- a/example-job.yaml
+++ b/example-job.yaml
@@ -11,6 +11,17 @@ spec:
   jarUri: "https://git.logicamp.tech/api/packages/logiline/generic/facility-enrichment/1.0.0/facility-enrichment.jar"
   jarURIBasicAuthUsername: logiline-actrunner
   jarURIBasicAuthPassword: daeweeb7ohpaiw3oojiCoong
+  # args accepts two formats:
+  #   Map (recommended):
+  args:
+    kafka-host: kafka.pg-release:9092
+    kafka-group-id: batch-sales-config-dedup-processor
+  #   List (also valid):
+  #   args:
+  #     - "--kafka-host"
+  #     - kafka.pg-release:9092
+  #     - "--kafka-group-id"
+  #     - batch-sales-config-dedup-processor
   flinkConfiguration:
     taskmanager.numberOfTaskSlots: "1"
     parallelism.default: "1"
\ No newline at end of file
diff --git a/go.mod b/go.mod
index 987bada..dfae126 100644
--- a/go.mod
+++ b/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.1
+	github.com/logi-camp/go-flink-client v0.2.2
 	github.com/samber/lo v1.47.0
 	go.uber.org/zap v1.27.0
 	k8s.io/apimachinery v0.31.3
diff --git a/go.sum b/go.sum
index b0e8fb3..2f856eb 100644
--- a/go.sum
+++ b/go.sum
@@ -62,8 +62,8 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 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.1 h1:STfKamFm9+2SxxfZO3ysdFsb5MViQdThB4UHbnkUOE8=
-github.com/logi-camp/go-flink-client v0.2.1/go.mod h1:A79abedX6wGQI0FoICdZI7SRoGHj15QwMwWowgsKYFI=
+github.com/logi-camp/go-flink-client v0.2.2 h1:aWqibzcJiv02Myf/WowZafI0hhHp5x7BJGRCmI6PT20=
+github.com/logi-camp/go-flink-client v0.2.2/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=
diff --git a/helm/chart/values.yaml b/helm/chart/values.yaml
index f8acb32..4ff742c 100644
--- a/helm/chart/values.yaml
+++ b/helm/chart/values.yaml
@@ -11,7 +11,7 @@ image:
   # This sets the pull policy for images.
   pullPolicy: Always
   # Overrides the image tag whose default is the chart appVersion.
-  tag: v1.3.0
+  tag: v1.3.1
 
 # This is for the secretes for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
 imagePullSecrets: []
diff --git a/internal/crd/v1alpha1/flink_job.go b/internal/crd/v1alpha1/flink_job.go
index 53e8af6..a920710 100644
--- a/internal/crd/v1alpha1/flink_job.go
+++ b/internal/crd/v1alpha1/flink_job.go
@@ -1,7 +1,10 @@
 package v1alpha1
 
 import (
+	"encoding/json"
 	"errors"
+	"sort"
+	"strings"
 	"time"
 
 	metaV1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -9,6 +12,53 @@ import (
 
 //go:generate go run sigs.k8s.io/controller-tools/cmd/controller-gen object paths=$GOFILE
 
+// Args is a list of program arguments passed to the Flink job.
+// It accepts two formats in JSON/YAML:
+//
+//	List:  ["--kafka-host", "broker:9092", "--kafka-group-id", "my-group"]
+//	Map:   {kafka-host: "broker:9092", kafka-group-id: "my-group"}
+//
+// Map keys are automatically prefixed with "--" and expanded into alternating
+// key/value pairs. The internal representation is always a flat []string.
+type Args []string
+
+// UnmarshalJSON implements dual-format deserialization.
+func (a *Args) UnmarshalJSON(data []byte) error {
+	// Try map format first: {"key": "value", ...}
+	var m map[string]string
+	if err := json.Unmarshal(data, &m); err == nil {
+		*a = make(Args, 0, len(m)*2)
+		// Sort keys for deterministic ordering.
+		keys := make([]string, 0, len(m))
+		for k := range m {
+			keys = append(keys, k)
+		}
+		sort.Strings(keys)
+		for _, k := range keys {
+			*a = append(*a, "--"+k, m[k])
+		}
+		return nil
+	}
+
+	// Fall back to list format: ["--key", "value", ...]
+	var s []string
+	if err := json.Unmarshal(data, &s); err != nil {
+		return err
+	}
+	*a = Args(s)
+	return nil
+}
+
+// MarshalJSON serializes Args in the compact map format.
+func (a Args) MarshalJSON() ([]byte, error) {
+	m := make(map[string]string, len(a)/2)
+	for i := 0; i < len(a)-1; i += 2 {
+		key := strings.TrimPrefix(a[i], "--")
+		m[key] = a[i+1]
+	}
+	return json.Marshal(m)
+}
+
 type FlinkJobSpec struct {
 	Key                     string          `json:"key"`
 	Name                    string          `json:"name"`
@@ -19,7 +69,7 @@ type FlinkJobSpec struct {
 	JarURIBasicAuthPassword *string         `json:"jarURIBasicAuthPassword"`
 	SavepointInterval       metaV1.Duration `json:"savepointInterval"`
 	EntryClass              string          `json:"entryClass"`
-	Args                    []string        `json:"args"`
+	Args                    Args            `json:"args,omitempty"`
 }
 
 type FlinkJobStatus struct {