fix: use configured object mapper and executor service (#14917)

airbyte-workload-launcher/src/main/kotlin/config/ContainerConfigBeanFactory.kt

@@ -4,6 +4,7 @@
 
 package io.airbyte.workload.launcher.config
 
+import com.fasterxml.jackson.databind.ObjectMapper
 import io.airbyte.commons.workers.config.WorkerConfigs
 import io.airbyte.commons.workers.config.WorkerConfigsProvider
 import io.airbyte.config.ResourceRequirements
@@ -13,21 +14,35 @@ import io.fabric8.kubernetes.api.model.Toleration
 import io.fabric8.kubernetes.api.model.TolerationBuilder
 import io.fabric8.kubernetes.client.KubernetesClient
 import io.fabric8.kubernetes.client.KubernetesClientBuilder
+import io.fabric8.kubernetes.client.utils.KubernetesSerialization
 import io.micronaut.context.annotation.Factory
 import io.micronaut.context.annotation.Value
 import io.micronaut.core.util.StringUtils
 import jakarta.inject.Named
 import jakarta.inject.Singleton
+import java.util.concurrent.ExecutorService
 
 /**
  * Micronaut bean factory for container configuration-related singletons.
  */
 @Factory
 class ContainerConfigBeanFactory {
   @Singleton
-  fun kubeClient(customOkHttpClientFactory: CustomOkHttpClientFactory): KubernetesClient {
+  fun kubeSerialization(objectMapper: ObjectMapper): KubernetesSerialization {
+    return KubernetesSerialization(objectMapper, true)
+  }
+
+  @Singleton
+  fun kubeClient(
+    customOkHttpClientFactory: CustomOkHttpClientFactory,
+    kubernetesSerialization: KubernetesSerialization,
+    // Configured in application.yml under micronaut.executors.kube-client
+    @Named("kube-client") executorService: ExecutorService,
+  ): KubernetesClient {
     return KubernetesClientBuilder()
       .withHttpClientFactory(customOkHttpClientFactory)
+      .withKubernetesSerialization(kubernetesSerialization)
+      .withTaskExecutor(executorService)
       .build()
   }
 

airbyte-workload-launcher/src/main/resources/application.yml

@@ -23,6 +23,11 @@ micronaut:
     network-policy-label-cache:
       expire-after-write: 10m
       maximum-size: 100_000
+  executors:
+    kube-client:
+      type: cached
+      n-threads: ${WORKLOAD_LAUNCHER_PARALLELISM:10}
+      core-pool-size: 0
 airbyte:
   connector:
     source: