fix(workload-launcher): SocketException handling

airbyte-workload-launcher/src/main/kotlin/ClaimedProcessor.kt

@@ -9,6 +9,7 @@ import datadog.trace.api.Trace
 import dev.failsafe.Failsafe
 import dev.failsafe.FailsafeException
 import dev.failsafe.RetryPolicy
+import dev.failsafe.function.CheckedSupplier
 import io.airbyte.metrics.lib.ApmTraceUtils
 import io.airbyte.metrics.lib.MetricAttribute
 import io.airbyte.workload.api.client.WorkloadApiClient
@@ -32,6 +33,7 @@ import reactor.core.publisher.Mono
 import reactor.core.scheduler.Schedulers
 import reactor.kotlin.core.publisher.toFlux
 import java.net.ConnectException
+import java.net.SocketException
 import java.net.SocketTimeoutException
 import java.time.Duration
 
@@ -94,26 +96,33 @@ class ClaimedProcessor(
   private fun getWorkloadList(workloadListRequest: WorkloadListRequest): WorkloadListResponse {
     while (true) {
       try {
-        // TODO: consider tuning the retry policy here, since we currently get the default 2 retries.
         return Failsafe.with(
           RetryPolicy.builder<Any>()
             .withBackoff(Duration.ofSeconds(20), Duration.ofDays(365))
             .onRetry { logger.error { "Retrying to fetch workloads for dataplane $dataplaneId" } }
             .abortOn { exception ->
               when (exception) {
-                // This makes us to retry only on 5XX errors
                 is ServerException -> exception.statusCode / 100 != 5
+                is SocketException -> false
                 else -> true
               }
             }
             .build(),
+        ).get<WorkloadListResponse>(
+          CheckedSupplier {
+            apiClient.workloadApi.workloadList(workloadListRequest)
+          },
         )
-          .get { -> apiClient.workloadApi.workloadList(workloadListRequest) }
       } catch (e: FailsafeException) {
+        if (e.cause is SocketException) {
+          // Directly handle and propagate the SocketException
+          logger.error { "Operation not permitted: ${e.cause?.message}" }
+          throw e // Surface the SocketException immediately
+        }
         if (e.cause !is ConnectException && e.cause !is SocketTimeoutException) {
-          throw e; // Surface all other errors.
+          throw e // Surface all other errors
         }
-        // On a ConnectionException or SocketTimeoutException, we'll retry indefinitely.
+        // Continue retrying for connection timeouts and connection exceptions
         logger.warn { "Failed to connect to workload API fetching workloads for dataplane $dataplaneId, retrying..." }
       }
     }