diff --git a/docs/docs/schema/defaults.json b/docs/docs/schema/defaults.json
index c8c7ca909..fb978b0d9 100644
--- a/docs/docs/schema/defaults.json
+++ b/docs/docs/schema/defaults.json
@@ -295,7 +295,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "List of JMX metric rules.",
                     "title": "Metricrules"
                 },
@@ -893,7 +893,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "matchExpressions is a list of label selector requirements. The requirements are ANDed.",
                     "title": "Matchexpressions"
                 },
@@ -909,7 +909,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "matchLabels is a map of {key,value} pairs. A single {key,value} in the matchLabels map is equivalent to an element of matchExpressions, whose key field is *key*, the operator is *In*, and the values array contains only *value*. The requirements are ANDed.",
                     "title": "Matchlabels"
                 }
@@ -971,7 +971,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "The scheduler will prefer to schedule pods to nodes that satisfy the affinity expressions specified by this field, but it may choose a node that violates one or more of the expressions. The node that is most preferred is the one with the greatest sum of weights, i.e. for each node that meets all of the scheduling requirements (resource request, requiredDuringScheduling affinity expressions, etc.), compute a sum by iterating through the elements of this field and adding *weight* to the sum if the node matches the corresponding matchExpressions; the node(s) with the highest sum are the most preferred.",
                     "title": "Preferredduringschedulingignoredduringexecution"
                 },
@@ -1065,7 +1065,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "A list of node selector requirements by node's labels.",
                     "title": "Matchexpressions"
                 },
@@ -1081,7 +1081,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "A list of node selector requirements by node's fields.",
                     "title": "Matchfields"
                 }
@@ -1168,7 +1168,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "The scheduler will prefer to schedule pods to nodes that satisfy the affinity expressions specified by this field, but it may choose a node that violates one or more of the expressions. The node that is most preferred is the one with the greatest sum of weights, i.e. for each node that meets all of the scheduling requirements (resource request, requiredDuringScheduling affinity expressions, etc.), compute a sum by iterating through the elements of this field and adding weight to the sum if the node has pods which matches the corresponding podAffinityTerm; the node(s) with the highest sum are the most preferred.",
                     "title": "Preferredduringschedulingignoredduringexecution"
                 },
@@ -1184,7 +1184,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "If the affinity requirements specified by this field are not met at scheduling time, the pod will not be scheduled onto the node. If the affinity requirements specified by this field cease to be met at some point during pod execution (e.g. due to a pod label update), the system may or may not try to eventually evict the pod from its node. When there are multiple elements, the lists of nodes corresponding to each podAffinityTerm are intersected, i.e. all terms must be satisfied.",
                     "title": "Requiredduringschedulingignoredduringexecution"
                 }
@@ -1219,7 +1219,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "MatchLabelKeys is a set of pod label keys to select which pods will be taken into consideration. The keys are used to lookup values from the incoming pod labels, those key-value labels are merged with `labelSelector` as `key in (value)` to select the group of existing pods which pods will be taken into consideration for the incoming pod's pod (anti) affinity. Keys that don't exist in the incoming pod labels will be ignored. The default value is empty. The same key is forbidden to exist in both matchLabelKeys and labelSelector. Also, matchLabelKeys cannot be set when labelSelector isn't set. This is a beta field and requires enabling MatchLabelKeysInPodAffinity feature gate (enabled by default).",
                     "title": "Matchlabelkeys"
                 },
@@ -1235,7 +1235,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "MismatchLabelKeys is a set of pod label keys to select which pods will be taken into consideration. The keys are used to lookup values from the incoming pod labels, those key-value labels are merged with `labelSelector` as `key notin (value)` to select the group of existing pods which pods will be taken into consideration for the incoming pod's pod (anti) affinity. Keys that don't exist in the incoming pod labels will be ignored. The default value is empty. The same key is forbidden to exist in both mismatchLabelKeys and labelSelector. Also, mismatchLabelKeys cannot be set when labelSelector isn't set. This is a beta field and requires enabling MatchLabelKeysInPodAffinity feature gate (enabled by default).",
                     "title": "Mismatchlabelkeys"
                 },
@@ -1263,7 +1263,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "",
                     "title": "Namespaces"
                 },
@@ -1590,7 +1590,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "Array containing taint references. When defined, pods can run on nodes, which would otherwise deny scheduling.",
                     "title": "Tolerations"
                 }
@@ -1652,7 +1652,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Map of command line arguments passed to the streams app.",
                     "title": "Commandline"
                 },
@@ -1694,7 +1694,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Custom environment variables.",
                     "title": "Env"
                 },
@@ -1720,7 +1720,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Map of files to mount for the app. File will be mounted as $value.mountPath/$key. $value.content denotes file content (recommended to be used with --set-file).",
                     "title": "Files"
                 },
@@ -1756,7 +1756,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "",
                     "title": "Imagepullsecrets"
                 },
@@ -1803,7 +1803,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.25/#probe-v1-core",
                     "title": "Livenessprobe"
                 },
@@ -1833,7 +1833,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Map of custom annotations to attach to the pod spec.",
                     "title": "Podannotations"
                 },
@@ -1849,7 +1849,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Map of custom labels to attach to the pod spec.",
                     "title": "Podlabels"
                 },
@@ -1865,7 +1865,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "",
                     "title": "Ports"
                 },
@@ -1878,7 +1878,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.25/#probe-v1-core",
                     "title": "Readinessprobe"
                 },
@@ -1931,7 +1931,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "Mount existing secrets as volumes",
                     "title": "Secretfilesrefs"
                 },
@@ -1944,7 +1944,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Inject existing secrets as environment variables. Map key is used as environment variable name. Value consists of secret name and key.",
                     "title": "Secretrefs"
                 },
@@ -1960,7 +1960,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Custom secret environment variables. Prefix with configurationEnvPrefix in order to pass secrets to command line or prefix with KAFKA_ to pass secrets to Kafka Streams configuration.",
                     "title": "Secrets"
                 },
@@ -2014,7 +2014,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "Array containing taint references. When defined, pods can run on nodes, which would otherwise deny scheduling.",
                     "title": "Tolerations"
                 },
@@ -2743,7 +2743,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "Array containing taint references. When defined, pods can run on nodes, which would otherwise deny scheduling.",
                     "title": "Tolerations"
                 }
