Proposal: Generic HTTP Sink Connector

[mlevkov]

Hi team,

We've been running Iggy in production for our ad mediation platform since January 2026 (migrated from Fluvio) and would like to contribute a generic HTTP sink connector back to the ecosystem.

The connector framework has 6 sinks today, but no generic HTTP sink for arbitrary endpoints — webhooks, Lambda functions, REST APIs, SaaS integrations. Every HTTP-using sink (Quickwit, Elasticsearch, Iceberg) re-implements its own client logic, error handling, and retry strategy independently.

We've put together a detailed design document (attached) covering the full proposal: four batch modes (individual, ndjson, json_array, raw), configurable metadata envelopes with proper u128/binary serialization, exponential backoff retry with transient error classification, and an opt-in health check. The design follows patterns from the existing sinks — reqwest from Quickwit, retry logic from PostgreSQL/MongoDB, AtomicU64 counters from MongoDB. No new workspace dependencies required.

During our analysis of the runtime and existing sinks, we also identified two issues in the connector runtime that affect
all sinks — details in the attached document under "Runtime Issues Discovered During Analysis." Happy to file these separately and contribute fixes if the team agrees they should be addressed.

We'd love your feedback on the design before we start implementation. A few specific questions:

1. Should we open a tracking issue first, or go directly to a PR?
2. Are there any design preferences we should be aware of (e.g., batch mode naming, config field conventions)?
3. Any thoughts on the runtime issues — should we file those as separate issues?

Looking forward to your feedback.

IDEA-010-iggy-http-sink-github-discussion.md

[hubcio]

hey @mlevkov, it always warms my heart when I see someone talking positive about iggy.

I went through the full design doc and the codebase. about http sink: it looks well researched, and since you will be using it, it's even better that you will implement it. one thing is that the individual mode might be perf problem at any real thrughput - if you set batch_length to 50 with individual mode, then you will have sequential 50 HTTP round trips per poll cycle. other than that, from me you have a green light for implementation (but create issue first).

I have a bit more to add about runtime issues (both confirmed, i checkedin code):

1. consume() return value discarded: itis a legit bug, the increment_errors() metric method exists but is never called for sink consumption failures. interestingly the source side has a similar issue at line 434 where the callback return is also discarded, though sources at least have some error tracking in the forwarding loop.
2. AutoCommitWhen::PollingMessages commits before procesing: yep. due to this we cant achieve at-least-once delivery for any sink connector. unfortunately, auto-commit strategy is also hardoded in sink.rs.

your proposed fix for retry logic is somewhat OK but it does not fix resolve the coree issue: if process crashes mid-consume those messages are gone regardless of what happens inside the callback.

I think I have somewhat of a plan for proper resolution:

step 1 - visibility:

• capture the FFI return code in process_messages() instead of discarding it
• on non-zero: log error with context (plugin_id, stream, topic, return code), increment the errors metric
• once we surface FFI errors, change the caller (consume_messages()) to continue on callback failure instead of return Err(...) which kills the consumer loop permanently - one transient failure shouldn't brick the sink forever
• add a warn log at consumer setup about PollingMessages auto-commit semantics

this gives operators immediate visibility into failures that are currently invisible, without changing the FFI contract or auto-commit strategy. all existing sinks benefit.

step 2 - proper at-least-once delivery

• enrich the FFI return codes so the runtime can distinguish transient from permanent errors (backwards-compatible -- existing plugins returning 0/1 still work)
• switch sink consumers to manual commit (AutoCommit::Disabled) and only commit after successful consume(). i checked the SDK - ConsumingAllMessages won't work either, it commits when the last buffered message is yielded from .next(), which is still before process_messages() runs (see store_offset_after_all_messages in consumer.rs). manual commit is the only correct option.
• add a consecutive failure circuit breaker so a broken sink doesn't loop forever
• expose commit strategy in config so users can opt into the old at-most-once behavior if they prefer lower latency (notsure about this)

step 1 is a standalone PR. step 2 probably splits into SDK changes and runtime changes - we can finalize the details in the tracking issues.

finally, answering your questions:

1. yes, open a tracking issue first, link this discussion.
2. the runtime bug fixes should be separate issues + separate PRs from the http sink itself. smaller PRs are easier to review and the fixes benefit all sinks immediately
3. skip body_template for v1, that's the right call. if someone needs custom envelopes, the transform system already exists for that
4. env var expansion in headers - worth considering but also fine as a follow-up. for v1, users can just set the literal value in config
   and unfortunately I'm gonna end this overly long message with more open questions:

looking forward to seeing the issues and the PR!

[hubcio]

ignore the "and unfortunately I'm gonna end this overly long message with more open questions:" :)

[mlevkov]

PR and Issues Filed

The HTTP sink connector implementation is ready for review:

