HTTP source

Source The HTTP Source connector turns an Ingest Engine pipeline into an HTTP server that accepts POST requests. Each request body becomes one or more records in the pipeline. It is useful for webhooks, application event ingestion, and any integration where you want to push data directly into a streaming pipeline without an intermediate message broker. The connector runs an embedded Axum HTTP server inside the Ingest Engine worker process. There is no connection profile - the server config (port, path, auth) lives entirely in the table config. Incoming request bodies are buffered in a bounded in-memory channel, deserialized according to the schema you define, and emitted as Arrow record batches for downstream processing. The server also transparently decompresses gzip-encoded request bodies (Content-Encoding: gzip).

Source Config

The table config tells Ingest Engine what port to listen on, which path to accept requests at, how large requests can be, and how much to buffer before dropping events.

port: 8080
path: /events
buffer_size: 1000
max_body_size: 5242880

port (required, u16) - Port number the embedded HTTP server binds to. The Ingest Engine worker listens on this port locally. To expose it externally, use a Kubernetes Service/Ingress or Docker port mapping. There is no default - you must specify it.
path (String, default "/") - HTTP endpoint path where POST requests are accepted. Requests to any other path receive a 404. Use this to namespace endpoints when running multiple HTTP source pipelines on different paths behind a shared ingress, e.g. /webhooks/stripe or /events/clickstream.
bind_address (String, default "0.0.0.0") - Network interface to bind to. 0.0.0.0 listens on all interfaces (the right choice for containers and production). Use 127.0.0.1 to restrict to localhost during local development.
buffer_size (usize, default 1000) - Capacity of the in-memory channel between the HTTP handler and the pipeline processing loop. Each accepted request occupies one slot. When the buffer is full, new requests are rejected with HTTP 429 Too Many Requests so clients can back off (see Response Codes). The default of 1000 assumes roughly 100 KB payloads, targeting around 100 MB of memory usage. Increase for smaller payloads, decrease for larger ones. This value also controls the concurrency semaphore - at most buffer_size requests are processed concurrently.
max_body_size (usize, default 5242880 / 5 MB) - Maximum allowed request body size in bytes. Requests exceeding this limit are rejected with 413 Payload Too Large by the Axum body limit layer before the handler ever sees the data. Set this based on your expected payload sizes to protect the worker from memory pressure caused by unexpectedly large requests.
auth (optional) - Authentication configuration. When set, every request must include a valid Authorization header. Unauthenticated requests receive 401 Unauthorized. See Authentication below.
service_name (String, default "") - An identifier used to generate the Kubernetes Service resource name for this HTTP source. When set, the controller creates a Kubernetes Service named http-src-{service_name} so that other services in the cluster can route traffic to this pipeline. If left empty, the default naming applies. Only relevant for Kubernetes deployments.

Authentication

The HTTP source supports optional authentication to protect the endpoint. When configured, every incoming request must carry a valid Authorization header - requests without one, or with incorrect credentials, receive a 401 Unauthorized response and the body bytes are tracked in the laminar_worker_http_source_bytes_rejected_total metric (label reason). Bearer Token - the simplest option. The client sends Authorization: Bearer <token> and Ingest Engine does a constant-time string comparison:

auth:
  type: bearer
  token: "my-secret-token"

Basic Auth - the client sends Authorization: Basic <base64(username:password)>. Ingest Engine base64-decodes the header and compares the username:password string:

auth:
  type: basic
  username: "user"
  password: "pass"

Parallelism Constraint

Parallelism must be 1. The HTTP source binds to a specific TCP port, and only one OS process can bind to a given port at a time. If the pipeline is configured with parallelism greater than 1, the operator panics at startup with a clear error message. This is enforced in the on_start hook - Ingest Engine checks ctx.task_info.parallelism and aborts before attempting to bind. If you need to scale HTTP ingestion throughput beyond what a single job can handle, use a Pipeline Replica Set, which runs multiple independent replicas of the pipeline behind one load-balanced endpoint. Alternatively, use a message broker (Kafka, Redpanda) as a fan-out layer.

Relay and Replay

Unlike Kafka or CDC, an HTTP producer has no durable offset to rewind to. If the source is restarting, recovering, or shedding load (returning 429/503), any data pushed during that window is simply lost - the producer would have to retry it itself. The relay and replay services close this gap, giving HTTP ingestion broker-like durability without introducing a message broker. Both are optional: a deployment that can tolerate dropped requests, or that already fronts the source with its own queue, can skip them. Relay is an HTTP reverse proxy you place in front of the source pods. It forwards each request to the source and, whenever the source returns a 5xx/429 or is unreachable, transparently buffers the request body to object storage (S3/GCS/Azure) and still returns 202 Accepted to the producer. Producers never observe the outage. The relay discovers which pipelines to buffer for automatically - the controller labels each HTTP source’s Kubernetes Service so the relay can find it. Replay is the catch-up half. It runs as a periodic job (on a cron schedule rather than as a long-lived daemon): each pass reads the buffered objects back and re-POSTs them to the source once it has recovered, then exits. To avoid racing the relay as it writes the current buffer, replay stays a couple of minutes behind real time, and it checkpoints its progress per topic so an interrupted pass resumes where it left off. Delivery is at-least-once - if your pipeline needs to deduplicate replayed records, do so using a content-derived key. Together they form a durable buffer in front of the source: the relay absorbs and persists traffic during outages, and replay drains that buffer back into the pipeline after recovery.

Schema

The schema block defines the expected structure of incoming request bodies. It is required - the HTTP source connector will not start without it.

