feat(connectors): add Apache Airflow trigger sink

[avirajkhare00]

Summary

• Add iggy_connector_airflow_sink that triggers Airflow DAG runs (POST /api/v1/dags/{dag_id}/dagRuns) from Iggy poll batches (one run per batch, not per message)
• conf.messages carries the batch; optional conf.iggy metadata; deterministic batch dag_run_id; HTTP 409 treated as success
• Auth modes none / basic / bearer; optional dag_id_header groups a poll into one run per DAG id
• Permanent 4xx fails the batch (no silent drop); WireMock + real Airflow e2e; tracked by feat(connectors): add Apache Airflow trigger connector #3715

Design note

Airflow runs are jobs, not stream events. High-traffic 1:1 message→run would overwhelm the scheduler. Volume is controlled via stream batch_length / poll_interval.

Test plan

• cargo fmt / cargo sort --no-format --workspace / clippy on the sink crate
• cargo test -p iggy_connector_airflow_sink (12 unit tests)
• cargo test -p integration -- connectors::airflow (WireMock: 3 messages → 1 DAG-run POST)
• Local e2e: Iggy server + connectors + Airflow 2.10.5 standalone (5 messages → 1 run iggy-0-0-4-5, re-POST 409)

Review follow-up

Addressed @kriti-sc feedback: batch trigger default, explicit idempotent dag_run_id, fail batch on permanent HTTP errors.

[github-actions]

Thanks for the PR. It is labeled S-waiting-on-review and queued for review.

Slash commands (own line, regular comment) move it around the queue:

• /ready - back to S-waiting-on-review after addressing feedback
• /author - flip to S-waiting-on-author while you finish changes
• /request-review @user-or-team - request a reviewer

See CONTRIBUTING.md for details.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 72.91%. Comparing base (4c30a88) to head (db2442e).

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #3716      +/-   ##
============================================
- Coverage     74.66%   72.91%   -1.75%     
  Complexity      950      950              
============================================
  Files          1304     1301       -3     
  Lines        149553   148290    -1263     
  Branches     125064   123956    -1108     
============================================
- Hits         111663   108132    -3531     
- Misses        34393    36766    +2373     
+ Partials       3497     3392     -105     

Components	Coverage Δ
Rust Core	74.62% <ø> (-0.33%)	⬇️
Java SDK	62.64% <ø> (ø)
C# SDK	72.26% <ø> (ø)
Python SDK	92.27% <ø> (ø)
PHP SDK	84.52% <ø> (ø)
Node SDK	81.37% <ø> (-11.47%)	⬇️
Go SDK	18.18% <ø> (-24.91%)	⬇️
see 117 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[kriti-sc]

I feel like this sink is an anti-pattern because Airflow is meant for the batch processing usecase while Iggy is a streaming system. More so, because a lot of other Iggy sinks also operate on message batch level and not individual message level.

This sink creates a dag run for each Iggy message, so for a high-traffic topics, it will end up overwhelming Airflow.

Two solutions:

1. implement an apply_function like the Kafka support in Airflow. That way the sink can be configured to trigger dag only when a message passes that filter.
2. Consider triggering Airflow on the batch and not individual messages.

[avirajkhare00]

Agreed — one DAG run per message is the wrong unit of work for Airflow.

Pushed a redesign: one DAG run per consume poll batch (not per message).

• conf.messages is an array of {offset, id, timestamp, payload} for the batch
• Optional conf.iggy carries stream/topic/partition + first/last offset + message_count
• Stream batch_length / poll_interval control volume; batch_length = 1 covers low-rate command topics
• If dag_id_header yields multiple DAG ids in one poll, we still do one run per DAG-id group, not one run per message

Verified with unit tests, WireMock integration (3 msgs → 1 POST), and real Airflow 2.10.5 e2e (5 msgs → 1 run iggy-0-0-4-5).

In-sink apply_function-style filtering is still out of scope for this PR; selective triggering is better via a dedicated topic (or a later transform). Happy to discuss if you want that in a follow-up.

[kriti-sc]

My Airflow knowledge maybe outdated, but afaik dag_run_id would be manual__timestamp which wouldn't be idempotant.

[avirajkhare00]

Good catch on the default Airflow form — that would only apply if we omitted dag_run_id.

We always set an explicit deterministic id, so Airflow does not generate manual__timestamp:

• Before (per-message): iggy-{partition}-{offset}-{message_id_hex}
• Now (batch): iggy-{partition}-{first_offset}-{last_offset}-{message_count}

Same poll range → same id on redelivery; Airflow returns 409, which we treat as success. Documented in the README with the request shape and a note that we never rely on auto-generated run ids.

[kriti-sc]

I am not sure about skipping messages here.
Classifying errors into 3 parts-

1. Transient Airflow related - Might not apply to the remainder of messages in the batch.
2. Malformed Iggy message - Same as above.
3. Permanent error such as auth or Airflow down - Might be better to fail the sink.

Feel free to add your reasoning or cases I might have missed.

[avirajkhare00]

Agreed the old per-message skip path was too loose for permanent failures.

With batch triggering the consume path is simpler:

1. Transient (429/5xx/network): middleware retries, then Err so the runtime can redeliver the same batch (same dag_run_id → 409 if already created).
2. Malformed single payload: skip that entry when building conf.messages, count an error; if every entry fails conversion, return error for the batch.
3. Permanent HTTP (401/403/404/400/422, auth/missing DAG, etc.): return PermanentHttpError for the whole batch — do not silent-drop and advance as if the run was created.

Auth / Airflow-down style permanent failures no longer drop messages and continue.

[avirajkhare00]

/ready