@@ -2804,7 +2804,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Map of command line arguments passed to the streams app.",
                     "title": "Commandline"
                 },
@@ -2833,7 +2833,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Custom environment variables.",
                     "title": "Env"
                 },
@@ -2846,7 +2846,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Map of files to mount for the app. File will be mounted as $value.mountPath/$key. $value.content denotes file content (recommended to be used with --set-file).",
                     "title": "Files"
                 },
@@ -2882,7 +2882,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "",
                     "title": "Imagepullsecrets"
                 },
@@ -2941,7 +2941,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.25/#probe-v1-core",
                     "title": "Livenessprobe"
                 },
@@ -2983,7 +2983,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Map of custom annotations to attach to the pod spec.",
                     "title": "Podannotations"
                 },
@@ -2999,7 +2999,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Map of custom labels to attach to the pod spec.",
                     "title": "Podlabels"
                 },
@@ -3015,7 +3015,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "",
                     "title": "Ports"
                 },
@@ -3040,7 +3040,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.25/#probe-v1-core",
                     "title": "Readinessprobe"
                 },
@@ -3068,7 +3068,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "Mount existing secrets as volumes",
                     "title": "Secretfilesrefs"
                 },
@@ -3081,7 +3081,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Inject existing secrets as environment variables. Map key is used as environment variable name. Value consists of secret name and key.",
                     "title": "Secretrefs"
                 },
@@ -3097,7 +3097,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Custom secret environment variables. Prefix with configurationEnvPrefix in order to pass secrets to command line or prefix with KAFKA_ to pass secrets to Kafka Streams configuration.",
                     "title": "Secrets"
                 },
@@ -3144,7 +3144,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "Array containing taint references. When defined, pods can run on nodes, which would otherwise deny scheduling.",
                     "title": "Tolerations"
                 }
@@ -3395,7 +3395,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "Array containing taint references. When defined, pods can run on nodes, which would otherwise deny scheduling.",
                     "title": "Tolerations"
                 }
@@ -3444,7 +3444,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Map of command line arguments passed to the streams app.",
                     "title": "Commandline"
                 },
@@ -3473,7 +3473,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Custom environment variables.",
                     "title": "Env"
                 },
@@ -3486,7 +3486,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Map of files to mount for the app. File will be mounted as $value.mountPath/$key. $value.content denotes file content (recommended to be used with --set-file).",
                     "title": "Files"
                 },
@@ -3522,7 +3522,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "",
                     "title": "Imagepullsecrets"
                 },
@@ -3569,7 +3569,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.25/#probe-v1-core",
                     "title": "Livenessprobe"
                 },
@@ -3599,7 +3599,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Map of custom annotations to attach to the pod spec.",
                     "title": "Podannotations"
                 },
@@ -3615,7 +3615,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Map of custom labels to attach to the pod spec.",
                     "title": "Podlabels"
                 },
@@ -3631,7 +3631,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "",
                     "title": "Ports"
                 },
@@ -3644,7 +3644,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.25/#probe-v1-core",
                     "title": "Readinessprobe"
                 },
@@ -3672,7 +3672,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "Mount existing secrets as volumes",
                     "title": "Secretfilesrefs"
                 },
@@ -3685,7 +3685,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Inject existing secrets as environment variables. Map key is used as environment variable name. Value consists of secret name and key.",
                     "title": "Secretrefs"
                 },
@@ -3701,7 +3701,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Custom secret environment variables. Prefix with configurationEnvPrefix in order to pass secrets to command line or prefix with KAFKA_ to pass secrets to Kafka Streams configuration.",
                     "title": "Secrets"
                 },
@@ -3729,7 +3729,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "Array containing taint references. When defined, pods can run on nodes, which would otherwise deny scheduling.",
                     "title": "Tolerations"
                 }
@@ -4009,7 +4009,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "List of additional KEDA triggers, see https://keda.sh/docs/latest/scalers/",
                     "title": "Additionaltriggers"
                 },
@@ -4057,7 +4057,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "List of auto-generated Kafka Streams topics used by the streams app",
                     "title": "Internaltopics"
                 },
@@ -4138,7 +4138,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "List of topics used by the streams app",
                     "title": "Topics"
                 }
diff --git a/docs/docs/schema/pipeline.json b/docs/docs/schema/pipeline.json
index 226f3c99b..4afa89596 100644
--- a/docs/docs/schema/pipeline.json
+++ b/docs/docs/schema/pipeline.json
@@ -295,7 +295,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "List of JMX metric rules.",
                     "title": "Metricrules"
                 },
@@ -600,7 +600,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "matchExpressions is a list of label selector requirements. The requirements are ANDed.",
                     "title": "Matchexpressions"
                 },
@@ -616,7 +616,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "matchLabels is a map of {key,value} pairs. A single {key,value} in the matchLabels map is equivalent to an element of matchExpressions, whose key field is *key*, the operator is *In*, and the values array contains only *value*. The requirements are ANDed.",
                     "title": "Matchlabels"
                 }
@@ -678,7 +678,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "The scheduler will prefer to schedule pods to nodes that satisfy the affinity expressions specified by this field, but it may choose a node that violates one or more of the expressions. The node that is most preferred is the one with the greatest sum of weights, i.e. for each node that meets all of the scheduling requirements (resource request, requiredDuringScheduling affinity expressions, etc.), compute a sum by iterating through the elements of this field and adding *weight* to the sum if the node matches the corresponding matchExpressions; the node(s) with the highest sum are the most preferred.",
                     "title": "Preferredduringschedulingignoredduringexecution"
                 },
@@ -772,7 +772,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "A list of node selector requirements by node's labels.",
                     "title": "Matchexpressions"
                 },
@@ -788,7 +788,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "A list of node selector requirements by node's fields.",
                     "title": "Matchfields"
                 }
@@ -828,7 +828,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "The scheduler will prefer to schedule pods to nodes that satisfy the affinity expressions specified by this field, but it may choose a node that violates one or more of the expressions. The node that is most preferred is the one with the greatest sum of weights, i.e. for each node that meets all of the scheduling requirements (resource request, requiredDuringScheduling affinity expressions, etc.), compute a sum by iterating through the elements of this field and adding weight to the sum if the node has pods which matches the corresponding podAffinityTerm; the node(s) with the highest sum are the most preferred.",
                     "title": "Preferredduringschedulingignoredduringexecution"
                 },
