Implementation:ArroyoSystems Arroyo Webhook Connector

Knowledge Sources	ArroyoSystems_Arroyo
Domains	Streaming, Connectors
Last Updated	2026-02-08 08:00 GMT

Overview

WebhookConnector implements the Arroyo Connector trait for sending pipeline results as HTTP POST requests to a configurable webhook endpoint with concurrency-limited delivery.

Description

The Webhook connector is a sink-only connector that sends serialized records as HTTP POST requests to a configured endpoint URL. It supports custom HTTP headers (via VarStr with environment variable substitution) and limits concurrent in-flight requests to MAX_INFLIGHT (50) using a Semaphore. The test method validates the endpoint by sending a test JSON message. The make_operator method constructs a WebhookSinkFunc operator with an ArrowSerializer, a shared reqwest::Client with default headers, and error rate limiting via a last_reported_error_at mutex. The endpoint URL supports environment variable substitution via VarStr::sub_env_vars(). During from_options, the connector validates the URL by constructing a test request to check for invalid URLs before creating the connection.

Usage

Use WebhookConnector when you need to push Arroyo pipeline results to external HTTP services, notification systems, or webhook-compatible APIs.

Code Reference

Source Location

• Repository: ArroyoSystems_Arroyo
• File: crates/arroyo-connectors/src/webhook/mod.rs

Signature

pub struct WebhookConnector {}

const MAX_INFLIGHT: u32 = 50;

impl Connector for WebhookConnector {
    type ProfileT = EmptyConfig;
    type TableT = WebhookTable;

    fn name(&self) -> &'static str; // returns "webhook"
    fn table_type(&self, _: EmptyConfig, _: WebhookTable) -> ConnectionType; // always Sink
    fn from_config(&self, id: Option<i64>, name: &str, config: EmptyConfig,
        table: WebhookTable, schema: Option<&ConnectionSchema>) -> anyhow::Result<Connection>;
    fn from_options(&self, name: &str, options: &mut ConnectorOptions,
        schema: Option<&ConnectionSchema>, _: Option<&ConnectionProfile>) -> anyhow::Result<Connection>;
    fn make_operator(&self, _: EmptyConfig, table: WebhookTable,
        config: OperatorConfig) -> anyhow::Result<ConstructedOperator>;
}

Import

use arroyo_connectors::webhook::WebhookConnector;

I/O Contract

Inputs

Name	Type	Required	Description
endpoint	VarStr	Yes	Webhook URL (supports environment variable substitution)
headers	Option<VarStr>	No	Comma-separated colon-delimited header pairs
format	Format	Yes	Serialization format for the POST body
schema	ConnectionSchema	Yes	Schema for the data being sent

Outputs

Name	Type	Description
HTTP POST requests	HTTP responses	Serialized records sent as POST requests to the webhook endpoint

Usage Examples

CREATE TABLE webhook_sink (
    alert TEXT,
    severity TEXT
) WITH (
    connector = 'webhook',
    endpoint = 'https://hooks.example.com/alerts',
    headers = 'Authorization:Bearer {{ WEBHOOK_TOKEN }}',
    format = 'json'
);

Related Pages

• Environment:ArroyoSystems_Arroyo_Rust_Runtime