Feature idea: Per-message HTTP header forwarding in sink connectors

[mlevkov]

Feature idea: Per-message HTTP header forwarding

Moved from discussion #2919 at @hubcio's request.

Context: 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.

Design considerations

• Batch modes: For ndjson and json_array, per-message headers only apply if the batch contains a single message, or we'd need to pick "representative" headers from the batch (e.g., first message, or common subset). For individual mode it's straightforward — each message is its own HTTP request.
• Header name conflicts: If a user-defined Iggy header name conflicts with a static config header, the static header should win (config is explicit intent). Alternatively, we could log a warning.
• Performance: Adding per-message headers is a few string allocations per request. For individual mode at high throughput this could add up, but it's behind an opt-in flag so the default path is unchanged.

This could be a follow-up PR after the base HTTP sink (#2925) lands. Thoughts?

[spetz]

I think it could be a useful feature, one thing I'd change - we could ditch x- prefix, and just use e.g. iggy-offset instead.

[kparisa]

@mlevkov If we leverage OTEL, this would be the Trace-Id right?

and +1 to @spetz recommendation to drop X- prefix. In modern implementations, we don't need that as per the proposal https://datatracker.ietf.org/doc/html/rfc6648

So we should have Apache-Iggy-Offset, Apache-Iggy-Topic, Apache-Iggy-Partition etc or without Apache- prefix to save space. Just hope that they won't collide with any other headers :)

[mlevkov]

Yes, likely so, or even any other header.