@@ -844,7 +844,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "If the affinity requirements specified by this field are not met at scheduling time, the pod will not be scheduled onto the node. If the affinity requirements specified by this field cease to be met at some point during pod execution (e.g. due to a pod label update), the system may or may not try to eventually evict the pod from its node. When there are multiple elements, the lists of nodes corresponding to each podAffinityTerm are intersected, i.e. all terms must be satisfied.",
                     "title": "Requiredduringschedulingignoredduringexecution"
                 }
@@ -879,7 +879,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "MatchLabelKeys is a set of pod label keys to select which pods will be taken into consideration. The keys are used to lookup values from the incoming pod labels, those key-value labels are merged with `labelSelector` as `key in (value)` to select the group of existing pods which pods will be taken into consideration for the incoming pod's pod (anti) affinity. Keys that don't exist in the incoming pod labels will be ignored. The default value is empty. The same key is forbidden to exist in both matchLabelKeys and labelSelector. Also, matchLabelKeys cannot be set when labelSelector isn't set. This is a beta field and requires enabling MatchLabelKeysInPodAffinity feature gate (enabled by default).",
                     "title": "Matchlabelkeys"
                 },
@@ -895,7 +895,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "MismatchLabelKeys is a set of pod label keys to select which pods will be taken into consideration. The keys are used to lookup values from the incoming pod labels, those key-value labels are merged with `labelSelector` as `key notin (value)` to select the group of existing pods which pods will be taken into consideration for the incoming pod's pod (anti) affinity. Keys that don't exist in the incoming pod labels will be ignored. The default value is empty. The same key is forbidden to exist in both mismatchLabelKeys and labelSelector. Also, mismatchLabelKeys cannot be set when labelSelector isn't set. This is a beta field and requires enabling MatchLabelKeysInPodAffinity feature gate (enabled by default).",
                     "title": "Mismatchlabelkeys"
                 },
@@ -923,7 +923,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "",
                     "title": "Namespaces"
                 },
@@ -1250,7 +1250,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "Array containing taint references. When defined, pods can run on nodes, which would otherwise deny scheduling.",
                     "title": "Tolerations"
                 }
@@ -1312,7 +1312,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Map of command line arguments passed to the streams app.",
                     "title": "Commandline"
                 },
@@ -1354,7 +1354,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Custom environment variables.",
                     "title": "Env"
                 },
@@ -1380,7 +1380,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Map of files to mount for the app. File will be mounted as $value.mountPath/$key. $value.content denotes file content (recommended to be used with --set-file).",
                     "title": "Files"
                 },
@@ -1416,7 +1416,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "",
                     "title": "Imagepullsecrets"
                 },
@@ -1463,7 +1463,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.25/#probe-v1-core",
                     "title": "Livenessprobe"
                 },
@@ -1493,7 +1493,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Map of custom annotations to attach to the pod spec.",
                     "title": "Podannotations"
                 },
@@ -1509,7 +1509,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Map of custom labels to attach to the pod spec.",
                     "title": "Podlabels"
                 },
@@ -1525,7 +1525,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "",
                     "title": "Ports"
                 },
@@ -1538,7 +1538,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.25/#probe-v1-core",
                     "title": "Readinessprobe"
                 },
@@ -1591,7 +1591,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "Mount existing secrets as volumes",
                     "title": "Secretfilesrefs"
                 },
@@ -1604,7 +1604,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Inject existing secrets as environment variables. Map key is used as environment variable name. Value consists of secret name and key.",
                     "title": "Secretrefs"
                 },
@@ -1620,7 +1620,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Custom secret environment variables. Prefix with configurationEnvPrefix in order to pass secrets to command line or prefix with KAFKA_ to pass secrets to Kafka Streams configuration.",
                     "title": "Secrets"
                 },
@@ -1674,7 +1674,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "Array containing taint references. When defined, pods can run on nodes, which would otherwise deny scheduling.",
                     "title": "Tolerations"
                 },
@@ -2403,7 +2403,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "Array containing taint references. When defined, pods can run on nodes, which would otherwise deny scheduling.",
                     "title": "Tolerations"
                 }
@@ -2464,7 +2464,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Map of command line arguments passed to the streams app.",
                     "title": "Commandline"
                 },
@@ -2493,7 +2493,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Custom environment variables.",
                     "title": "Env"
                 },
@@ -2506,7 +2506,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Map of files to mount for the app. File will be mounted as $value.mountPath/$key. $value.content denotes file content (recommended to be used with --set-file).",
                     "title": "Files"
                 },
@@ -2542,7 +2542,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "",
                     "title": "Imagepullsecrets"
                 },
@@ -2601,7 +2601,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.25/#probe-v1-core",
                     "title": "Livenessprobe"
                 },
@@ -2643,7 +2643,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Map of custom annotations to attach to the pod spec.",
                     "title": "Podannotations"
                 },
@@ -2659,7 +2659,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Map of custom labels to attach to the pod spec.",
                     "title": "Podlabels"
                 },
@@ -2675,7 +2675,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "",
                     "title": "Ports"
                 },
@@ -2700,7 +2700,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.25/#probe-v1-core",
                     "title": "Readinessprobe"
                 },
@@ -2728,7 +2728,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "Mount existing secrets as volumes",
                     "title": "Secretfilesrefs"
                 },
@@ -2741,7 +2741,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Inject existing secrets as environment variables. Map key is used as environment variable name. Value consists of secret name and key.",
                     "title": "Secretrefs"
                 },
@@ -2757,7 +2757,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": {},
                     "description": "Custom secret environment variables. Prefix with configurationEnvPrefix in order to pass secrets to command line or prefix with KAFKA_ to pass secrets to Kafka Streams configuration.",
                     "title": "Secrets"
                 },
@@ -2804,7 +2804,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "Array containing taint references. When defined, pods can run on nodes, which would otherwise deny scheduling.",
                     "title": "Tolerations"
                 }
@@ -3084,7 +3084,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "List of additional KEDA triggers, see https://keda.sh/docs/latest/scalers/",
                     "title": "Additionaltriggers"
                 },
@@ -3132,7 +3132,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "List of auto-generated Kafka Streams topics used by the streams app",
                     "title": "Internaltopics"
                 },
@@ -3213,7 +3213,7 @@
                             "type": "null"
                         }
                     ],
-                    "default": null,
+                    "default": [],
                     "description": "List of topics used by the streams app",
                     "title": "Topics"
                 }
