[dagster-airbyte] don't cancel completed syncs (#7888)

dagster-io · May 18, 2022 · 0424134 · 0424134
1 parent d6b4eaf
commit 0424134
Show file tree

Hide file tree

Showing 4 changed files with 7 additions and 6 deletions.
diff --git a/python_modules/libraries/dagster-airbyte/dagster_airbyte/resources.py b/python_modules/libraries/dagster-airbyte/dagster_airbyte/resources.py
@@ -138,6 +138,7 @@ def sync_and_poll(
         start = time.monotonic()
         logged_attempts = 0
         logged_lines = 0
+        state = None
 
         try:
             while True:
@@ -177,9 +178,10 @@ def sync_and_poll(
                 else:
                     raise Failure(f"Encountered unexpected state `{state}` for job_id {job_id}")
         finally:
-            # make sure that the Airbyte job does not outlive the python process
-            # cancelling a successfully completed job has no effect
-            self.cancel_job(job_id)
+            # if Airbyte sync has not completed, make sure to cancel it so that it doesn't outlive
+            # the python process
+            if state not in (AirbyteState.SUCCEEDED, AirbyteState.ERROR, AirbyteState.CANCELLED):
+                self.cancel_job(job_id)
 
         return AirbyteOutput(job_details=job_details, connection_details=connection_details)
 

diff --git a/python_modules/libraries/dagster-airbyte/dagster_airbyte_tests/test_asset_defs.py b/python_modules/libraries/dagster-airbyte/dagster_airbyte_tests/test_asset_defs.py
@@ -56,7 +56,6 @@ def test_assets(schema_prefix):
         json=get_sample_job_json(schema_prefix=schema_prefix),
         status=200,
     )
-    responses.add(responses.POST, f"{ab_resource.api_base_url}/jobs/cancel", status=204)
 
     ab_job = build_assets_job(
         "ab_job",

diff --git a/python_modules/libraries/dagster-airbyte/dagster_airbyte_tests/test_ops.py b/python_modules/libraries/dagster-airbyte/dagster_airbyte_tests/test_ops.py
@@ -40,7 +40,6 @@ def airbyte_sync_job():
         rsps.add(rsps.POST, f"{ab_url}/connections/sync", json={"job": {"id": 1}})
         rsps.add(rsps.POST, f"{ab_url}/jobs/get", json={"job": {"id": 1, "status": "running"}})
         rsps.add(rsps.POST, f"{ab_url}/jobs/get", json={"job": {"id": 1, "status": "succeeded"}})
-        rsps.add(rsps.POST, f"{ab_url}/jobs/cancel", status=204)
 
         result = airbyte_sync_job.execute_in_process()
         assert result.output_for_node("airbyte_sync_op") == AirbyteOutput(

diff --git a/python_modules/libraries/dagster-airbyte/dagster_airbyte_tests/test_resources.py b/python_modules/libraries/dagster-airbyte/dagster_airbyte_tests/test_resources.py
@@ -70,7 +70,8 @@ def test_sync_and_poll(state):
         json={"job": {"id": 1, "status": state}},
         status=200,
     )
-    responses.add(responses.POST, f"{ab_resource.api_base_url}/jobs/cancel", status=204)
+    if state == "unrecognized":
+        responses.add(responses.POST, f"{ab_resource.api_base_url}/jobs/cancel", status=204)
 
     if state == AirbyteState.ERROR:
         with pytest.raises(Failure, match="Job failed"):