risingwave_connector/schema/pulsar_schema/
client.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::fmt;
16use std::time::Duration;
17
18use anyhow::Context;
19use reqwest::header::LOCATION;
20use reqwest::redirect::Policy;
21use reqwest::{Client as HttpClient, Response, StatusCode, Url};
22use risingwave_common::bail;
23use risingwave_common::util::retry::exponential_backoff;
24use thiserror_ext::AsReport;
25use tokio_retry::RetryIf;
26use tokio_retry::strategy::jitter;
27
28use super::PulsarSchemaInfo;
29use crate::error::ConnectorResult;
30use crate::schema::{AWS_GLUE_SCHEMA_ARN_KEY, SCHEMA_LOCATION_KEY, SCHEMA_REGISTRY_KEY};
31use crate::source::pulsar::topic::Topic;
32use crate::with_options::{Get, GetKeyIter};
33
34const PULSAR_SCHEMA_PREFIX: &str = "schema.pulsar.";
35pub const PULSAR_SCHEMA_URL_KEY: &str = "schema.pulsar.url";
36pub const PULSAR_SCHEMA_AUTH_TOKEN_KEY: &str = "schema.pulsar.auth.token";
37
38const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
39const DEFAULT_RETRY_DELAY_MS: u64 = 100;
40const DEFAULT_MAX_RETRY_DELAY: Duration = Duration::from_secs(3);
41const DEFAULT_MAX_RETRIES: usize = 3;
42const MAX_REDIRECTS: usize = 5;
43
44#[derive(Debug, thiserror::Error)]
45enum RequestError {
46    #[error("failed to send request")]
47    Send(#[source] reqwest::Error),
48    #[error("failed to read response body")]
49    Body(#[source] reqwest::Error),
50    #[error("request returned HTTP status {0}")]
51    Status(StatusCode),
52    #[error("invalid redirect: {0}")]
53    InvalidRedirect(String),
54    #[error("too many redirects")]
55    TooManyRedirects,
56    #[error("failed to parse response")]
57    Json(#[source] serde_json::Error),
58}
59
60impl RequestError {
61    fn is_retryable(&self) -> bool {
62        match self {
63            Self::Send(_) | Self::Body(_) => true,
64            Self::Status(status) => {
65                *status == StatusCode::TOO_MANY_REQUESTS || status.is_server_error()
66            }
67            Self::InvalidRedirect(_) | Self::TooManyRedirects | Self::Json(_) => false,
68        }
69    }
70}
71
72#[derive(Clone)]
73pub struct PulsarSchemaConfig {
74    url: String,
75    auth_token: Option<String>,
76}
77
78impl fmt::Debug for PulsarSchemaConfig {
79    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
80        f.debug_struct("PulsarSchemaConfig")
81            .field("url", &self.url)
82            .field(
83                "auth_token",
84                &self.auth_token.as_ref().map(|_| "[REDACTED]"),
85            )
86            .finish()
87    }
88}
89
90impl PulsarSchemaConfig {
91    pub fn from_options<T: Get + GetKeyIter>(options: &T) -> ConnectorResult<Option<Self>> {
92        if let Some(option) = options.key_iter().find(|option| {
93            option.starts_with(PULSAR_SCHEMA_PREFIX)
94                && !matches!(
95                    *option,
96                    PULSAR_SCHEMA_URL_KEY | PULSAR_SCHEMA_AUTH_TOKEN_KEY
97                )
98        }) {
99            bail!("unsupported Pulsar schema option `{option}`");
100        }
101
102        let url = options.get(PULSAR_SCHEMA_URL_KEY).cloned();
103        let auth_token = options.get(PULSAR_SCHEMA_AUTH_TOKEN_KEY).cloned();
104
105        let Some(url) = url else {
106            if auth_token.is_some() {
107                bail!("`{PULSAR_SCHEMA_AUTH_TOKEN_KEY}` requires `{PULSAR_SCHEMA_URL_KEY}`");
108            }
109            return Ok(None);
110        };
111        if url.is_empty() {
112            bail!("`{PULSAR_SCHEMA_URL_KEY}` must not be empty");
113        }
114        if auth_token.as_ref().is_some_and(String::is_empty) {
115            bail!("`{PULSAR_SCHEMA_AUTH_TOKEN_KEY}` must not be empty");
116        }
117        if let Some(option) = options.key_iter().find(|option| {
118            matches!(*option, SCHEMA_LOCATION_KEY | AWS_GLUE_SCHEMA_ARN_KEY)
119                || *option == SCHEMA_REGISTRY_KEY
120                || option.starts_with("schema.registry.")
121        }) {
122            bail!("`{option}` cannot be combined with `{PULSAR_SCHEMA_URL_KEY}`");
123        }
124
125        Ok(Some(Self { url, auth_token }))
126    }
127}
128
129/// HTTP client for Pulsar schemas. The configured Admin endpoint and the brokers it redirects to
130/// are trusted to receive the schema bearer token. Authenticated requests require HTTPS at every hop.
131pub struct Client {
132    inner: HttpClient,
133    url: Url,
134    auth_token: Option<String>,
135}
136
137impl fmt::Debug for Client {
138    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
139        f.debug_struct("Client")
140            .field("url", &self.url)
141            .field(
142                "auth_token",
143                &self.auth_token.as_ref().map(|_| "[REDACTED]"),
144            )
145            .finish_non_exhaustive()
146    }
147}
148
149impl Client {
150    pub fn new(config: &PulsarSchemaConfig) -> ConnectorResult<Self> {
151        let url = Url::parse(&config.url).context("invalid Pulsar schema URL")?;
152        if url.cannot_be_a_base() {
153            bail!("Pulsar schema URL must be a base URL");
154        }
155        if !matches!(url.scheme(), "http" | "https") {
156            bail!("Pulsar schema URL must use HTTP or HTTPS");
157        }
158        if config.auth_token.is_some() && url.scheme() != "https" {
159            bail!("Pulsar schema bearer authentication requires HTTPS");
160        }
161        if !url.username().is_empty() || url.password().is_some() {
162            bail!("Pulsar schema URL must not contain credentials");
163        }
164        if url.query().is_some() || url.fragment().is_some() {
165            bail!("Pulsar schema URL must not contain a query or fragment");
166        }
167
168        // Follow Pulsar's broker redirects manually so bearer authentication is retained across
169        // origins. Validate each redirect before sending a new request.
170        let inner = HttpClient::builder()
171            .timeout(DEFAULT_REQUEST_TIMEOUT)
172            .redirect(Policy::none())
173            .build()
174            .context("failed to build Pulsar schema client")?;
175        Ok(Self {
176            inner,
177            url,
178            auth_token: config.auth_token.clone(),
179        })
180    }
181
182    fn build_schema_url(&self, topic: &Topic, version: Option<i64>) -> ConnectorResult<Url> {
183        let mut url = self.url.clone();
184        let mut path = url
185            .path_segments_mut()
186            .map_err(|_| anyhow::anyhow!("Pulsar schema URL must be a base URL"))?;
187        path.pop_if_empty().extend([
188            "admin",
189            "v2",
190            "schemas",
191            topic.tenant.as_str(),
192            topic.namespace.as_str(),
193            topic.topic.as_str(),
194            "schema",
195        ]);
196        if let Some(version) = version {
197            path.push(version.to_string().as_str());
198        }
199        drop(path);
200        Ok(url)
201    }
202
203    fn redirect_url(current_url: &Url, response: &Response) -> Result<Url, RequestError> {
204        let location = response
205            .headers()
206            .get(LOCATION)
207            .ok_or_else(|| RequestError::InvalidRedirect("missing Location header".to_owned()))?
208            .to_str()
209            .map_err(|error| RequestError::InvalidRedirect(error.to_report_string()))?;
210        let redirect_url = current_url
211            .join(location)
212            .map_err(|error| RequestError::InvalidRedirect(error.to_report_string()))?;
213        if !matches!(redirect_url.scheme(), "http" | "https") {
214            return Err(RequestError::InvalidRedirect(format!(
215                "unsupported URL scheme `{}`",
216                redirect_url.scheme()
217            )));
218        }
219        if current_url.scheme() == "https" && redirect_url.scheme() == "http" {
220            return Err(RequestError::InvalidRedirect(
221                "refusing to redirect from HTTPS to HTTP".to_owned(),
222            ));
223        }
224        if !redirect_url.username().is_empty() || redirect_url.password().is_some() {
225            return Err(RequestError::InvalidRedirect(
226                "redirect URL must not contain credentials".to_owned(),
227            ));
228        }
229        Ok(redirect_url)
230    }
231
232    async fn request_schema(&self, url: &Url) -> Result<PulsarSchemaInfo, RequestError> {
233        let mut request_url = url.clone();
234        let mut redirects = 0;
235        loop {
236            let mut request = self.inner.get(request_url.clone());
237            if let Some(token) = self.auth_token.as_ref() {
238                request = request.bearer_auth(token);
239            }
240
241            let response = request.send().await.map_err(RequestError::Send)?;
242            if matches!(
243                response.status(),
244                StatusCode::MOVED_PERMANENTLY
245                    | StatusCode::FOUND
246                    | StatusCode::SEE_OTHER
247                    | StatusCode::TEMPORARY_REDIRECT
248                    | StatusCode::PERMANENT_REDIRECT
249            ) {
250                if redirects >= MAX_REDIRECTS {
251                    return Err(RequestError::TooManyRedirects);
252                }
253                redirects += 1;
254                request_url = Self::redirect_url(&request_url, &response)?;
255                continue;
256            }
257            if !response.status().is_success() {
258                return Err(RequestError::Status(response.status()));
259            }
260            let body = response.bytes().await.map_err(RequestError::Body)?;
261            return serde_json::from_slice(&body).map_err(RequestError::Json);
262        }
263    }
264
265    /// Fetches the schema of `topic`, which should be the non-partitioned topic since all
266    /// partitions share the same schema.
267    pub async fn get_schema(
268        &self,
269        topic: &Topic,
270        version: Option<i64>,
271    ) -> ConnectorResult<PulsarSchemaInfo> {
272        let url = self.build_schema_url(topic, version)?;
273        let retry_strategy = exponential_backoff(
274            Duration::from_millis(DEFAULT_RETRY_DELAY_MS),
275            2,
276            DEFAULT_MAX_RETRY_DELAY,
277        )
278        .take(DEFAULT_MAX_RETRIES)
279        .map(jitter);
280        RetryIf::spawn(
281            retry_strategy,
282            || self.request_schema(&url),
283            |error: &RequestError| {
284                let retryable = error.is_retryable();
285                if retryable {
286                    tracing::debug!(error = %error.as_report(), "retrying Pulsar schema request");
287                }
288                retryable
289            },
290        )
291        .await
292        .map_err(|error| {
293            anyhow::Error::new(error)
294                .context(format!("failed to fetch Pulsar schema from {url}"))
295                .into()
296        })
297    }
298}
299
300#[cfg(test)]
301mod tests {
302    use std::collections::BTreeMap;
303    #[cfg(not(madsim))]
304    use std::io::{Read, Write};
305    #[cfg(not(madsim))]
306    use std::net::TcpListener;
307    #[cfg(not(madsim))]
308    use std::sync::mpsc;
309    #[cfg(not(madsim))]
310    use std::thread;
311
312    use super::*;
313    use crate::source::pulsar::topic::parse_topic;
314
315    fn topic(topic: &str) -> Topic {
316        parse_topic(topic).unwrap()
317    }
318
319    fn config(url: String, token: Option<&str>) -> PulsarSchemaConfig {
320        let mut options = BTreeMap::from([(PULSAR_SCHEMA_URL_KEY.to_owned(), url)]);
321        if let Some(token) = token {
322            options.insert(PULSAR_SCHEMA_AUTH_TOKEN_KEY.to_owned(), token.to_owned());
323        }
324        PulsarSchemaConfig::from_options(&options).unwrap().unwrap()
325    }
326
327    fn client() -> Client {
328        Client::new(&config("http://localhost:8080".to_owned(), None)).unwrap()
329    }
330
331    #[test]
332    fn schema_url_from_full_topic() {
333        let client = client();
334        assert_eq!(
335            client
336                .build_schema_url(&topic("persistent://tenant/ns/events"), None)
337                .unwrap()
338                .as_str(),
339            "http://localhost:8080/admin/v2/schemas/tenant/ns/events/schema"
340        );
341        assert_eq!(
342            client
343                .build_schema_url(&topic("persistent://tenant/ns/events"), Some(42))
344                .unwrap()
345                .as_str(),
346            "http://localhost:8080/admin/v2/schemas/tenant/ns/events/schema/42"
347        );
348    }
349
350    #[test]
351    fn schema_url_preserves_base_paths_with_or_without_a_trailing_slash() {
352        for base in [
353            "http://localhost:8080/pulsar",
354            "http://localhost:8080/pulsar/",
355        ] {
356            let client = Client::new(&config(base.to_owned(), None)).unwrap();
357            assert_eq!(
358                client
359                    .build_schema_url(&topic("events"), Some(42))
360                    .unwrap()
361                    .as_str(),
362                "http://localhost:8080/pulsar/admin/v2/schemas/public/default/events/schema/42"
363            );
364        }
365    }
366
367    #[test]
368    fn bearer_authentication_requires_https() {
369        let error = Client::new(&config(
370            "http://localhost:8080".to_owned(),
371            Some("test-token"),
372        ))
373        .unwrap_err();
374        assert!(error.to_string().contains("requires HTTPS"));
375        assert!(
376            Client::new(&config(
377                "https://localhost:8443".to_owned(),
378                Some("test-token")
379            ))
380            .is_ok()
381        );
382    }
383
384    #[test]
385    fn schema_url_from_short_and_escaped_topics() {
386        let client = client();
387        assert_eq!(
388            client
389                .build_schema_url(&topic("events"), None)
390                .unwrap()
391                .as_str(),
392            "http://localhost:8080/admin/v2/schemas/public/default/events/schema"
393        );
394        assert_eq!(
395            client
396                .build_schema_url(&topic("persistent://tenant/ns/events?region=us"), None)
397                .unwrap()
398                .as_str(),
399            "http://localhost:8080/admin/v2/schemas/tenant/ns/events%3Fregion=us/schema"
400        );
401    }
402
403    #[test]
404    fn config_is_optional_and_requires_url_for_token() {
405        assert!(
406            PulsarSchemaConfig::from_options(&BTreeMap::<String, String>::new())
407                .unwrap()
408                .is_none()
409        );
410        let options =
411            BTreeMap::from([(PULSAR_SCHEMA_AUTH_TOKEN_KEY.to_owned(), "token".to_owned())]);
412        assert!(PulsarSchemaConfig::from_options(&options).is_err());
413    }
414
415    #[test]
416    fn config_rejects_unknown_and_overlapping_options() {
417        for option in [
418            "schema.pulsar.ca",
419            SCHEMA_REGISTRY_KEY,
420            "schema.registry.username",
421            SCHEMA_LOCATION_KEY,
422            AWS_GLUE_SCHEMA_ARN_KEY,
423        ] {
424            let options = BTreeMap::from([
425                (
426                    PULSAR_SCHEMA_URL_KEY.to_owned(),
427                    "http://localhost:8080".to_owned(),
428                ),
429                (option.to_owned(), "value".to_owned()),
430            ]);
431            assert!(
432                PulsarSchemaConfig::from_options(&options).is_err(),
433                "expected `{option}` to be rejected"
434            );
435        }
436    }
437
438    #[test]
439    fn schema_auth_is_separate_from_broker_auth() {
440        let options = BTreeMap::from([
441            (
442                PULSAR_SCHEMA_URL_KEY.to_owned(),
443                "http://localhost:8080".to_owned(),
444            ),
445            (
446                PULSAR_SCHEMA_AUTH_TOKEN_KEY.to_owned(),
447                "schema-token".to_owned(),
448            ),
449            ("auth.token".to_owned(), "broker-token".to_owned()),
450        ]);
451        let config = PulsarSchemaConfig::from_options(&options).unwrap().unwrap();
452        assert_eq!(config.auth_token.as_deref(), Some("schema-token"));
453    }
454
455    #[test]
456    fn config_debug_redacts_token() {
457        let config = config("http://localhost:8080".to_owned(), Some("secret-token"));
458        let debug = format!("{config:?}");
459        assert!(!debug.contains("secret-token"));
460        assert!(debug.contains("[REDACTED]"));
461    }
462
463    #[cfg(not(madsim))]
464    fn spawn_http_server(
465        responses: Vec<String>,
466    ) -> (String, mpsc::Receiver<String>, thread::JoinHandle<()>) {
467        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
468        let addr = listener.local_addr().unwrap();
469        let (request_tx, request_rx) = mpsc::channel();
470        let handle = thread::spawn(move || {
471            for response in responses {
472                let (mut stream, _) = listener.accept().unwrap();
473                let mut request = Vec::new();
474                loop {
475                    let mut buffer = [0; 1024];
476                    let read = stream.read(&mut buffer).unwrap();
477                    assert_ne!(read, 0);
478                    request.extend_from_slice(&buffer[..read]);
479                    if request.windows(4).any(|window| window == b"\r\n\r\n") {
480                        break;
481                    }
482                }
483                request_tx
484                    .send(String::from_utf8(request).unwrap())
485                    .unwrap();
486                stream.write_all(response.as_bytes()).unwrap();
487            }
488        });
489        (format!("http://{addr}"), request_rx, handle)
490    }
491
492    #[cfg(not(madsim))]
493    fn response(status: &str, body: &str) -> String {
494        format!(
495            "HTTP/1.1 {status}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
496            body.len()
497        )
498    }
499
500    #[cfg(not(madsim))]
501    #[tokio::test]
502    async fn anonymous_cross_origin_redirect_is_followed() {
503        let body = r#"{"version":1,"type":"AVRO","data":"{}"}"#;
504        let (target_url, target_requests, target_handle) =
505            spawn_http_server(vec![response("200 OK", body)]);
506        let redirect = format!(
507            "HTTP/1.1 307 Temporary Redirect\r\nLocation: {target_url}/schema\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
508        );
509        let (admin_url, admin_requests, admin_handle) = spawn_http_server(vec![redirect]);
510        let client = Client::new(&config(admin_url, None)).unwrap();
511
512        let schema = client
513            .get_schema(&topic("tenant/ns/events"), None)
514            .await
515            .unwrap();
516        assert_eq!(schema.version, 1);
517        admin_requests.recv().unwrap();
518        let redirected_request = target_requests.recv().unwrap().to_ascii_lowercase();
519        assert!(redirected_request.starts_with("get /schema "));
520        assert!(!redirected_request.contains("authorization:"));
521        admin_handle.join().unwrap();
522        target_handle.join().unwrap();
523    }
524
525    #[cfg(not(madsim))]
526    #[tokio::test]
527    async fn authenticated_redirects_follow_https_brokers() {
528        let client = Client::new(&config(
529            "https://admin.example:8443".to_owned(),
530            Some("test-token"),
531        ))
532        .unwrap();
533        for (location, allowed) in [
534            ("/schema", true),
535            ("https://admin.example:8443/schema?authoritative=true", true),
536            ("https://other.example:8443/schema?authoritative=true", true),
537            ("https://admin.example:8444/schema", true),
538            ("http://admin.example:8443/schema", false),
539            ("http://other.example:8080/schema", false),
540            ("file:///tmp/schema", false),
541            ("https://user:password@admin.example:8443/schema", false),
542            ("https://user:password@other.example:8443/schema", false),
543        ] {
544            let redirect = format!(
545                "HTTP/1.1 307 Temporary Redirect\r\nLocation: {location}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
546            );
547            let (url, _requests, handle) = spawn_http_server(vec![redirect]);
548            let response = HttpClient::builder()
549                .redirect(Policy::none())
550                .build()
551                .unwrap()
552                .get(url)
553                .send()
554                .await
555                .unwrap();
556            let result = Client::redirect_url(&client.url, &response);
557            assert_eq!(result.is_ok(), allowed, "{location}");
558            if let Err(error) = result {
559                assert!(!error.is_retryable());
560            }
561            handle.join().unwrap();
562        }
563    }
564
565    #[cfg(not(madsim))]
566    #[tokio::test]
567    async fn interrupted_response_body_is_retried() {
568        let body = r#"{"version":2,"type":"AVRO","data":"{}"}"#;
569        let (admin_url, requests, handle) = spawn_http_server(vec![
570            "HTTP/1.1 200 OK\r\nContent-Length: 999\r\nConnection: close\r\n\r\n{\"version\":2"
571                .to_owned(),
572            response("200 OK", body),
573        ]);
574        let client = Client::new(&config(admin_url, None)).unwrap();
575
576        assert_eq!(
577            client
578                .get_schema(&topic("events"), None)
579                .await
580                .unwrap()
581                .version,
582            2
583        );
584        requests.recv().unwrap();
585        requests.recv().unwrap();
586        assert!(requests.try_recv().is_err());
587        handle.join().unwrap();
588    }
589
590    #[cfg(not(madsim))]
591    #[tokio::test]
592    async fn server_error_is_retried() {
593        let body = r#"{"version":2,"type":"AVRO","data":"{}"}"#;
594        let (admin_url, requests, handle) = spawn_http_server(vec![
595            response("503 Service Unavailable", ""),
596            response("200 OK", body),
597        ]);
598        let client = Client::new(&config(admin_url, None)).unwrap();
599
600        let schema = client
601            .get_schema(&topic("tenant/ns/events"), None)
602            .await
603            .unwrap();
604        assert_eq!(schema.version, 2);
605        requests.recv().unwrap();
606        requests.recv().unwrap();
607        assert!(requests.try_recv().is_err());
608        handle.join().unwrap();
609    }
610
611    #[cfg(not(madsim))]
612    #[tokio::test]
613    async fn client_error_and_malformed_body_are_not_retried() {
614        for response in [response("404 Not Found", ""), response("200 OK", "{")] {
615            let (admin_url, requests, handle) = spawn_http_server(vec![response]);
616            let client = Client::new(&config(admin_url, None)).unwrap();
617
618            client
619                .get_schema(&topic("tenant/ns/events"), None)
620                .await
621                .unwrap_err();
622            requests.recv().unwrap();
623            assert!(requests.try_recv().is_err());
624            handle.join().unwrap();
625        }
626    }
627}
risingwave_connector/schema/pulsar_schema/client.rs

risingwave_connector/schema/pulsar_schema/
client.rs