diff --git a/kpops/components/common/kubernetes_model.py b/kpops/components/common/kubernetes_model.py
index 81d01e1a7..967182218 100644
--- a/kpops/components/common/kubernetes_model.py
+++ b/kpops/components/common/kubernetes_model.py
@@ -7,7 +7,12 @@
 from pydantic import Field, model_validator
 
 from kpops.utils.docstring import describe_attr
-from kpops.utils.pydantic import CamelCaseConfigModel, DescConfigModel
+from kpops.utils.pydantic import (
+    CamelCaseConfigModel,
+    DescConfigModel,
+    SerializeAsOptional,
+    SerializeAsOptionalModel,
+)
 
 if TYPE_CHECKING:
     try:
@@ -91,18 +96,18 @@ def validate_values(self) -> Self:
         return self
 
 
-class NodeSelectorTerm(DescConfigModel, CamelCaseConfigModel):
+class NodeSelectorTerm(SerializeAsOptionalModel, DescConfigModel, CamelCaseConfigModel):
     """A null or empty node selector term matches no objects. The requirements of them are ANDed. The TopologySelectorTerm type implements a subset of the NodeSelectorTerm.
 
     :param match_expressions: A list of node selector requirements by node's labels.
     :param match_fields: A list of node selector requirements by node's fields.
     """
 
