risingwave_connector/sink/
http.rs

1// Copyright 2026 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::BTreeMap;
16
17use anyhow::{Context, anyhow};
18use reqwest::header::{CONTENT_TYPE, HeaderMap, HeaderName, HeaderValue};
19use risingwave_common::array::{Op, StreamChunk};
20use risingwave_common::catalog::Schema;
21use risingwave_common::row::Row;
22use risingwave_common::types::{DataType, JsonbRef, ScalarRefImpl};
23use serde::Deserialize;
24use thiserror_ext::AsReport;
25use with_options::WithOptions;
26
27use crate::enforce_secret::EnforceSecret;
28use crate::sink::log_store::DeliveryFutureManagerAddFuture;
29use crate::sink::writer::{
30    AsyncTruncateLogSinkerOf, AsyncTruncateSinkWriter, AsyncTruncateSinkWriterExt,
31};
32use crate::sink::{Result, SINK_TYPE_APPEND_ONLY, Sink, SinkError, SinkParam, SinkWriterParam};
33
34pub const HTTP_SINK: &str = "http";
35const HTTP_SINK_PAYLOAD_COLUMN: &str = "payload";
36const HTTP_SINK_URL_COLUMN: &str = "url";
37
38#[derive(Clone, Debug, Deserialize, WithOptions)]
39pub struct HttpConfig {
40    /// The endpoint URL to POST data to.
41    pub url: Option<String>,
42
43    /// Content-Type header value. Defaults to `text/plain` for `varchar` and `application/json`
44    /// for `jsonb`.
45    pub content_type: Option<String>,
46
47    /// Sink type, must be "append-only".
48    pub r#type: String,
49
50    #[serde(flatten)]
51    pub unknown_fields: std::collections::HashMap<String, String>,
52}
53
54crate::impl_sink_unknown_fields!(HttpConfig);
55
56impl EnforceSecret for HttpConfig {}
57
58impl HttpConfig {
59    pub fn from_btreemap(
60        values: BTreeMap<String, String>,
61    ) -> Result<(Self, BTreeMap<String, String>)> {
62        // Extract header.* keys before serde parsing
63        let mut headers = BTreeMap::new();
64        let mut rest = BTreeMap::new();
65        for (k, v) in &values {
66            if let Some(header_name) = k.strip_prefix("header.") {
67                headers.insert(header_name.to_owned(), v.clone());
68            } else {
69                rest.insert(k.clone(), v.clone());
70            }
71        }
72
73        let config = serde_json::from_value::<HttpConfig>(serde_json::to_value(rest).unwrap())
74            .map_err(|e| SinkError::Config(anyhow!(e)))?;
75
76        if config.r#type != SINK_TYPE_APPEND_ONLY {
77            return Err(SinkError::Config(anyhow!(
78                "HTTP sink only supports append-only mode"
79            )));
80        }
81
82        Ok((config, headers))
83    }
84}
85
86#[derive(Clone, Debug)]
87enum HttpUrl {
88    Static(reqwest::Url),
89    Dynamic { url_index: usize },
90}
91
92/// Validates the HTTP sink parameters and returns the sink so callers can use it directly without
93/// re-parsing.
94fn validate_http_sink(
95    is_append_only: bool,
96    ignore_delete: bool,
97    schema: &Schema,
98    url: Option<&str>,
99    content_type: Option<&str>,
100    headers: &BTreeMap<String, String>,
101    unknown_fields: std::collections::HashMap<String, String>,
102) -> Result<HttpSink> {
103    if !is_append_only && !ignore_delete {
104        return Err(SinkError::Config(anyhow!(
105            "HTTP sink only supports append-only mode"
106        )));
107    }
108
109    let fields = schema.fields();
110    let (payload_index, url, payload_type) = if fields.len() == 1 {
111        let Some(url) = url else {
112            return Err(SinkError::Config(anyhow!(
113                "HTTP sink requires url option when schema has exactly 1 column"
114            )));
115        };
116        let url = url
117            .parse()
118            .context("invalid URL")
119            .map_err(SinkError::Config)?;
120        (0, HttpUrl::Static(url), fields[0].data_type.clone())
121    } else {
122        for field in fields {
123            match field.name.as_str() {
124                HTTP_SINK_PAYLOAD_COLUMN | HTTP_SINK_URL_COLUMN => {}
125                _ => {
126                    return Err(SinkError::Config(anyhow!(
127                        "HTTP sink with multiple columns only supports payload and url columns, got {}",
128                        field.name
129                    )));
130                }
131            }
132        }
133
134        let payload_index = fields
135            .iter()
136            .position(|field| field.name == HTTP_SINK_PAYLOAD_COLUMN)
137            .ok_or_else(|| {
138                SinkError::Config(anyhow!(
139                    "HTTP sink with multiple columns requires a payload column"
140                ))
141            })?;
142        let url_index = fields
143            .iter()
144            .position(|field| field.name == HTTP_SINK_URL_COLUMN);
145        let url = match (url, url_index) {
146            (Some(_), Some(_)) => {
147                return Err(SinkError::Config(anyhow!(
148                    "HTTP sink url option cannot coexist with url column"
149                )));
150            }
151            (Some(url), None) => {
152                let url = url
153                    .parse()
154                    .context("invalid URL")
155                    .map_err(SinkError::Config)?;
156                HttpUrl::Static(url)
157            }
158            (None, Some(url_index)) => {
159                if fields[url_index].data_type != DataType::Varchar {
160                    return Err(SinkError::Config(anyhow!(
161                        "HTTP sink url column must be varchar, got {:?}",
162                        fields[url_index].data_type
163                    )));
164                }
165                HttpUrl::Dynamic { url_index }
166            }
167            (None, None) => {
168                return Err(SinkError::Config(anyhow!(
169                    "HTTP sink requires either url option or url column"
170                )));
171            }
172        };
173
174        (payload_index, url, fields[payload_index].data_type.clone())
175    };
176
177    if payload_type != DataType::Varchar && payload_type != DataType::Jsonb {
178        return Err(SinkError::Config(anyhow!(
179            "HTTP sink payload column must be varchar or jsonb, got {:?}",
180            payload_type
181        )));
182    }
183
184    let mut header_map = HeaderMap::new();
185    header_map.insert(
186        CONTENT_TYPE,
187        content_type
188            .unwrap_or(match payload_type {
189                DataType::Varchar => "text/plain",
190                DataType::Jsonb => "application/json",
191                _ => unreachable!("validated HTTP sink column type"),
192            })
193            .parse()
194            .context("invalid content_type")
195            .map_err(SinkError::Config)?,
196    );
197    for (k, v) in headers {
198        let name: HeaderName = k
199            .parse()
200            .with_context(|| format!("invalid header name '{k}'"))
201            .map_err(SinkError::Config)?;
202        let value: HeaderValue = v
203            .parse()
204            .with_context(|| format!("invalid header value for '{k}'"))
205            .map_err(SinkError::Config)?;
206        header_map.insert(name, value);
207    }
208
209    Ok(HttpSink {
210        url,
211        payload_index,
212        header_map,
213        unknown_fields,
214    })
215}
216
217#[derive(Clone, Debug)]
218pub struct HttpSink {
219    url: HttpUrl,
220    payload_index: usize,
221    header_map: HeaderMap,
222    unknown_fields: std::collections::HashMap<String, String>,
223}
224
225impl EnforceSecret for HttpSink {
226    fn enforce_secret<'a>(
227        prop_iter: impl Iterator<Item = &'a str>,
228    ) -> crate::error::ConnectorResult<()> {
229        for prop in prop_iter {
230            HttpConfig::enforce_one(prop)?;
231        }
232        Ok(())
233    }
234}
235
236impl TryFrom<SinkParam> for HttpSink {
237    type Error = SinkError;
238
239    fn try_from(param: SinkParam) -> std::result::Result<Self, Self::Error> {
240        let schema = param.schema();
241        let (config, headers) = HttpConfig::from_btreemap(param.properties)?;
242        validate_http_sink(
243            param.sink_type.is_append_only(),
244            param.ignore_delete,
245            &schema,
246            config.url.as_deref(),
247            config.content_type.as_deref(),
248            &headers,
249            config.unknown_fields,
250        )
251    }
252}
253
254impl Sink for HttpSink {
255    type LogSinker = AsyncTruncateLogSinkerOf<HttpSinkWriter>;
256
257    const SINK_NAME: &'static str = HTTP_SINK;
258
259    fn validate_unknown_fields(&self) -> Result<()> {
260        crate::sink::validate_sink_unknown_fields(&self.unknown_fields)
261    }
262
263    async fn validate(&self) -> Result<()> {
264        Ok(())
265    }
266
267    async fn new_log_sinker(&self, _writer_param: SinkWriterParam) -> Result<Self::LogSinker> {
268        Ok(HttpSinkWriter::new(
269            self.url.clone(),
270            self.payload_index,
271            self.header_map.clone(),
272        )?
273        .into_log_sinker(usize::MAX))
274    }
275}
276
277pub struct HttpSinkWriter {
278    client: reqwest::Client,
279    url: HttpUrl,
280    payload_index: usize,
281}
282
283impl HttpSinkWriter {
284    fn new(url: HttpUrl, payload_index: usize, header_map: HeaderMap) -> Result<Self> {
285        let client = reqwest::Client::builder()
286            .default_headers(header_map)
287            .build()
288            .context("failed to build HTTP client")
289            .map_err(SinkError::Http)?;
290
291        Ok(Self {
292            client,
293            url,
294            payload_index,
295        })
296    }
297
298    fn extract_url(&self, row: &impl Row) -> Result<Option<reqwest::Url>> {
299        match &self.url {
300            HttpUrl::Static(url) => Ok(Some(url.clone())),
301            HttpUrl::Dynamic { url_index } => match row.datum_at(*url_index) {
302                Some(ScalarRefImpl::Utf8(url)) if !url.is_empty() => {
303                    match url.parse::<reqwest::Url>() {
304                        Ok(url) => Ok(Some(url)),
305                        Err(err) => {
306                            tracing::warn!(
307                                error = %err.as_report(),
308                                payload = %self.strip_payload_for_log(row),
309                                "skip HTTP sink row due to invalid URL in url column"
310                            );
311                            Ok(None)
312                        }
313                    }
314                }
315                Some(ScalarRefImpl::Utf8(_)) | None => {
316                    tracing::warn!(
317                        payload = %self.strip_payload_for_log(row),
318                        "skip HTTP sink row due to null or empty url column"
319                    );
320                    Ok(None)
321                }
322                Some(_) => Err(SinkError::Http(anyhow!(
323                    "unexpected url column type, expected varchar"
324                ))),
325            },
326        }
327    }
328
329    fn strip_payload_for_log(&self, row: &impl Row) -> String {
330        match row.datum_at(self.payload_index) {
331            Some(ScalarRefImpl::Utf8(s)) => strip_text_payload(s),
332            Some(ScalarRefImpl::Jsonb(j)) => strip_jsonb_payload(j),
333            Some(_) => "<unexpected payload type>".to_owned(),
334            None => "NULL".to_owned(),
335        }
336    }
337
338    fn extract_payload(&self, row: &impl Row) -> Result<Option<String>> {
339        Ok(match row.datum_at(self.payload_index) {
340            Some(ScalarRefImpl::Utf8(s)) => Some(s.to_owned()),
341            Some(ScalarRefImpl::Jsonb(j)) => Some(j.to_string()),
342            Some(_) => {
343                return Err(SinkError::Http(anyhow!(
344                    "unexpected payload column type, expected varchar or jsonb"
345                )));
346            }
347            None => None, // skip NULL rows
348        })
349    }
350}
351
352fn strip_text_payload(payload: &str) -> String {
353    const EDGE_CHAR_COUNT: usize = 100;
354    let char_count = payload.chars().count();
355    if char_count <= EDGE_CHAR_COUNT * 2 {
356        return payload.to_owned();
357    }
358
359    let prefix: String = payload.chars().take(EDGE_CHAR_COUNT).collect();
360    let suffix: String = payload.chars().skip(char_count - EDGE_CHAR_COUNT).collect();
361    format!("{prefix}...{suffix}")
362}
363
364fn strip_jsonb_payload(payload: JsonbRef<'_>) -> String {
365    if payload.is_array() {
366        let len = payload.array_len().expect("checked JSON array type");
367        return match len {
368            0 => "[]".to_owned(),
369            1 => {
370                let first = payload.access_array_element(0).expect("JSON array element");
371                format!("[{}]", strip_jsonb_payload(first))
372            }
373            2 => {
374                let first = payload.access_array_element(0).expect("JSON array element");
375                let second = payload.access_array_element(1).expect("JSON array element");
376                format!(
377                    "[{},{}]",
378                    strip_jsonb_payload(first),
379                    strip_jsonb_payload(second)
380                )
381            }
382            _ => {
383                let first = payload.access_array_element(0).expect("JSON array element");
384                let last = payload
385                    .access_array_element(len - 1)
386                    .expect("JSON array element");
387                format!(
388                    "[{},...,{}]",
389                    strip_jsonb_payload(first),
390                    strip_jsonb_payload(last)
391                )
392            }
393        };
394    }
395
396    if let Ok(fields) = payload.object_key_values() {
397        let fields = fields
398            .map(|(key, value)| {
399                let key = serde_json::to_string(key).expect("serialize JSON object key");
400                format!("{key}:{}", strip_jsonb_payload(value))
401            })
402            .collect::<Vec<_>>();
403        return format!("{{{}}}", fields.join(","));
404    }
405
406    if let Ok(value) = payload.as_str() {
407        return serde_json::to_string(&strip_text_payload(value)).expect("serialize JSON string");
408    }
409
410    payload.to_string()
411}
412
413impl AsyncTruncateSinkWriter for HttpSinkWriter {
414    async fn write_chunk<'a>(
415        &'a mut self,
416        chunk: StreamChunk,
417        _add_future: DeliveryFutureManagerAddFuture<'a, Self::DeliveryFuture>,
418    ) -> Result<()> {
419        for (op, row) in chunk.rows() {
420            if op != Op::Insert {
421                continue;
422            }
423
424            let Some(payload) = self.extract_payload(&row)? else {
425                continue;
426            };
427            let Some(url) = self.extract_url(&row)? else {
428                continue;
429            };
430
431            let resp = self
432                .client
433                .post(url)
434                .body(payload)
435                .send()
436                .await
437                .context("HTTP request failed")
438                .map_err(SinkError::Http)?;
439
440            if !resp.status().is_success() {
441                let status = resp.status();
442                let body = resp.text().await.unwrap_or_default();
443                return Err(SinkError::Http(anyhow!(
444                    "HTTP sink received non-success response: {} {}",
445                    status,
446                    body
447                )));
448            }
449        }
450
451        Ok(())
452    }
453}
454
455#[cfg(test)]
456mod tests {
457    use risingwave_common::types::{JsonbVal, Scalar};
458
459    use super::*;
460
461    #[test]
462    fn test_strip_text_payload() {
463        assert_eq!(strip_text_payload("short payload"), "short payload");
464
465        let payload = format!("{}{}", "a".repeat(100), "z".repeat(100));
466        assert_eq!(strip_text_payload(&payload), payload);
467
468        let payload = format!("{}middle{}", "a".repeat(101), "z".repeat(101));
469        assert_eq!(
470            strip_text_payload(&payload),
471            format!("{}...{}", "a".repeat(100), "z".repeat(100))
472        );
473    }
474
475    #[test]
476    fn test_strip_jsonb_payload() {
477        let payload: JsonbVal = r#"{
478            "id": 1,
479            "items": [1, {"nested": ["first", "middle", "last"]}, 3],
480            "message": "short"
481        }"#
482        .parse()
483        .unwrap();
484
485        assert_eq!(
486            strip_jsonb_payload(payload.as_scalar_ref()),
487            r#"{"id":1,"items":[1,...,3],"message":"short"}"#
488        );
489
490        let payload: JsonbVal = r#"["only"]"#.parse().unwrap();
491        assert_eq!(strip_jsonb_payload(payload.as_scalar_ref()), r#"["only"]"#);
492
493        let payload: JsonbVal = r#"["first","last"]"#.parse().unwrap();
494        assert_eq!(
495            strip_jsonb_payload(payload.as_scalar_ref()),
496            r#"["first","last"]"#
497        );
498    }
499}
risingwave_connector/sink/http.rs

risingwave_connector/sink/
http.rs
