Airflow

왜 폴링을 공유하는가: shared_stream_key

서로 다른 Asset에 붙은 여러 ``AssetWatcher``가 같은 업스트림 리소스(플래그 파일이 쌓이는 디렉터리, 폴링 대상 REST 엔드포인트 등)를 읽는 트리거로 뒷받침되는 경우, 공유 메커니즘이 없다면 트리거러는 트리거마다 독립된 폴 루프를 하나씩 띄운다 — 구독자 20명이면 폴 루프 20개, 연결 20개, 주기마다 API 호출 20번이 든다.

``BaseEventTrigger``는 opt-in 방식의 공유 경로를 제공한다: 형제 트리거들이 하나의 폴을 공유하면서도, 각자 자기 DB 로우, 자기 ``run_trigger`` 태스크, 자기 인스턴스 단위 필터링은 그대로 유지한다. 참여하려면 서브클래스가 세 훅을 오버라이드한다.

* ``shared_stream_key()`` — 공유할 업스트림을 식별하는 키(보통 문자열 튜플)를 반환한다. 키가 같은 트리거끼리 폴 하나를 공유한다. ``None``(기본값)을 반환하면 공유에서 빠져 예전처럼 독립적인 ``run()`` 루프를 돈다. 이 값은 트리거러가 트리거를 시작할 때 '한 번만' 읽으므로, 생명주기 중간에 값을 바꿔도 그룹 소속에는 영향이 없다 — 키는 설정 필드에서 결정론적으로 유도해야 하며 ``time.time()``이나 ``uuid.uuid4()`` 같은 호출별 값을 쓰면 안 된다(그룹 생명주기 동안 비교가 안정적이어야 하기 때문). * ``open_shared_stream()`` — 트리거러가 '공유 그룹당 한 번씩만' 구동하는 ``@classmethod`` 코루틴으로, 업스트림에서 원시(raw) 이벤트를 산출한다. 트리거러가 이 공유 폴을 구동할 때 한 트리거의 kwargs를 재사용하므로, ``shared_stream_key``에 반영된 필드만 신뢰해서 써야 한다. * ``filter_shared_stream()`` — 인스턴스 메서드로, 브로드캐스트된 원시 스트림을 소비해 이 트리거가 실제로 발동시켜야 할 ``TriggerEvent``만 산출한다. (예: 이 인스턴스의 filename에만 매칭되는) 트리거별 필터링은 여기서 이루어진다.

예를 들어 여러 형제 트리거가 같은 인박스 디렉터리를 감시하며 각자의 플래그 파일이 나타날 때 발동하는 ``DirectoryFileDeleteTrigger``라면, ``shared_stream_key``로 ``("directory-scan", self.directory, self.poke_interval)``을 반환해 같은 디렉터리·주기를 감시하는 트리거끼리 디렉터리 스캔 하나를 공유하고, ``filter_shared_stream``에서는 브로드캐스트된 각 스냅샷에서 자신의 filename이 있는지만 확인한다.

공유 범위와 적합한 업스트림

공유 범위는 이름이 암시하는 것보다 좁다. 공유되는 것(키당 하나)은 ``open_shared_stream`` 비동기 제너레이터와 그 업스트림 I/O(예: 디렉터리 ``iterdir`` 호출, REST API 폴링 호출) 뿐이다. 공유되지 '않는' 것(트리거당 하나)은 ``Trigger`` DB 로우, 트리거 인스턴스, ``run_trigger`` asyncio 태스크, ``filter_shared_stream`` 비동기 제너레이터다 — 각 ``AssetWatcher``는 UI와 메타데이터 DB에서 여전히 자신만의 트리거로 보인다. 즉 절감 효과는 폴 루프·업스트림 I/O 계층에서만 발생하고, 영속성이나 스케줄링 계층에서는 발생하지 않는다.

이 패턴에 잘 맞는 업스트림은 다음과 같다: 멱등적/읽기 전용 읽기(디렉터리 스캔, 폴링 REST API), 구독자 측 부수효과 정리(트리거의 이벤트별 동작이 공유 프로듀서 핸들과 무관하게 구독자가 소유한 API를 거치는 경우, 예: unlink, 로컬 마킹), 그리고 모든 구독자가 처리를 마친 뒤에만 프로듀서가 커밋/삭제/ack해야 하는 메시지 브로커 업스트림(Kafka, SQS, Pub/Sub, Azure Service Bus) — 이 마지막 경우는 다음 절의 ack 채널이 필요하다.

Producer-side Ack 채널