-    match_expressions: list[NodeSelectorRequirement] | None = Field(
-        default=None, description=describe_attr("match_expressions", __doc__)
+    match_expressions: SerializeAsOptional[list[NodeSelectorRequirement]] = Field(
+        default=[], description=describe_attr("match_expressions", __doc__)
     )
-    match_fields: list[NodeSelectorRequirement] | None = Field(
-        default=None, description=describe_attr("match_fields", __doc__)
+    match_fields: SerializeAsOptional[list[NodeSelectorRequirement]] = Field(
+        default=[], description=describe_attr("match_fields", __doc__)
     )
 
 
@@ -130,7 +135,7 @@ class PreferredSchedulingTerm(DescConfigModel, CamelCaseConfigModel):
     weight: Weight = Field(description=describe_attr("weight", __doc__))
 
 
-class NodeAffinity(DescConfigModel, CamelCaseConfigModel):
+class NodeAffinity(SerializeAsOptionalModel, DescConfigModel, CamelCaseConfigModel):
     """Node affinity is a group of node affinity scheduling rules.
 
     :param required_during_scheduling_ignored_during_execution: If the affinity requirements specified by this field are not met at scheduling time, the pod will not be scheduled onto the node. If the affinity requirements specified by this field cease to be met at some point during pod execution (e.g. due to an update), the system may or may not try to eventually evict the pod from its node.
@@ -143,10 +148,10 @@ class NodeAffinity(DescConfigModel, CamelCaseConfigModel):
             "required_during_scheduling_ignored_during_execution", __doc__
         ),
     )
-    preferred_during_scheduling_ignored_during_execution: (
-        list[PreferredSchedulingTerm] | None
-    ) = Field(
-        default=None,
+    preferred_during_scheduling_ignored_during_execution: SerializeAsOptional[
+        list[PreferredSchedulingTerm]
+    ] = Field(
+        default=[],
         description=describe_attr(
             "preferred_during_scheduling_ignored_during_execution", __doc__
         ),
@@ -190,24 +195,24 @@ def validate_values(self) -> Self:
         return self
 
 
-class LabelSelector(DescConfigModel, CamelCaseConfigModel):
+class LabelSelector(SerializeAsOptionalModel, DescConfigModel, CamelCaseConfigModel):
     """A label selector is a label query over a set of resources. The result of matchLabels and matchExpressions are ANDed. An empty label selector matches all objects. A null label selector matches no objects.
 
     :param match_labels: matchLabels is a map of {key,value} pairs. A single {key,value} in the matchLabels map is equivalent to an element of matchExpressions, whose key field is *key*, the operator is *In*, and the values array contains only *value*. The requirements are ANDed.
     :param match_expressions: matchExpressions is a list of label selector requirements. The requirements are ANDed.
     """
 
-    match_labels: dict[str, str] | None = Field(
-        default=None,
+    match_labels: SerializeAsOptional[dict[str, str]] = Field(
+        default={},
         description=describe_attr("match_labels", __doc__),
     )
-    match_expressions: list[LabelSelectorRequirement] | None = Field(
-        default=None,
+    match_expressions: SerializeAsOptional[list[LabelSelectorRequirement]] = Field(
+        default=[],
         description=describe_attr("match_expressions", __doc__),
     )
 
 
-class PodAffinityTerm(DescConfigModel, CamelCaseConfigModel):
+class PodAffinityTerm(SerializeAsOptionalModel, DescConfigModel, CamelCaseConfigModel):
     """Defines a set of pods (namely those matching the labelSelector relative to the given namespace(s)) that this pod should be co-located (affinity) or not co-located (anti-affinity) with, where co-located is defined as running on a node whose value of the label with key <topologyKey> matches that of any node on which a pod of the set of pods is running.
 
     :param label_selector: A label query over a set of resources, in this case pods. If it's null, this PodAffinityTerm matches with no Pods.
@@ -222,19 +227,19 @@ class PodAffinityTerm(DescConfigModel, CamelCaseConfigModel):
         default=None,
         description=describe_attr("label_selector", __doc__),
     )
-    match_label_keys: list[str] | None = Field(
-        default=None,
+    match_label_keys: SerializeAsOptional[list[str]] = Field(
+        default=[],
         description=describe_attr("match_label_keys", __doc__),
     )
-    mismatch_label_keys: list[str] | None = Field(
-        default=None,
+    mismatch_label_keys: SerializeAsOptional[list[str]] = Field(
+        default=[],
         description=describe_attr("mismatch_label_keys", __doc__),
     )
     topology_key: str = Field(
         description=describe_attr("topology_key", __doc__),
     )
-    namespaces: list[str] | None = Field(
-        default=None,
+    namespaces: SerializeAsOptional[list[str]] = Field(
+        default=[],
         description=describe_attr("namespaces", __doc__),
     )
     namespace_selector: LabelSelector | None = Field(
@@ -258,25 +263,25 @@ class WeightedPodAffinityTerm(DescConfigModel, CamelCaseConfigModel):
     )
 
 
-class PodAffinity(DescConfigModel, CamelCaseConfigModel):
+class PodAffinity(SerializeAsOptionalModel, DescConfigModel, CamelCaseConfigModel):
     """Pod affinity is a group of inter pod affinity scheduling rules.
 
     :param required_during_scheduling_ignored_during_execution: If the affinity requirements specified by this field are not met at scheduling time, the pod will not be scheduled onto the node. If the affinity requirements specified by this field cease to be met at some point during pod execution (e.g. due to a pod label update), the system may or may not try to eventually evict the pod from its node. When there are multiple elements, the lists of nodes corresponding to each podAffinityTerm are intersected, i.e. all terms must be satisfied.
     :param preferred_during_scheduling_ignored_during_execution: The scheduler will prefer to schedule pods to nodes that satisfy the affinity expressions specified by this field, but it may choose a node that violates one or more of the expressions. The node that is most preferred is the one with the greatest sum of weights, i.e. for each node that meets all of the scheduling requirements (resource request, requiredDuringScheduling affinity expressions, etc.), compute a sum by iterating through the elements of this field and adding weight to the sum if the node has pods which matches the corresponding podAffinityTerm; the node(s) with the highest sum are the most preferred.
     """
 
-    required_during_scheduling_ignored_during_execution: (
-        list[PodAffinityTerm] | None
-    ) = Field(
-        default=None,
+    required_during_scheduling_ignored_during_execution: SerializeAsOptional[
+        list[PodAffinityTerm]
+    ] = Field(
+        default=[],
         description=describe_attr(
             "required_during_scheduling_ignored_during_execution", __doc__
         ),
     )
-    preferred_during_scheduling_ignored_during_execution: (
-        list[WeightedPodAffinityTerm] | None
-    ) = Field(
-        default=None,
+    preferred_during_scheduling_ignored_during_execution: SerializeAsOptional[
+        list[WeightedPodAffinityTerm]
+    ] = Field(
+        default=[],
         description=describe_attr(
             "preferred_during_scheduling_ignored_during_execution", __doc__
         ),
diff --git a/kpops/components/streams_bootstrap/model.py b/kpops/components/streams_bootstrap/model.py
index 2bff9b6c2..bc70e37e5 100644
--- a/kpops/components/streams_bootstrap/model.py
+++ b/kpops/components/streams_bootstrap/model.py
@@ -11,6 +11,7 @@
     ImagePullPolicy,
     ProtocolSchema,
     Resources,
+    SerializeAsOptional,
     ServiceType,
     Toleration,
 )
@@ -19,6 +20,7 @@
 from kpops.utils.pydantic import (
     CamelCaseConfigModel,
     DescConfigModel,
+    SerializeAsOptionalModel,
     exclude_by_value,
     exclude_defaults,
 )
@@ -93,7 +95,7 @@ class JavaOptions(CamelCaseConfigModel, DescConfigModel):
     )
 
 
-class StreamsBootstrapValues(HelmAppValues):
+class StreamsBootstrapValues(SerializeAsOptionalModel, HelmAppValues):
     """Base value class for all streams bootstrap related components.
 
     :param image: Docker image of the Kafka producer app.
@@ -132,8 +134,8 @@ class StreamsBootstrapValues(HelmAppValues):
         description=describe_attr("image_pull_policy", __doc__),
     )
 
-    image_pull_secrets: list[dict[str, str]] | None = Field(
-        default=None,
+    image_pull_secrets: SerializeAsOptional[list[dict[str, str]]] = Field(
+        default=[],
         description=describe_attr("image_pull_secret", __doc__),
     )
 
@@ -146,8 +148,8 @@ class StreamsBootstrapValues(HelmAppValues):
         description=describe_attr("resources", __doc__),
     )
 
-    ports: list[PortConfig] | None = Field(
-        default=None,
+    ports: SerializeAsOptional[list[PortConfig]] = Field(
+        default=[],
         description=describe_attr("ports", __doc__),
     )
 
@@ -161,33 +163,33 @@ class StreamsBootstrapValues(HelmAppValues):
         description=describe_attr("configuration_env_prefix", __doc__),
     )
 
-    command_line: dict[str, str | bool | int] | None = Field(
-        default=None,
+    command_line: SerializeAsOptional[dict[str, str | bool | int]] = Field(
+        default={},
         description=describe_attr("command_line", __doc__),
     )
 
-    env: dict[str, str] | None = Field(
-        default=None,
+    env: SerializeAsOptional[dict[str, str]] = Field(
+        default={},
         description=describe_attr("env", __doc__),
     )
 
-    secrets: dict[str, str] | None = Field(
-        default=None,
+    secrets: SerializeAsOptional[dict[str, str]] = Field(
+        default={},
         description=describe_attr("secrets", __doc__),
     )
 
-    secret_refs: dict[str, Any] | None = Field(
-        default=None,
+    secret_refs: SerializeAsOptional[dict[str, Any]] = Field(
+        default={},
         description=describe_attr("secret_refs", __doc__),
     )
 
-    secret_files_refs: list[str] | None = Field(
-        default=None,
+    secret_files_refs: SerializeAsOptional[list[str]] = Field(
+        default=[],
         description=describe_attr("secret_files_refs", __doc__),
     )
 
-    files: dict[str, Any] | None = Field(
-        default=None,
+    files: SerializeAsOptional[dict[str, Any]] = Field(
+        default={},
         description=describe_attr("files", __doc__),
     )
 
@@ -196,23 +198,23 @@ class StreamsBootstrapValues(HelmAppValues):
         description=describe_attr("java_options", __doc__),
     )
 
-    pod_annotations: dict[str, str] | None = Field(
-        default=None,
+    pod_annotations: SerializeAsOptional[dict[str, str]] = Field(
+        default={},
         description=describe_attr("pod_annotations", __doc__),
     )
 
-    pod_labels: dict[str, str] | None = Field(
-        default=None,
+    pod_labels: SerializeAsOptional[dict[str, str]] = Field(
+        default={},
         description=describe_attr("pod_labels", __doc__),
     )
 
-    liveness_probe: dict[str, Any] | None = Field(
-        default=None,
+    liveness_probe: SerializeAsOptional[dict[str, Any]] = Field(
+        default={},
         description=describe_attr("liveness_probe", __doc__),
     )
 
-    readiness_probe: dict[str, Any] | None = Field(
-        default=None,
+    readiness_probe: SerializeAsOptional[dict[str, Any]] = Field(
+        default={},
         description=describe_attr("readiness_probe", __doc__),
     )
 
@@ -221,8 +223,8 @@ class StreamsBootstrapValues(HelmAppValues):
         description=describe_attr("affinity", __doc__),
     )
 
-    tolerations: list[Toleration] | None = Field(
-        default=None,
+    tolerations: SerializeAsOptional[list[Toleration]] = Field(
+        default=[],
         description=describe_attr("tolerations", __doc__),
     )
 
diff --git a/kpops/components/streams_bootstrap/producer/producer_app.py b/kpops/components/streams_bootstrap/producer/producer_app.py
index 44fe2e592..be445079d 100644
--- a/kpops/components/streams_bootstrap/producer/producer_app.py
+++ b/kpops/components/streams_bootstrap/producer/producer_app.py
@@ -63,9 +63,12 @@ def is_cron_job(self) -> bool:
     @computed_field
     @cached_property
     def _cleaner(self) -> ProducerAppCleaner:
-        return ProducerAppCleaner(
-            **self.model_dump(by_alias=True, exclude={"_cleaner", "from_", "to"})
-        )
+        kwargs = {
+            name: getattr(self, name)
+            for name in self.model_fields_set
+            if name not in {"_cleaner", "from_", "to"}
+        }
+        return ProducerAppCleaner.model_validate(kwargs)
 
     @override
     def apply_to_outputs(self, name: str, topic: TopicConfig) -> None:
diff --git a/kpops/components/streams_bootstrap/streams/model.py b/kpops/components/streams_bootstrap/streams/model.py
index 63fe1f10e..c9626a22f 100644
--- a/kpops/components/streams_bootstrap/streams/model.py
+++ b/kpops/components/streams_bootstrap/streams/model.py
@@ -8,6 +8,7 @@
 from kpops.components.common.kubernetes_model import (
     ImagePullPolicy,
     Resources,
+    SerializeAsOptional,
 )
 from kpops.components.common.topic import KafkaTopic, KafkaTopicStr
 from kpops.components.streams_bootstrap.model import (
@@ -18,6 +19,7 @@
 from kpops.utils.pydantic import (
     CamelCaseConfigModel,
     DescConfigModel,
+    SerializeAsOptionalModel,
 )
 
 
@@ -117,7 +119,9 @@ def add_labeled_input_topics(self, label: str, topics: list[KafkaTopic]) -> None
         )
 
 
-class StreamsAppAutoScaling(CamelCaseConfigModel, DescConfigModel):
+class StreamsAppAutoScaling(
+    SerializeAsOptionalModel, CamelCaseConfigModel, DescConfigModel
+):
     """Kubernetes Event-driven Autoscaling config.
 
     :param enabled: Whether to enable auto-scaling using KEDA., defaults to False
@@ -191,16 +195,16 @@ class StreamsAppAutoScaling(CamelCaseConfigModel, DescConfigModel):
         title="Idle replica count",
         description=describe_attr("idle_replicas", __doc__),
     )
-    internal_topics: list[str] | None = Field(
-        default=None,
+    internal_topics: SerializeAsOptional[list[str]] = Field(
+        default=[],
         description=describe_attr("internal_topics", __doc__),
     )
-    topics: list[str] | None = Field(
-        default=None,
+    topics: SerializeAsOptional[list[str]] = Field(
+        default=[],
         description=describe_attr("topics", __doc__),
     )
-    additional_triggers: list[str] | None = Field(
-        default=None,
+    additional_triggers: SerializeAsOptional[list[str]] = Field(
+        default=[],
         description=describe_attr("additional_triggers", __doc__),
     )
     model_config = ConfigDict(extra="allow")
@@ -277,7 +281,7 @@ class PrometheusJMXExporterConfig(CamelCaseConfigModel, DescConfigModel):
     )
 
 
-class JMXConfig(CamelCaseConfigModel, DescConfigModel):
+class JMXConfig(SerializeAsOptionalModel, CamelCaseConfigModel, DescConfigModel):
     """JMX configuration options.
 
     :param port: The jmx port which JMX style metrics are exposed.
@@ -289,8 +293,8 @@ class JMXConfig(CamelCaseConfigModel, DescConfigModel):
         description=describe_attr("port", __doc__),
     )
 
-    metric_rules: list[str] | None = Field(
-        default=None,
+    metric_rules: SerializeAsOptional[list[str]] = Field(
+        default=[],
         description=describe_attr("metric_rules", __doc__),
     )
 
diff --git a/kpops/components/streams_bootstrap/streams/streams_app.py b/kpops/components/streams_bootstrap/streams/streams_app.py
index c0a21e7b6..6e5d8c037 100644
--- a/kpops/components/streams_bootstrap/streams/streams_app.py
+++ b/kpops/components/streams_bootstrap/streams/streams_app.py
@@ -67,9 +67,12 @@ class StreamsApp(StreamsBootstrap):
     @computed_field
     @cached_property
     def _cleaner(self) -> StreamsAppCleaner:
-        return StreamsAppCleaner(
-            **self.model_dump(by_alias=True, exclude={"_cleaner", "from_", "to"})
-        )
+        kwargs = {
+            name: getattr(self, name)
+            for name in self.model_fields_set
+            if name not in {"_cleaner", "from_", "to"}
+        }
+        return StreamsAppCleaner.model_validate(kwargs)
 
     @property
     @override
diff --git a/kpops/components/streams_bootstrap_v2/base.py b/kpops/components/streams_bootstrap_v2/base.py
index 27fc1e9b6..2fa44354b 100644
--- a/kpops/components/streams_bootstrap_v2/base.py
+++ b/kpops/components/streams_bootstrap_v2/base.py
@@ -17,6 +17,8 @@
 from kpops.utils.pydantic import (
     CamelCaseConfigModel,
     DescConfigModel,
+    SerializeAsOptional,
+    SerializeAsOptionalModel,
     exclude_by_value,
     exclude_defaults,
 )
@@ -97,7 +99,7 @@ def serialize_model(
         )
 
 
-class StreamsBootstrapV2Values(HelmAppValues):
+class StreamsBootstrapV2Values(SerializeAsOptionalModel, HelmAppValues):
     """Base value class for all streams bootstrap v2 related components.
 
     :param image_tag: Docker image tag of the streams-bootstrap-v2 app.
@@ -120,8 +122,8 @@ class StreamsBootstrapV2Values(HelmAppValues):
         description=describe_attr("affinity", __doc__),
     )
 
-    tolerations: list[Toleration] | None = Field(
-        default=None,
+    tolerations: SerializeAsOptional[list[Toleration]] = Field(
+        default=[],
         description=describe_attr("tolerations", __doc__),
     )
 
diff --git a/kpops/utils/pydantic.py b/kpops/utils/pydantic.py
index 3b3fb28ae..47293bdad 100644
--- a/kpops/utils/pydantic.py
+++ b/kpops/utils/pydantic.py
@@ -1,11 +1,21 @@
 import json
 import logging
 from pathlib import Path
-from typing import Any
+from typing import Annotated, Any
 
 import humps
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import (
+    BaseModel,
+    BeforeValidator,
+    ConfigDict,
+    Field,
+    GetCoreSchemaHandler,
+    SerializationInfo,
+    SerializerFunctionWrapHandler,
+    model_serializer,
+)
 from pydantic.fields import FieldInfo
+from pydantic_core import PydanticUseDefault, core_schema
 from pydantic_settings import PydanticBaseSettingsSource
 from typing_extensions import TypeVar, override
 
@@ -224,3 +234,65 @@ def __call__(self) -> dict[str, Any]:
             if field_value is not None:
                 d[field_key] = field_value
         return d
+
+
+_T = TypeVar("_T")
+
+
+def validate_optional_to_default(value: Any | None) -> Any:
+    if value is None:
+        raise PydanticUseDefault
+    return value
+
+
+def serialize_to_optional(
+    value: _T,
+    default_serialize_handler: SerializerFunctionWrapHandler,
+    # info: SerializationInfo,
+) -> _T | None:
+    return default_serialize_handler(value) or None
+    # TODO: another potential solution, depends on https://github.com/pydantic/pydantic/issues/6969
+    # if not result and info.exclude_none:
+    #     raise PydanticOmit
+
+
+class WrapNullableSchema:
+    def __get_pydantic_core_schema__(
+        self,
+        source: type[Any],
+        handler: GetCoreSchemaHandler,
+    ) -> core_schema.CoreSchema:
+        schema = handler(source)
+        # wrap generated schema in nullable
+        return core_schema.NullableSchema(
+            type="nullable",
+            schema=schema,
+            serialization=core_schema.wrap_serializer_function_ser_schema(
+                serialize_to_optional,
+                schema=core_schema.nullable_schema(schema),
+            ),
+        )
+
+
+SerializeAsOptional = Annotated[
+    _T,
+    WrapNullableSchema(),
+    BeforeValidator(validate_optional_to_default),
+    "Optional that is serialized to `None` if falsy",
+    "similarly an input of `None` is translated to its default during validation",
+    "requires inheriting from SerializeAsOptionalModel for `model_dump(exclude_none=True)` to work",
+]
+
+
+class SerializeAsOptionalModel(BaseModel):
+    # HACK: workaround for exclude_none, which is otherwise evaluated too early
+    @model_serializer(mode="wrap", when_used="always")
+    def serialize_model(
+        self,
+        default_serialize_handler: SerializerFunctionWrapHandler,
+        info: SerializationInfo,
+    ) -> dict[str, Any]:
+        result = default_serialize_handler(self)
+        if info.exclude_none:
+            return exclude_by_value(result, None)
+        return result
diff --git a/tests/components/streams_bootstrap/test_streams_bootstrap.py b/tests/components/streams_bootstrap/test_streams_bootstrap.py
index 68c25b66c..de461dc1d 100644
--- a/tests/components/streams_bootstrap/test_streams_bootstrap.py
+++ b/tests/components/streams_bootstrap/test_streams_bootstrap.py
@@ -12,7 +12,12 @@
     HelmUpgradeInstallFlags,
 )
 from kpops.component_handlers.helm_wrapper.utils import create_helm_release_name
-from kpops.components.common.kubernetes_model import ResourceDefinition
+from kpops.components.common.kubernetes_model import (
+    NodeAffinity,
+    NodeSelectorTerm,
+    PreferredSchedulingTerm,
+    ResourceDefinition,
+)
 from kpops.components.streams_bootstrap.base import StreamsBootstrap
 from kpops.components.streams_bootstrap.model import StreamsBootstrapValues
 
@@ -227,3 +232,24 @@ def test_resource_definition(
     ):
         with expectation:
             assert ResourceDefinition.model_validate(input)
+
+    def test_node_affinity(self):
+        node_affinity = NodeAffinity()
+        assert node_affinity.preferred_during_scheduling_ignored_during_execution == []
+        assert node_affinity.model_dump(by_alias=True) == {
+            "requiredDuringSchedulingIgnoredDuringExecution": None,
+            "preferredDuringSchedulingIgnoredDuringExecution": None,
+        }
+        assert node_affinity.model_dump(by_alias=True, exclude_none=True) == {}
+
+        node_affinity.preferred_during_scheduling_ignored_during_execution.append(
+            PreferredSchedulingTerm(preference=NodeSelectorTerm(), weight=1)
+        )
+        assert node_affinity.model_dump(by_alias=True, exclude_none=True) == {
+            "preferredDuringSchedulingIgnoredDuringExecution": [
+                {
+                    "preference": {},
+                    "weight": 1,
+                }
+            ],
+        }
diff --git a/tests/pipeline/resources/streams-bootstrap/pipeline.yaml b/tests/pipeline/resources/streams-bootstrap/pipeline.yaml
index ad6308a4a..b2c9fae90 100644
--- a/tests/pipeline/resources/streams-bootstrap/pipeline.yaml
+++ b/tests/pipeline/resources/streams-bootstrap/pipeline.yaml
@@ -20,6 +20,19 @@
     imageTag: "1.0.0"
     kafka:
       applicationId: "my-streams-app-id"
+    affinity:
+      nodeAffinity:
+          preferredDuringSchedulingIgnoredDuringExecution:
+            - preference:
+                matchExpressions:
+                - key: foo
+                  operator: Exists
+              weight: 2
+            - preference:
+                matchExpressions:
+                - key: bar
+                  operator: DoesNotExist
+              weight: 1
     commandLine:
       CONVERT_XML: true
     javaOptions:
diff --git a/tests/pipeline/snapshots/test_generate/test_streams_bootstrap/pipeline.yaml b/tests/pipeline/snapshots/test_generate/test_streams_bootstrap/pipeline.yaml
index d72426d8d..4741ad72f 100644
--- a/tests/pipeline/snapshots/test_generate/test_streams_bootstrap/pipeline.yaml
+++ b/tests/pipeline/snapshots/test_generate/test_streams_bootstrap/pipeline.yaml
@@ -85,6 +85,21 @@
     suffix: -clean
     type: streams-app-cleaner
     values:
+      affinity:
+        nodeAffinity:
+          preferredDuringSchedulingIgnoredDuringExecution:
+          - preference:
+              matchExpressions:
+              - key: foo
+                operator: Exists
+                values: []
+            weight: 2
+          - preference:
+              matchExpressions:
+              - key: bar
+                operator: DoesNotExist
+                values: []
+            weight: 1
       commandLine:
         CONVERT_XML: true
       files:
@@ -162,6 +177,21 @@
         value_schema: com.bakdata.kafka.DeadLetter
   type: my-streams-app
   values:
+    affinity:
+      nodeAffinity:
+        preferredDuringSchedulingIgnoredDuringExecution:
+        - preference:
+            matchExpressions:
+            - key: foo
+              operator: Exists
+              values: []
+          weight: 2
+        - preference:
+            matchExpressions:
+            - key: bar
+              operator: DoesNotExist
+              values: []
+          weight: 1
     commandLine:
       CONVERT_XML: true
     files:
diff --git a/tests/pipeline/snapshots/test_manifest/test_streams_bootstrap/manifest.yaml b/tests/pipeline/snapshots/test_manifest/test_streams_bootstrap/manifest.yaml
index 075777bf9..84354ab8e 100644
--- a/tests/pipeline/snapshots/test_manifest/test_streams_bootstrap/manifest.yaml
+++ b/tests/pipeline/snapshots/test_manifest/test_streams_bootstrap/manifest.yaml
@@ -124,6 +124,19 @@ spec:
         app: resources-streams-bootstrap-my-streams-app
         release: resources-streams-bootstrap-my-streams-app
     spec:
+      affinity:
+        nodeAffinity:
+          preferredDuringSchedulingIgnoredDuringExecution:
+          - preference:
+              matchExpressions:
+              - key: foo
+                operator: Exists
+            weight: 2
+          - preference:
+              matchExpressions:
+              - key: bar
+                operator: DoesNotExist
+            weight: 1
       containers:
       - env:
         - name: ENV_PREFIX
diff --git a/tests/utils/test_pydantic.py b/tests/utils/test_pydantic.py
index 525db63b0..5d6be4a66 100644
--- a/tests/utils/test_pydantic.py
+++ b/tests/utils/test_pydantic.py
@@ -2,7 +2,15 @@
 
 import pytest
 
-from kpops.utils.pydantic import to_dash, to_dot, to_snake, to_str
+from kpops.components.common.kubernetes_model import SerializeAsOptional
+from kpops.utils.pydantic import (
+    SerializeAsOptionalModel,
+    exclude_by_value,
+    to_dash,
+    to_dot,
+    to_snake,
+    to_str,
+)
 
 
 @pytest.mark.parametrize(
@@ -64,3 +72,100 @@ def test_to_dot(input: str, expected: str):
 )
 def test_to_str(input: Any, expected: str):
     assert to_str(input) == expected
+
+
+@pytest.mark.parametrize(
+    ("dumped_model", "excluded_values", "expected"),
+    [
+        pytest.param(
+            {},
+            (),
+            {},
+        ),
+        pytest.param(
+            {},
+            (None,),
+            {},
+        ),
+        pytest.param(
+            {"foo": 0, "bar": 1},
+            (),
+            {"foo": 0, "bar": 1},
+        ),
+        pytest.param(
+            {"foo": 0, "bar": 1},
+            (None,),
+            {"foo": 0, "bar": 1},
+        ),
+        pytest.param(
+            {"foo": 0, "bar": 1},
+            (0,),
+            {"bar": 1},
+        ),
+        pytest.param(
+            {"foo": 0, "bar": 1},
+            (1,),
+            {"foo": 0},
+        ),
+        pytest.param(
+            {"foo": None},
+            (None,),
+            {},
+        ),
+        pytest.param(
+            {"foo": None, "bar": 0},
+            (None,),
+            {"bar": 0},
+        ),
+        pytest.param(
+            {"foo": None, "bar": 0},
+            (None, 0),
+            {},
+        ),
+    ],
+)
+def test_exclude_by_value(
+    dumped_model: dict[str, Any],
+    excluded_values: tuple[Any, ...],
+    expected: dict[str, Any],
+):
+    assert exclude_by_value(dumped_model, *excluded_values) == expected
+
+
+def test_serialize_as_optional():
+    class Model(SerializeAsOptionalModel):
+        optional_list: SerializeAsOptional[list[str]] = []
+        optional_dict: SerializeAsOptional[dict[str, str]] = {}
+
+    model = Model()
+    assert model.optional_list == []
+    assert model.optional_dict == {}
+    assert model.model_dump() == {"optional_list": None, "optional_dict": None}
+    assert model.model_dump(exclude_defaults=True) == {}
+    assert model.model_dump(exclude_unset=True) == {}
+    # this would fail without inheriting from SerializeAsOptionalModel
+    assert model.model_dump(exclude_none=True) == {}
+
+    model = Model.model_validate({"optional_list": None, "optional_dict": None})
+    assert model.optional_list == []
+    assert model.optional_dict == {}
+
+    model = Model(optional_list=["el"], optional_dict={"foo": "bar"})
+    assert model.optional_list == ["el"]
+    assert model.optional_dict == {"foo": "bar"}
+    assert model.model_dump() == {
+        "optional_list": ["el"],
+        "optional_dict": {"foo": "bar"},
+    }
+    assert model.model_dump(exclude_defaults=True) == {
+        "optional_list": ["el"],
+        "optional_dict": {"foo": "bar"},
+    }
+    assert model.model_dump(exclude_unset=True) == {
+        "optional_list": ["el"],
+        "optional_dict": {"foo": "bar"},
+    }
+    assert model.model_dump(exclude_none=True) == {
+        "optional_list": ["el"],
+        "optional_dict": {"foo": "bar"},
+    }