fix: connector commands atomically create and populate exit code file…

…. (#14049)
airbytehq · Sep 19, 2024 · 4430a18 · 4430a18
1 parent df84807
commit 4430a18
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 2 deletions.
diff --git a/...te-commons-worker/src/main/java/io/airbyte/workers/general/BufferedReplicationWorker.java b/...te-commons-worker/src/main/java/io/airbyte/workers/general/BufferedReplicationWorker.java
@@ -463,7 +463,21 @@ private void readFromDestination() {
 
     LOGGER.info("readFromDestination: start");
     try {
-      while (!replicationWorkerHelper.getShouldAbort() && !writeToDestFailed && !(destinationIsFinished = destinationIsFinished())) {
+      while (true) {
+        if (replicationWorkerHelper.getShouldAbort()) {
+          LOGGER.info("Abort signaled — exiting read dest...");
+          break;
+        }
+        if (writeToDestFailed) {
+          LOGGER.info("Write to destination failed — exiting read dest...");
+          break;
+        }
+        destinationIsFinished = destinationIsFinished();
+        if (destinationIsFinished) {
+          LOGGER.info("Destination finished successfully — exiting read dest...");
+          break;
+        }
+
         final Optional<AirbyteMessage> messageOptional;
         try (final var t = readFromDestStopwatch.start()) {
           messageOptional = destination.attemptRead();

diff --git a/airbyte-workload-launcher/src/main/kotlin/pods/factories/ContainerCommandFactory.kt b/airbyte-workload-launcher/src/main/kotlin/pods/factories/ContainerCommandFactory.kt
@@ -97,7 +97,9 @@ object ContainerCommandFactory {
     EXIT_CODE=$?
 
     # write its exit code to a file for the sidecar
-    echo ${'$'}EXIT_CODE > $EXIT_CODE_FILE
+    echo ${'$'}EXIT_CODE > TEMP_EXIT_CODE.txt
+    # use a swap file to make creation and writing atomic
+    mv TEMP_EXIT_CODE.txt $EXIT_CODE_FILE
 
     # propagate connector exit code by assuming it
     exit ${'$'}EXIT_CODE