• PR: feat(connectors): add generic HTTP sink connector #2925 — feat(connectors): add generic HTTP sink connector
  • 4 batch modes (individual, ndjson, json_array, raw), retry with exponential backoff, metadata envelope
  • 60 unit tests + 7 integration tests (WireMock in Docker)
  • Connection pooling with TCP keep-alive (30s) and pool idle timeout (90s)
  • Comprehensive README (810 lines): config reference, runtime model, deployment patterns, message flow
  • 4 rounds of automated code review (42 findings addressed)
  • Follows all patterns from existing sinks (MongoDB, PostgreSQL, Elasticsearch)
  • No new dependencies — all workspace deps

While analyzing the runtime to inform the design, we found two issues affecting all existing sinks. Filed separately:

• Issue bug(connectors): consume() return value discarded in sink runtime #2927 — consume() return value discarded in sink runtime (process_messages() in runtime/src/sink.rs)
• Issue bug(connectors): PollingMessages auto-commit commits offsets before sink processing #2928 — PollingMessages auto-commit commits offsets before sink processing

Integration tests

The PR includes 7 integration tests in core/integration/tests/connectors/http/ using WireMock to receive and verify HTTP deliveries. Tests cover:

• Single message delivery with metadata envelope verification
• Metadata field toggles (checksum, origin timestamp)
• All 3 batch modes (ndjson, json_array, individual)
• Raw byte passthrough
• Multi-topic delivery with per-topic metadata accuracy

Test counts

Category	Count
Unit tests	60
Integration tests	7
Code review rounds	4 (double-review protocol)
Review findings fixed	40
Review findings deferred (with issues)	2

[spetz]

I really like the idea of generic HTTP connector, given the existing transforms etc. this could be used to quickly build integrations for lots of APIs out there. Actually, at some point, it could be even possible to have the custom HTTP client + message processor, and reuse it across other sinks/sources directly using HTTP API so that for all these plugins we could rely on a single, configurable, well-tested HTTP component to do all the processing.

[mlevkov]

Thank you for the note.

[mlevkov]

Thank you @spetz — that's a great perspective on where this can go.

We designed the HTTP layer with that kind of reuse in mind, even if it's currently packaged as a single sink plugin. The internals are already separated into composable pieces:

• Connection management: reqwest::Client with configurable pool sizing, TCP keep-alive (30s for cloud LB detection), and pool idle timeout (90s) — all independent of the Sink trait
• Retry engine: Exponential backoff with Retry-After header parsing, transient status classification (429/5xx), and configurable max delay — generic enough for any HTTP call
• Batch serialization: 4 modes (individual, ndjson, json_array, raw) covering the common patterns for both JSON and binary payloads
• Metadata envelope: Optional {metadata, payload} wrapping with Iggy context (stream, topic, offset, timestamps)

Extracting the HTTP client + retry + serialization into a shared iggy-http-client crate that other sinks/sources can depend on feels like a natural next step. The Elasticsearch and Quickwit sinks already speak HTTP — a shared component would give them connection pooling, retry, and observability for free.

Happy to explore that direction once the base sink lands and stabilizes.

[mlevkov]

Feature idea: Per-message HTTP header forwarding

While working on the HTTP sink PR (#2925), a question came up about dynamic per-message headers. Today, all HTTP headers are static (config-derived, same for every request). But there are real use cases for per-message headers:

Use cases:

• X-Iggy-Offset / X-Iggy-Topic / X-Iggy-Partition — routing and dedup at the HTTP endpoint without body parsing
• Forwarding Iggy user-defined message headers (ConsumedMessage.headers) as HTTP headers — e.g., a producer attaches X-Correlation-ID and the sink forwards it to the downstream endpoint

Proposed design — additive layering:

// Static config headers (pre-built once in open(), cloned per-request)
let mut request = build_request(self.method, client, &self.url)
    .headers(self.request_headers.clone())
    .header("content-type", content_type);

// Per-message dynamic headers (opt-in via config)
if self.include_metadata_headers {
    request = request
        .header("x-iggy-offset", offset.to_string())
        .header("x-iggy-topic", &topic_metadata.topic)
        .header("x-iggy-partition", partition_id.to_string());
}

// Forward Iggy user headers as HTTP headers (opt-in)
if self.forward_iggy_headers {
    if let Some(headers) = &message.headers {
        for (key, value) in headers {
            request = request.header(
                format!("x-iggy-{}", key.to_string_value()),
                value.to_string_value(),
            );
        }
    }
}

Config surface:

[plugin_config]
# Include iggy metadata (offset, topic, partition) as HTTP headers
include_metadata_headers = false

# Forward iggy user-defined message headers as HTTP headers (prefixed with x-iggy-)
forward_iggy_headers = false

This preserves the current pre-built HeaderMap optimization for static headers while layering dynamic per-message headers on top. Both options are opt-in and disabled by default.

Thoughts? This could be a follow-up PR if there is interest.

[hubcio]

@mlevkov can we make it a separate discussion?

[mlevkov]

Done — moved to a separate discussion: #3029

Added some extra design considerations around batch mode implications and header name conflicts.

[mlevkov]

@hubcio please take a look