Implementation:ArroyoSystems Arroyo Polling Http Operator

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

Overview

PollingHttpSourceFunc is the Arroyo source operator that periodically polls an HTTP endpoint, deserializes the response, and emits records with configurable deduplication behavior and checkpointed state.

Description

The operator uses a reqwest::Client to send HTTP requests at a configurable polling_interval using a tokio::time::interval with MissedTickBehavior::Delay. Only the first parallel task (task_index == 0) performs polling; all other tasks mark themselves idle with a Watermark::Idle signal. The operator supports EmitBehavior::All (emit every response) and EmitBehavior::Changed (skip responses identical to the previous one). State is maintained in PollingHttpSourceState containing the last_message bytes, which is persisted to global keyed state during checkpoints for exactly-once semantics. The request method enforces a MAX_BODY_SIZE limit of 5MB, streaming the response body in chunks and bailing if the limit is exceeded. Non-successful HTTP status codes are logged as nonfatal errors with exponential backoff via connector_err!. The operator handles control messages (Checkpoint, Stop, LoadCompacted, NoOp) in the select loop.

Usage

Use PollingHttpSourceFunc as the runtime operator backing the Polling HTTP connector for periodic HTTP data ingestion in Arroyo pipelines.

Code Reference

Source Location

Repository: ArroyoSystems_Arroyo
File: crates/arroyo-connectors/src/polling_http/operator.rs

Signature

pub struct PollingHttpSourceFunc {
    pub state: PollingHttpSourceState,
    pub client: reqwest::Client,
    pub endpoint: url::Url,
    pub method: reqwest::Method,
    pub body: Option<Bytes>,
    pub polling_interval: Duration,
    pub emit_behavior: EmitBehavior,
    pub format: Format,
    pub framing: Option<Framing>,
    pub bad_data: Option<BadData>,
}

#[derive(Clone, Debug, Encode, Decode, PartialEq, PartialOrd, Default)]
pub struct PollingHttpSourceState {
    last_message: Option<Vec<u8>>,
}

#[async_trait]
impl SourceOperator for PollingHttpSourceFunc {
    fn name(&self) -> String; // "PollingHttpSource"
    fn tables(&self) -> HashMap<String, TableConfig>;
    async fn on_start(&mut self, ctx: &mut SourceContext) -> DataflowResult<()>;
    async fn run(&mut self, ctx: &mut SourceContext, collector: &mut SourceCollector)
        -> DataflowResult<SourceFinishType>;
}

Import

use arroyo_connectors::polling_http::operator::{PollingHttpSourceFunc, PollingHttpSourceState};

I/O Contract

Inputs

Name	Type	Required	Description
endpoint	url::Url	Yes	HTTP URL to poll
method	reqwest::Method	Yes	HTTP method (GET, POST, PUT, PATCH)
polling_interval	Duration	Yes	Time between polls
emit_behavior	EmitBehavior	Yes	All or Changed
format	Format	Yes	Deserialization format

Outputs

Name	Type	Description
records	RecordBatch	Deserialized Arrow record batches from HTTP response data

Usage Examples

let source = PollingHttpSourceFunc {
    state: PollingHttpSourceState::default(),
    client: reqwest::Client::new(),
    endpoint: url::Url::parse("https://api.example.com/data").unwrap(),
    method: reqwest::Method::GET,
    body: None,
    polling_interval: Duration::from_secs(5),
    emit_behavior: EmitBehavior::Changed,
    format: Format::Json(JsonFormat::default()),
    framing: None,
    bad_data: None,
};

Related Pages

Environment:ArroyoSystems_Arroyo_Rust_Runtime

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment