chore: make resource error as external error message (#13971)

Author: subodh1810

airbyte-commons-worker/src/main/java/io/airbyte/workers/helper/FailureHelper.java

@@ -14,6 +14,7 @@
 import io.airbyte.config.Metadata;
 import io.airbyte.config.StreamDescriptor;
 import io.airbyte.protocol.models.AirbyteTraceMessage;
+import io.airbyte.workers.exception.ResourceConstraintException;
 import io.airbyte.workers.exception.WorkloadLauncherException;
 import io.airbyte.workers.exception.WorkloadMonitorException;
 import java.util.Comparator;
@@ -294,6 +295,11 @@ public static FailureReason checkFailure(final Throwable t,
   public static FailureReason replicationFailure(final Throwable t, final Long jobId, final Integer attemptNumber) {
     final FailureReason failure = genericFailure(t, jobId, attemptNumber)
         .withFailureOrigin(FailureOrigin.REPLICATION);
+    if (isInstanceOf(t, ResourceConstraintException.class)) {
+      return failure.withFailureType(FailureType.TRANSIENT_ERROR)
+          .withExternalMessage("Airbyte could not start the sync process."
+              + " This may be due to insufficient system resources. Please check available resources and try again.");
+    }
     if (isInstanceOf(t, WorkloadLauncherException.class)) {
       return failure.withFailureType(FailureType.TRANSIENT_ERROR)
           .withExternalMessage("Airbyte could not start the sync process.");

airbyte-workload-launcher/src/main/kotlin/pods/KubeClientException.kt renamed to airbyte-commons-worker/src/main/kotlin/io/airbyte/workers/exception/KubeClientException.kt

@@ -1,6 +1,6 @@
-package io.airbyte.workload.launcher.pods
+package io.airbyte.workers.exception
 
-class KubeClientException(
+open class KubeClientException(
   override val message: String,
   override val cause: Throwable,
   val commandType: KubeCommandType,

airbyte-commons-worker/src/main/kotlin/io/airbyte/workers/exception/ResourceConstraintException.kt

@@ -0,0 +1,4 @@
+package io.airbyte.workers.exception
+
+class ResourceConstraintException(message: String, cause: Throwable, commandType: KubeCommandType, podType: PodType? = null) :
+  KubeClientException(message, cause, commandType, podType)

airbyte-workload-launcher/src/main/kotlin/pipeline/handlers/FailureHandler.kt

@@ -6,13 +6,13 @@ package io.airbyte.workload.launcher.pipeline.handlers
 
 import io.airbyte.metrics.lib.ApmTraceUtils
 import io.airbyte.metrics.lib.MetricAttribute
+import io.airbyte.workers.exception.KubeClientException
 import io.airbyte.workload.launcher.client.WorkloadApiClient
 import io.airbyte.workload.launcher.metrics.CustomMetricPublisher
 import io.airbyte.workload.launcher.metrics.MeterFilterFactory
 import io.airbyte.workload.launcher.metrics.WorkloadLauncherMetricMetadata
 import io.airbyte.workload.launcher.pipeline.stages.model.LaunchStageIO
 import io.airbyte.workload.launcher.pipeline.stages.model.StageError
-import io.airbyte.workload.launcher.pods.KubeClientException
 import io.github.oshai.kotlinlogging.KotlinLogging
 import io.github.oshai.kotlinlogging.withLoggingContext
 import jakarta.inject.Named

airbyte-workload-launcher/src/main/kotlin/pods/KubePodClient.kt

@@ -10,6 +10,10 @@ import io.airbyte.featureflag.Multi
 import io.airbyte.featureflag.Workspace
 import io.airbyte.metrics.lib.ApmTraceUtils
 import io.airbyte.persistence.job.models.ReplicationInput
+import io.airbyte.workers.exception.KubeClientException
+import io.airbyte.workers.exception.KubeCommandType
+import io.airbyte.workers.exception.PodType
+import io.airbyte.workers.exception.ResourceConstraintException
 import io.airbyte.workers.models.CheckConnectionInput
 import io.airbyte.workers.models.DiscoverCatalogInput
 import io.airbyte.workers.models.SpecInput
@@ -265,7 +269,7 @@ class KubePodClient(
       when (e) {
         is TimeoutException, is KubernetesClientTimeoutException -> {
           ApmTraceUtils.addExceptionToTrace(e)
-          throw KubeClientException(
+          throw ResourceConstraintException(
             "Unable to start the $podLogLabel pod. This may be due to insufficient system resources. Please check available resources and try again.",
             e,
             KubeCommandType.WAIT_INIT,

airbyte-workload-launcher/src/test/kotlin/pods/KubePodClientTest.kt

@@ -15,6 +15,9 @@ import io.airbyte.featureflag.TestClient
 import io.airbyte.persistence.job.models.IntegrationLauncherConfig
 import io.airbyte.persistence.job.models.JobRunConfig
 import io.airbyte.persistence.job.models.ReplicationInput
+import io.airbyte.workers.exception.KubeClientException
+import io.airbyte.workers.exception.KubeCommandType
+import io.airbyte.workers.exception.PodType
 import io.airbyte.workers.models.CheckConnectionInput
 import io.airbyte.workers.models.DiscoverCatalogInput
 import io.airbyte.workers.models.SpecInput