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 71.18%. Comparing base (c5fec1f) to head (2946a60).

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #3716      +/-   ##
============================================
- Coverage     74.54%   71.18%   -3.37%     
+ Complexity      969      950      -19     
============================================
  Files          1306     1301       -5     
  Lines        150028   148192    -1836     
  Branches     125462   123915    -1547     
============================================
- Hits         111840   105487    -6353     
- Misses        34686    39406    +4720     
+ Partials       3502     3299     -203     

Components	Coverage Δ
Rust Core	74.62% <ø> (-0.22%)	⬇️
Java SDK	63.10% <ø> (+0.39%)	⬆️
C# SDK	33.11% <ø> (-39.17%)	⬇️
Python SDK	92.27% <ø> (ø)
PHP SDK	84.52% <ø> (ø)
Node SDK	80.77% <ø> (-11.48%)	⬇️
Go SDK	18.18% <ø> (-24.91%)	⬇️
see 210 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

[slbotbm]

@avirajkhare00 looks like your lock file is stale which is leading to CI failures.