1fanwang opened a new pull request, #72414:
URL: https://github.com/apache/airflow/pull/72414
A synchronous Airbyte task can crash after Airbyte accepts a sync, then
start a duplicate sync when Airflow retries it. With this change, retries
resume active Airbyte jobs and recover completed jobs. A new job is created
only when no durable state exists or the previous job failed or was cancelled.
The operator uses AIP-103 `ResumableJobMixin` only for the synchronous,
non-deferrable wait path. Recovery restores `job_id` before polling or
returning so task results, the Airbyte job link, and `on_kill()` keep pointing
at the recovered job. Asynchronous and deferrable execution are unchanged.
Airflow versions before 3.3 retain the existing submit behavior.
## Testing
| Scenario | Result |
| --- | --- |
| Provider behavior matrix | Sync recovery, async/deferrable isolation,
`on_kill()`, and pre-3.3 compatibility covered |
| Active job stored before retry | Reconnected to job 77 without submitting |
| Current `upstream/main` with the same stored job | Submitted duplicate job
101 |
| Airflow 2.11 compatibility probe | Existing submit behavior preserved |
<details>
<summary>Commands and raw logs</summary>
```console
$ AIRFLOW_HOME=/tmp/airbyte-tests .venv/bin/uv run --project
providers/airbyte pytest providers/airbyte/tests/unit/airbyte -q
collected 64 items
...
=================== 64 passed, 1 warning in 71.13s (0:01:11)
===================
```
The recovery probe calls the real operator `execute()` path with a
task-state store containing `airbyte_job_id=77`. A local stand-in `AirbyteHook`
reports that job as running and records submissions, status checks, and waits.
```console
$ REPRO_VARIANT=upstream/main \
AIRBYTE_PROVIDER_SOURCE_ROOT=$UPSTREAM_MAIN/providers/airbyte/src \
.venv/bin/uv run --project providers/airbyte python
airbyte_resumable_repro.py
{'variant': 'upstream/main', 'stored_before': 77, 'task_state_get_calls':
[], 'task_state_set_calls': [], 'submit_count': 1, 'status_calls': [],
'wait_calls': [101], 'result': 101, 'operator_job_id': 101}
Traceback (most recent call last):
File "airbyte_resumable_repro.py", line 90, in <module>
raise RuntimeError("operator did not reconnect to the stored active job")
RuntimeError: operator did not reconnect to the stored active job
$ REPRO_VARIANT="PR branch" \
AIRBYTE_PROVIDER_SOURCE_ROOT=$PWD/providers/airbyte/src \
.venv/bin/uv run --project providers/airbyte python
airbyte_resumable_repro.py
2026-09-02T05:01:37.864774Z [info ] Reconnecting to existing job
[airflow.task.operators.airflow.providers.airbyte.operators.airbyte.AirbyteTriggerSyncOperator]
external_id=77 external_id_key=airbyte_job_id loc=resumablejobmixin.py:161
status=running
{'variant': 'PR branch', 'stored_before': 77, 'task_state_get_calls':
['airbyte_job_id'], 'task_state_set_calls': [], 'submit_count': 0,
'status_calls': [77], 'wait_calls': [77], 'result': 77, 'operator_job_id': 77}
```
</details>
<details>
<summary>Stand-in hook reproducer</summary>
Save this as `airbyte_resumable_repro.py` before running the commands above.
```python
from __future__ import annotations
import os
import sys
from importlib import import_module, invalidate_caches
from pathlib import Path
from types import SimpleNamespace
from typing import Any
from airbyte_api.models import JobStatusEnum
import airflow.providers
if source_root := os.environ.get("AIRBYTE_PROVIDER_SOURCE_ROOT"):
airflow.providers.__path__.insert(0, str(Path(source_root) / "airflow" /
"providers"))
for module_name in tuple(sys.modules):
if module_name == "airflow.providers.airbyte" or
module_name.startswith(
"airflow.providers.airbyte."
):
del sys.modules[module_name]
invalidate_caches()
airbyte_module = import_module("airflow.providers.airbyte.operators.airbyte")
class TaskStateStore:
def __init__(self) -> None:
self.values: dict[str, int] = {"airbyte_job_id": 77}
self.get_calls: list[str] = []
self.set_calls: list[tuple[str, int]] = []
def get(self, key: str) -> int | None:
self.get_calls.append(key)
return self.values.get(key)
def set(self, key: str, value: int) -> None:
self.set_calls.append((key, value))
self.values[key] = value
class StandInAirbyteHook:
def __init__(self) -> None:
self.submit_count = 0
self.status_calls: list[int] = []
self.wait_calls: list[int] = []
def submit_sync_connection(self, *, connection_id: str) -> Any:
self.submit_count += 1
return SimpleNamespace(job_id=101, status=JobStatusEnum.RUNNING)
def get_job_status(self, *, job_id: int) -> str:
self.status_calls.append(job_id)
return JobStatusEnum.RUNNING
def wait_for_job(self, *, job_id: int, wait_seconds: float, timeout:
float) -> None:
self.wait_calls.append(job_id)
task_store = TaskStateStore()
hook = StandInAirbyteHook()
airbyte_module.AirbyteHook = lambda **kwargs: hook
operator = airbyte_module.AirbyteTriggerSyncOperator(
task_id="airbyte_live_recovery",
connection_id="connection-uuid",
deferrable=False,
asynchronous=False,
wait_seconds=0,
)
result = operator.execute(context={"task_state_store": task_store})
print(
{
"variant": os.environ["REPRO_VARIANT"],
"stored_before": 77,
"task_state_get_calls": task_store.get_calls,
"task_state_set_calls": task_store.set_calls,
"submit_count": hook.submit_count,
"status_calls": hook.status_calls,
"wait_calls": hook.wait_calls,
"result": result,
"operator_job_id": operator.job_id,
}
)
if (
hook.submit_count != 0
or hook.status_calls != [77]
or hook.wait_calls != [77]
or result != 77
or operator.job_id != 77
):
raise RuntimeError("operator did not reconnect to the stored active job")
```
</details>
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (GitHub Copilot CLI, GPT-5.6 Sol)
Generated-by: GitHub Copilot CLI (GPT-5.6 Sol) following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
---
* Read the **[Pull Request
Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)**
for more information. Note: commit author/co-author name and email in commits
become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal
([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals))
is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License
Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment:
`{pr_number}.significant.rst`, in
[airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
You can add this file in a follow-up commit after the PR is created so you
know the PR number.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
