Airflow

재시도 정책의 동작 원리

재시도 정책은 예외 타입을 액션에 매핑하는 규칙들로 정의하고, 이를 태스크나 오퍼레이터의 파라미터로 적용한다. 정책은 태스크 코드를 수정하지 않고도 예외별 재시도 동작을 설정할 수 있게 해준다. 이 정책은 스케줄러가 아니라 항상 '태스크 워커 프로세스' 안에서, 예외를 캐치한 시점과 태스크의 다음 상태를 결정하는 시점 사이에 실행된다. 각 정책 결정은 태스크 로그에 ``Retry policy decision action=<action> reason=<reason>`` 형식으로 기록된다.

태스크가 실패하면 정책은 예외를 평가해 세 가지 액션 중 하나를 반환한다: **RETRY**(태스크를 재시도하며, 선택적으로 ``retry_delay``를 오버라이드하는 커스텀 지연을 지정할 수 있다 — 다만 이 재시도도 여전히 태스크의 ``retries`` 카운트 제약을 받으므로, 정책이 더 일찍 실패시킬 수는 있어도 설정된 최대치를 넘어 재시도를 연장할 수는 없다), **FAIL**(남은 재시도를 건너뛰고 즉시 실패), **DEFAULT**(표준 재시도 로직 — ``retries`` 카운트와 ``retry_delay`` — 로 그대로 넘어감). 규칙은 순서대로 평가되며 처음 매칭되는 규칙이 적용된다. 어떤 규칙도 매칭되지 않으면 정책은 DEFAULT(표준 재시도 동작)를 반환한다.

예외 타입은 Python 클래스로 지정하거나 점(dot)으로 구분된 import 경로 문자열(예: ``"requests.exceptions.HTTPError"``)로 지정할 수 있다. 문자열 경로는 Dag 파싱 시점에 검증되는데, 점이 없는 경로는 즉시 ``ValueError``를 일으키고, 해석 불가능한 경로는 경고를 낸다. 기본적으로 규칙은 ``isinstance`` 매칭을 쓰므로 ``OSError``에 대한 규칙은 그 서브클래스인 ``ConnectionError``에도 매칭된다. 정확한 타입 매칭만 원한다면 ``match_subclasses=False``를 설정하면 된다.

기존 파라미터와의 조합, 매핑된 태스크, 커스텀 정책

``retry_policy``가 설정되어 있어도 ``retries``는 여전히 최대 재시도 횟수의 상한으로 작동한다 — 정책이 더 일찍 실패시킬 수는 있지만 이를 초과해 연장할 수는 없다. ``retry_delay``/``retry_exponential_backoff``/``max_retry_delay``는 정책이 DEFAULT를 반환하거나 ``RetryDecision.retry_delay``가 None일 때 사용된다. ``on_retry_callback``은 정책 주도 재시도를 포함해 모든 재시도에서 실행된다. ``AirflowFailException``은 항상 최우선으로 적용되며, 이 예외에 대해서는 정책이 아예 참조되지 않는다.

재시도 정책은 ``.partial()``을 통한 동적 태스크 매핑과 함께 동작한다. 정책은 매핑된 태스크 인스턴스 각각에 개별 적용되므로 — 10개 중 인스턴스 2가 FAIL을 만나도 나머지 9개는 독립적으로 계속 진행된다. 정책은 ``.partial()``에서 태스크 레벨로 설정되어 모든 매핑된 인스턴스가 하나의 정책을 공유하며, ``.expand()``에서는 인덱스별로 다른 정책을 지정할 수는 없지만, 정책의 ``evaluate()`` 메서드가 예외·try_number·전체 컨텍스트를 전달받으므로 필요하다면 정책 내부에서 인덱스별 분기 로직을 구현할 수 있다.

고급 사용 사례에서는 ``RetryPolicy``를 서브클래싱하고 ``evaluate()``를 구현할 수 있다. 예외의 속성(상태 코드, 응답 헤더, 페이로드 등)까지 검사해야 하는 경우, 선언적인 ``ExceptionRetryPolicy`` 규칙만으로는 표현할 수 없으므로 서브클래싱이 적합하다. ``evaluate()``는 예외, try 번호, 최대 시도 횟수, 그리고 ``dag_run``·``params`` 등을 포함한 전체 Airflow 컨텍스트를 받는다. 예를 들어 HTTP 429 응답이면 ``Retry-After`` 헤더를 존중해 지연을 지정하고, 5xx면 재시도하고, 4xx면 재시도하지 않고 실패시키는 정책을 만들 수 있고, 또는 컨텍스트의 ``dag_run.run_type``이 ``backfill``인지 확인해 백필 실행에서는 재시도 없이 즉시 실패시켜 과거 오류를 빠르게 드러내는 정책도 만들 수 있다.

Task Instance Heartbeat Timeout

완벽하게 동작하는 시스템은 없으며, Task Instance는 이따금 죽을 수 있다. TaskInstance는 연관된 job이 비활성 상태가 되었음에도(예: TaskInstance의 워커가 메모리 부족으로 죽은 경우) ``running`` 상태에 갇혀 있을 수 있다. 이런 태스크는 과거에 'zombie task'라고 불렸다. Airflow는 주기적으로 이런 태스크를 찾아 정리하고, TaskInstance를 failed로 표시하거나 재시도 가능 횟수가 남아 있다면 재시도시킨다. TaskInstance의 heartbeat이 타임아웃되는 원인은 다양하다: Airflow 워커가 메모리 부족으로 OOMKilled되었거나, 워커가 liveness probe에 실패해 시스템(예: Kubernetes)이 워커를 재시작했거나, 시스템이 스케일 다운하면서 워커를 다른 노드로 옮긴 경우 등이다.

로컬에서 재현하려면 ``AIRFLOW__SCHEDULER__TASK_INSTANCE_HEARTBEAT_SEC``, ``AIRFLOW__SCHEDULER__TASK_INSTANCE_HEARTBEAT_TIMEOUT``, ``AIRFLOW__SCHEDULER__TASK_INSTANCE_HEARTBEAT_TIMEOUT_DETECTION_INTERVAL`` 환경변수(또는 airflow.cfg의 대응 설정값)를 설정하고, 약 10분이 걸리는 장기 실행 태스크(예: ``sleep 600``을 실행하는 BashOperator)를 가진 Dag를 실행한 뒤 기다리면, 설정한 heartbeat timeout 초 뒤에 TaskInstance가 failed로 표시되는 것을 볼 수 있다.

Executor Configuration

일부 Executor는 태스크별 선택적 설정을 허용한다 — 예를 들어 KubernetesExecutor는 태스크를 실행할 이미지를 지정할 수 있게 해준다. 이는 Task/Operator의 ``executor_config`` 인자를 통해 이루어지며, ``MyOperator(..., executor_config={"KubernetesExecutor": {"image": "myCustomDockerImage"}})``처럼 KubernetesExecutor에서 실행할 태스크의 Docker 이미지를 설정할 수 있다. ``executor_config``에 넣을 수 있는 설정값은 Executor마다 다르므로, 개별 Executor 문서를 참고해 무엇을 설정할 수 있는지 확인해야 한다.