risingwave_connector/schema/pulsar_schema/
schema.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 anyhow::Context;
16use apache_avro::Schema;
17use risingwave_common::bail;
18use serde::Deserialize;
19
20use crate::error::ConnectorResult;
21use crate::source::pulsar::topic::{Topic, parse_topic};
22
23#[derive(Debug, Deserialize)]
24pub struct PulsarSchemaInfo {
25    pub version: i64,
26    pub r#type: String,
27    pub data: String,
28}
29
30/// A Pulsar schema by type. Pulsar stores both `AVRO` and `JSON` schemas as Avro record
31/// definitions, but their payloads are Avro binary and plain JSON respectively.
32#[derive(Debug)]
33pub enum PulsarSchema {
34    Avro(Schema),
35    Json(Schema),
36}
37
38impl PulsarSchemaInfo {
39    /// Parses the schema definition according to its type.
40    pub fn parse(&self) -> ConnectorResult<PulsarSchema> {
41        let wrap = match self.r#type.to_ascii_uppercase().as_str() {
42            "AVRO" => PulsarSchema::Avro,
43            "JSON" => PulsarSchema::Json,
44            _ => bail!("unsupported Pulsar schema type {}", self.r#type),
45        };
46        let schema = Schema::parse_str(&self.data)
47            .with_context(|| format!("failed to parse Pulsar {} schema", self.r#type))?;
48        Ok(wrap(schema))
49    }
50}
51
52/// Returns the partitioned topic, i.e. the topic without partition suffix, as schemas are shared
53/// by all partitions.
54pub fn parse_partitioned_topic(topic: &str) -> ConnectorResult<Topic> {
55    let mut topic = parse_topic(topic)?;
56    topic.topic = topic.topic_str_without_partition()?;
57    topic.partition_index = None;
58    Ok(topic)
59}
60
61#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
62pub struct PulsarSchemaVersion(pub i64);
63
64impl TryFrom<&[u8]> for PulsarSchemaVersion {
65    type Error = crate::error::ConnectorError;
66
67    fn try_from(version: &[u8]) -> Result<Self, Self::Error> {
68        let bytes: [u8; 8] = version.try_into().with_context(|| {
69            format!(
70                "expected 8-byte Pulsar LongSchemaVersion, got {} bytes",
71                version.len()
72            )
73        })?;
74        Ok(Self(i64::from_be_bytes(bytes)))
75    }
76}
77
78#[cfg(test)]
79mod tests {
80    use super::*;
81
82    #[test]
83    fn pulsar_schema_info_deserializes_type() {
84        let schema: PulsarSchemaInfo =
85            serde_json::from_str(r#"{"version":0,"type":"AVRO","data":"{}"}"#).unwrap();
86        assert_eq!(schema.r#type, "AVRO");
87    }
88
89    #[test]
90    fn json_schema_definition_is_parsed_as_avro() {
91        let schema: PulsarSchemaInfo = serde_json::from_value(serde_json::json!({
92            "version": 0,
93            "type": "JSON",
94            "data": r#"{"type":"record","name":"Event","fields":[{"name":"id","type":"int"}]}"#,
95        }))
96        .unwrap();
97        assert!(matches!(
98            schema.parse().unwrap(),
99            PulsarSchema::Json(Schema::Record(_))
100        ));
101    }
102
103    #[test]
104    fn unsupported_schema_type_is_rejected() {
105        let schema: PulsarSchemaInfo =
106            serde_json::from_str(r#"{"version":0,"type":"PROTOBUF","data":"{}"}"#).unwrap();
107        assert_eq!(
108            schema.parse().unwrap_err().to_string(),
109            "unsupported Pulsar schema type PROTOBUF"
110        );
111    }
112
113    #[test]
114    fn pulsar_schema_version_is_signed_big_endian() {
115        assert_eq!(
116            PulsarSchemaVersion::try_from(1_i64.to_be_bytes().as_slice())
117                .unwrap()
118                .0,
119            1
120        );
121        assert_eq!(
122            PulsarSchemaVersion::try_from((-1_i64).to_be_bytes().as_slice())
123                .unwrap()
124                .0,
125            -1
126        );
127    }
128
129    #[test]
130    fn pulsar_schema_version_requires_exactly_eight_bytes() {
131        assert!(PulsarSchemaVersion::try_from([].as_slice()).is_err());
132        assert!(PulsarSchemaVersion::try_from([1, 2, 3].as_slice()).is_err());
133        assert!(PulsarSchemaVersion::try_from([0; 9].as_slice()).is_err());
134    }
135}
risingwave_connector/schema/pulsar_schema/schema.rs

risingwave_connector/schema/pulsar_schema/
schema.rs