schema:
  format:
    json: {}
  fields:
    - field_name: event_type
      field_type:
        type:
          primitive: Utf8
      nullable: false
    - field_name: payload
      field_type:
        type:
          primitive: Utf8
      nullable: true

See the Schema Reference for field type details.

Format

The format field determines how request bodies are deserialized into typed Arrow columns:

json: {} - JSON (the most common choice for HTTP payloads)
avro: {} - Apache Avro
protobuf: {} - Protocol Buffers
parquet: {} - Apache Parquet
raw_string: {} - Raw text, no parsing
raw_bytes: {} - Raw binary, no parsing

Framing

By default, each HTTP request body is treated as a single record. If you want to send multiple records in one request, configure framing to split the body into individual records before deserialization:

schema:
  format:
    json: {}
  framing:
    newline_delimited: {}
  fields:
    # ...

With newline_delimited framing, the body is split on newline characters. Each line is deserialized independently. This works well with newline-delimited JSON (NDJSON):

curl -X POST http://localhost:8080/events \
  -d '{"event_type": "a", "user": "alice"}
{"event_type": "b", "user": "bob"}
{"event_type": "c", "user": "carol"}'

Without framing, that entire body would be treated as a single record (and likely fail JSON deserialization since it is not valid JSON).

Bad Data Handling

The bad_data field controls what happens when a record fails deserialization:

fail: {} (default) - The pipeline stops with an error. Use this when data quality is critical and you want to catch schema mismatches immediately.
drop: {} - The bad record is silently dropped and a warning is logged (rate-limited to avoid log flooding). The pipeline continues processing subsequent records. Use this when you expect occasional malformed data and prefer availability over completeness.

schema:
  format:
    json: {}
  bad_data:
    drop: {}
  fields:
    # ...

Sending Data

Send data to the running pipeline with any HTTP client. The only requirement is a POST request to the configured path with a body matching the expected format. Single record:

curl -X POST http://localhost:8080/events \
  -H "Content-Type: application/json" \
  -d '{"event_type": "order.placed", "payload": "{\"order_id\": 12345}"}'

With bearer auth:

curl -X POST http://localhost:8080/events \
  -H "Authorization: Bearer my-secret-token" \
  -d '{"event_type": "order.placed", "payload": "{\"order_id\": 12345}"}'

Multiple records (newline-delimited), gzip compressed:

echo '{"event_type": "a"}
{"event_type": "b"}
{"event_type": "c"}' | gzip | \
curl -X POST http://localhost:8080/events \
  -H "Content-Encoding: gzip" \
  --data-binary @-

The server automatically decompresses gzip-encoded bodies before processing. No special config is needed - the Content-Encoding: gzip header is sufficient.

Response Codes

The HTTP source returns truthful status codes: 200 on success, 429 when overloaded, and 503 when the pipeline is shutting down. The handler stays non-blocking - the response is sent as soon as the request is enqueued or rejected, without waiting for downstream processing - but clients and load balancers can now back off correctly when the engine sheds load.

Code	Meaning	When it happens
`200`	OK	Request accepted and buffered for processing.
`400`	Bad Request	The request body is empty (zero bytes).
`401`	Unauthorized	Authentication is configured but the `Authorization` header is missing, malformed, or contains incorrect credentials.
`413`	Payload Too Large	The request body exceeds the `max_body_size` limit. Enforced by the Axum body limit layer before the handler runs.
`429`	Too Many Requests	The in-flight concurrency limit or internal buffer is saturated. Clients should back off and retry.
`503`	Service Unavailable	The pipeline is shutting down or its internal channel is closed. Clients should retry against another endpoint.

Clients should treat 429 and 503 as retryable load-shedding signals. For monitoring rejection rates and buffer pressure in production, use the Prometheus metrics laminar_worker_http_source_events_dropped_total and laminar_worker_http_source_buffer_messages.

Complete Example

A webhook ingestion pipeline that accepts JSON events with bearer token authentication, drops malformed records, and handles newline-delimited batches. Table config:

port: 8080
path: /webhooks/events
bind_address: 0.0.0.0
buffer_size: 5000
max_body_size: 10485760
auth:
  type: bearer
  token: "${WEBHOOK_SECRET}"
service_name: webhook-ingest

Schema:

schema:
  format:
    json: {}
  framing:
    newline_delimited: {}
  bad_data:
    drop: {}
  fields:
    - field_name: event_type
      field_type:
        type:
          primitive: Utf8
      nullable: false
    - field_name: payload
      field_type:
        type:
          primitive: Utf8
      nullable: true
    - field_name: timestamp
      field_type:
        type:
          primitive: Utf8
      nullable: false
    - field_name: source_id
      field_type:
        type:
          primitive: Int64
      nullable: true

Sending data:

curl -X POST http://localhost:8080/webhooks/events \
  -H "Authorization: Bearer ${WEBHOOK_SECRET}" \
  -H "Content-Type: application/json" \
  -d '{"event_type": "user.created", "payload": "{\"user_id\": 42}", "timestamp": "2024-01-15T10:30:00Z", "source_id": 1}'

Ingest Engine

Get started

Connectors

Streaming SQL

Operate

Reference

Concepts

Source Config

Authentication

Parallelism Constraint

Relay and Replay

Schema

Format

Framing

Bad Data Handling

Sending Data

Response Codes

Complete Example

JSON Schema Reference

​Source Config

​Authentication

​Parallelism Constraint

​Relay and Replay

​Schema

​Format

​Framing

​Bad Data Handling

​Sending Data

​Response Codes

​Complete Example

​JSON Schema Reference

Source Config

Authentication

Parallelism Constraint

Relay and Replay

Schema

Format

Framing

Bad Data Handling

Sending Data

Response Codes

Complete Example

JSON Schema Reference