프로듀서가 모든 구독자의 처리가 끝난 뒤에만 커밋/삭제/ack로 진행해야 하는 업스트림에는 ``create_shared_stream_producer``를 오버라이드해 ``SharedStreamProducer``를 반환하게 만든다. 이 팩토리를 오버라이드하면 매니저가 'ack 모드'로 들어간다 — 구독자 쪽 코드(``filter_shared_stream``)는 바뀌지 않는다. 원시 이벤트를 그대로 받으므로 fast path와 ack 모드에서 같은 필터 코드를 쓸 수 있고, 프레임워크가 각 구독자의 소비 진행 상황과 그로부터 파생된 트리거 이벤트의 영속화 여부로부터 언제 브로커가 진행(advance)해도 되는지를 추론한다.

흐름은 다음과 같다. (1) 매니저는 그룹당 한 번 ``create_shared_stream_producer(kwargs)``를 호출한다 — 반환된 프로듀서는 한 폴의 생명주기 동안 브로커 연결을 소유하며, 연결은 팩토리가 아니라 ``open_stream`` 안에서 지연 오픈해야 한다. (2) 프로듀서의 ``open_stream``은 ``(event, broker_payload)`` 튜플을 산출한다(broker_payload는 나중에 필요한 것 — SQS 수신 핸들, Kafka 오프셋, Pub/Sub ack ID 등). (3) 각 구독자의 ``filter_shared_stream``은 fast path와 동일하게 원시 이벤트를 받는다. 구독자는 그 이벤트를 '지나간'(다음 원시 이벤트를 당겼거나 구독 해지) 시점, 그리고 그로부터 파생된 모든 ``TriggerEvent``가 메타데이터 DB에 영속화된 시점에 그 이벤트를 'resolve'한 것으로 간주된다. (4) 팬아웃 세트의 모든 구독자가 이벤트를 resolve(지나감, 구독 해지, 타임아웃, 큐 오버플로 중 하나)하면 매니저는 그 레인(lane)에서 완전히 resolve된 연속 구간(prefix)을 담아 ``await producer.advance(batch)``를 호출한다 — batch의 각 항목은 이벤트의 broker_payload와, 구독자들이 어떻게 resolve했는지 개수를 담은 ``AdvanceOutcome``을 가진 ``AdvanceItem``이다.

'거부(reject)'는 구독자의 필터가 원시 이벤트로부터 트리거 이벤트를 산출하는 대신 능동적으로 거부하는 것으로, ``reject_shared_stream_event()``를 호출해 표현하며 비자발적 실패와는 구분된다 — ``failed`` 카운트는 구독자가 제시간에 끝내지 못했거나(ack 타임아웃) 뒤처졌음(큐 오버플로)을 뜻하고 보통 재전달이 맞는 대응이지만, ``rejected`` 카운트는 구독자가 이 이벤트는 트리거 이벤트를 만들면 안 된다고 판단해 영구 폐기해야 함을 뜻하고 재전달이 아니라 데드레터/nack이 맞는 대응이다. ``AdvanceOutcome``은 이 두 카운트를 분리해서 보고하므로 프로듀서가 브로커별로 다르게 대응할 수 있다 — 예를 들어 Azure Service Bus는 rejected가 0이 아니면 데드레터, failed만 있으면 abandon(재전달 유도), 전원 수락이면 complete 처리하고, Pub/Sub은 거부가 있으면 nack 아니면 ack한다. 프레임워크 자체는 절대 데드레터·nack·재전달을 직접 하지 않으며, 이는 전적으로 프로듀서의 ``advance``에 달려 있다.

``reject_shared_stream_event``는 필터가 ack 모드에서 원시 이벤트를 처리하는 도중(그 이벤트의 바인딩 윈도가 열려 있는 동안)에만 의미가 있다. fast path, 독립 ``run()``, 또는 두 원시 이벤트 사이에서 호출하면 경고 로그만 남기고 아무 일도 하지 않는다 — 영향을 줄 브로커 advance가 없기 때문이다. 즉시 resolve되므로 영속화를 기다리지 않는다.

``is_clean``은 브로드캐스트 시점에 온라인이던 모든 구독자가 그 이벤트를 수락했을 때만(거부 없음, 실패 없음, 최소 1명 이상 ack) True다. 단 하나의 거부나 실패만 있어도 False가 되며, 구독자가 0명인 브로드캐스트(카운트 전부 0)도 False다 — 아무도 수락하지 않았으므로 프로듀서가 커밋할 근거가 없기 때문이다.

순서 보장(Lane)과 장애 처리

