revert: fix: orchestrator shouldn't fail fast on ReplicationTask error (#16969)

Author: gosusnp

airbyte-container-orchestrator/src/main/kotlin/io/airbyte/container/orchestrator/worker/ReplicationTask.kt

@@ -57,7 +57,6 @@ class DestinationReader(
       }
     } catch (e: Exception) {
       logger.error(e) { "DestinationReader error: " }
-      replicationWorkerState.abort()
       if (e is DestinationException) {
         throw e
       } else if (e !is CancellationException) {
@@ -105,7 +104,6 @@ class DestinationWriter(
       }
     } catch (e: Exception) {
       logger.error(e) { "DestinationWriter error: " }
-      replicationWorkerState.abort()
       handleException(e)
     } finally {
       notifyEndOfInput()
@@ -118,7 +116,6 @@ class DestinationWriter(
     try {
       destination.notifyEndOfInput()
     } catch (e: Exception) {
-      replicationWorkerState.abort()
       handleException(e)
     }
   }
@@ -210,7 +207,6 @@ class SourceReader(
       }
     } catch (e: Exception) {
       logger.error(e) { "SourceReader error: " }
-      replicationWorkerState.abort()
       if (e is SourceException) {
         throw e
       } else if (e !is CancellationException) {

airbyte-container-orchestrator/src/main/kotlin/io/airbyte/container/orchestrator/worker/ReplicationWorker.kt

@@ -25,7 +25,6 @@ import kotlinx.coroutines.awaitAll
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.slf4j.MDCContext
-import kotlinx.coroutines.supervisorScope
 import kotlinx.coroutines.withContext
 import org.slf4j.MDC
 import java.nio.file.Path
@@ -146,18 +145,13 @@ class ReplicationWorker(
     dispatcher: ExecutorCoroutineDispatcher,
     mdc: Map<String, String>,
   ) {
-    // Use supervisorScope instead of coroutineScope so that a failure in one job
-    // doesn't immediately cancel all the other jobs.
-    // This allows other jobs to terminate gracefully, which is necessary
-    // to avoid dropping trace messages.
-    // In particular, if the source connector emits a trace and then immediately exits,
-    // we want to allow the MessageProcessor tak to read that trace message.
-    // Tasks are responsible for calling replicationWorkerState.abort()
-    // to signal other tasks to terminate.
-    supervisorScope {
-      syncReplicationJobs.map { job ->
-        AsyncUtils.runAsync(dispatcher, this, mdc) { job.run() }
-      }
-    }.awaitAll()
+    coroutineScope {
+      val tasks =
+        syncReplicationJobs.map { job ->
+          AsyncUtils.runAsync(dispatcher, this, mdc) { job.run() }
+        }
+
+      tasks.awaitAll()
+    }
   }
 }