기본적으로 모든 이벤트는 동일한 레인(lane)에 속한다. 배치의 항목들은 이벤트 순서대로이며 그 레인의 연속된 resolve 구간을 이루고, 레인 내에서 배치는 엄격히 순서대로 도착한다(다음 advance는 이전 advance가 반환된 뒤에만 대기된다). ``advance``가 예외를 던지면 공유 스트림 그룹 전체가 종료되고 모든 구독자가 실패 시그널을 받으며 브로커는 커밋되지 않은 오프셋부터 재전달한다(조용한 데이터 스킵 대신 시끄럽고 안전한 실패를 택한 것). ``get_advance_lane``을 오버라이드하면 순서 보장 범위를 레인 안으로 좁힐 수 있다 — 같은 레인 값끼리는 서로 순서를 지키며 배치되고 advance되지만, 다른 레인의 이벤트는 서로를 기다리지 않는다. 어느 경우든 한 번에 대기(await)되는 advance 호출은 최대 하나뿐이며, Kafka 오프셋 커밋처럼 누적 방식인 경우 배치의 마지막 항목만 커밋하면 된다.

Kafka 예시로, ``(topic, partition)``을 레인으로 반환하면 파티션별로만 순서를 보장한다 — Kafka 커밋은 파티션 안에서 커밋한 오프셋까지 모두를 인정하므로, 같은 파티션의 더 이른 이벤트가 아직 대기 중일 때 나중 이벤트가 먼저 커밋되지 않아야만 안전하고, 다른 파티션의 이벤트는 서로 상관이 없다. 이렇게 파티션 단위로 좁히면 느린 파티션 하나가 다른 파티션들의 커밋을 지연시키지 않는다. 이때 Kafka 컨슈머의 auto-commit은 반드시 꺼야 한다(기본값은 켜짐이다) — 그렇지 않으면 컨슈머가 자체 스케줄로 커밋해버려 ack 채널이 브로커가 무엇을 '전달됨'으로 여기는지 더는 통제하지 못한다. 폴이 끝나면 매니저는 ``producer.aclose()``를 한 번, best-effort로 호출한다.

운영 시 알아야 할 장애/한계: 구독자 집합은 이벤트가 브로드캐스트되는 순간에 '스냅샷'으로 고정되며, 그 이후 합류한 구독자는 그 이벤트의 대기 집합에 포함되지 않는다. 이벤트별 ack 타임아웃(기본 5분, ``[triggerer] shared_stream_ack_timeout``)을 넘기면 매니저는 그 구독자의 트리거만 강제 실패시키고(다른 구독자는 영향 없이 계속 진행) 구독자 쪽에는 ``AckTimeout``(``airflow.triggers.shared_stream``)이 ``filter_shared_stream`` 안의 반복자에서 발생한다. 트리거러가 재시작되면 resolve 상태는 메모리에만 있었으므로 사라지고, 브로커는 advance되지 않은 메시지를 재전달한다 — 그래서 구독자는 반드시 멱등적이어야 한다. 그룹 재형성 시 먼저 재구독한 트리거가 새 그룹을 즉시 만들어 폴링을 시작하고, 늦게 재구독한 트리거는 이미 브로드캐스트된 이벤트를 놓칠 수 있는 '늦은 구독자'로 합류한다(``[triggerer] shared_stream_cohort_grace_period``로 그룹 생성 후 폴링 시작을 지연시켜 동시 재구독을 기다려줄 수 있지만 best-effort일 뿐 위험을 완전히 없애지는 못한다). 브로커 advance는 영속화에 게이팅되어 있어, 확인이 끝내 오지 않으면(트리거러 크래시 등) ack 타임아웃이 이벤트를 실패시키고 커밋하지 않아 재전달되므로 — 결과적으로 실패는 이벤트 유실이 아니라 '중복 전달'을 일으킬 뿐이며 멱등적 구독자가 이를 흡수한다. ack 모드에서도 ``shared_stream_subscriber_queue_size``는 구독자별 미처리 원시 이벤트 개수를 그대로 제한하며, 매니저는 미해결 resolve를 기다리지 않고 다음 업스트림 이벤트를 계속 당긴다(큐가 차면 그 구독자를 강제 실패시킴). 공유가 실제로 동작하는지는 트리거러 로그의 'Shared stream group started key=(...)' 한 줄(키당 한 번, 구독자 수와 무관)로 확인할 수 있고, 구독자마다 한 줄씩 찍힌다면 ``shared_stream_key``들이 서로 다르게 비교되고 있다는 뜻이다. 폴 생산 속도가 어느 구독자의 ``filter_shared_stream`` 소비 속도보다 빠르면 그 구독자의 유한한 인메모리 큐가 차서 ``_SubscriberOverflow``로 실패하는데, 이는 무한정 메모리가 늘어나는 대신 의도적으로 빠르게 실패시키는 설계다 — 해결책은 ``[triggerer] shared_stream_subscriber_queue_size``를 올리거나, ``shared_stream_key``를 더 세분화해 한 그룹을 공유하는 형제 수를 줄이는 것이다.