faucet_source_kafka/
decode.rs

1//! Decode Kafka message bytes into JSON `Value`s, dispatching on
2//! `KafkaValueFormat`.
3
4use base64::Engine;
5use faucet_common_kafka::KafkaValueFormat;
6use faucet_core::FaucetError;
7use serde_json::Value;
8
9#[cfg(feature = "schema-registry")]
10use faucet_common_kafka::schema_registry::{
11    avro, client::SchemaRegistryClient, json_schema, protobuf,
12};
13
14/// Decode `bytes` per `format`. For Confluent SR formats, `sr_client` must be `Some`.
15pub async fn decode(
16    bytes: Option<&[u8]>,
17    format: &KafkaValueFormat,
18    #[cfg(feature = "schema-registry")] sr_client: Option<&SchemaRegistryClient>,
19) -> Result<Value, FaucetError> {
20    let Some(bytes) = bytes else {
21        return Ok(Value::Null);
22    };
23    match format {
24        KafkaValueFormat::Json => serde_json::from_slice(bytes)
25            .map_err(|e| FaucetError::Source(format!("kafka json decode: {e}"))),
26        KafkaValueFormat::RawString => {
27            let s = std::str::from_utf8(bytes)
28                .map_err(|e| FaucetError::Source(format!("kafka raw_string decode: {e}")))?;
29            Ok(Value::String(s.to_string()))
30        }
31        KafkaValueFormat::Bytes => {
32            let encoded = base64::engine::general_purpose::STANDARD.encode(bytes);
33            Ok(Value::String(encoded))
34        }
35        #[cfg(feature = "schema-registry")]
36        KafkaValueFormat::ConfluentAvro { .. } => {
37            let client = sr_client.ok_or_else(|| {
38                FaucetError::Config(
39                    "ConfluentAvro selected but no SchemaRegistryClient available".into(),
40                )
41            })?;
42            avro::decode(client, bytes).await
43        }
44        #[cfg(feature = "schema-registry")]
45        KafkaValueFormat::ConfluentProtobuf { .. } => {
46            let client = sr_client.ok_or_else(|| {
47                FaucetError::Config(
48                    "ConfluentProtobuf selected but no SchemaRegistryClient available".into(),
49                )
50            })?;
51            protobuf::decode(client, bytes).await
52        }
53        #[cfg(feature = "schema-registry")]
54        KafkaValueFormat::ConfluentJsonSchema { validate, .. } => {
55            let client = sr_client.ok_or_else(|| {
56                FaucetError::Config(
57                    "ConfluentJsonSchema selected but no SchemaRegistryClient available".into(),
58                )
59            })?;
60            json_schema::decode(client, bytes, *validate).await
61        }
62    }
63}
64
65#[cfg(test)]
66mod tests {
67    use super::*;
68
69    #[tokio::test]
70    async fn decode_json_object() {
71        let bytes = br#"{"id": 7, "name": "x"}"#;
72        let v = decode(
73            Some(bytes),
74            &KafkaValueFormat::Json,
75            #[cfg(feature = "schema-registry")]
76            None,
77        )
78        .await
79        .unwrap();
80        assert_eq!(v["id"], 7);
81    }
82
83    #[tokio::test]
84    async fn decode_raw_string_utf8() {
85        let bytes = "hello".as_bytes();
86        let v = decode(
87            Some(bytes),
88            &KafkaValueFormat::RawString,
89            #[cfg(feature = "schema-registry")]
90            None,
91        )
92        .await
93        .unwrap();
94        assert_eq!(v, Value::String("hello".into()));
95    }
96
97    #[tokio::test]
98    async fn decode_bytes_base64_encodes() {
99        let bytes = &[0xDE, 0xAD, 0xBE, 0xEF];
100        let v = decode(
101            Some(bytes),
102            &KafkaValueFormat::Bytes,
103            #[cfg(feature = "schema-registry")]
104            None,
105        )
106        .await
107        .unwrap();
108        assert_eq!(v, Value::String("3q2+7w==".into()));
109    }
110
111    #[tokio::test]
112    async fn decode_none_returns_null() {
113        let v = decode(
114            None,
115            &KafkaValueFormat::Json,
116            #[cfg(feature = "schema-registry")]
117            None,
118        )
119        .await
120        .unwrap();
121        assert_eq!(v, Value::Null);
122    }
123
124    #[tokio::test]
125    async fn decode_invalid_json_errors() {
126        let bytes = b"{not json";
127        let err = decode(
128            Some(bytes),
129            &KafkaValueFormat::Json,
130            #[cfg(feature = "schema-registry")]
131            None,
132        )
133        .await
134        .unwrap_err();
135        assert!(format!("{err}").contains("json"));
136    }
137
138    #[tokio::test]
139    async fn decode_raw_string_rejects_invalid_utf8() {
140        let bytes = &[0xFF, 0xFE];
141        let err = decode(
142            Some(bytes),
143            &KafkaValueFormat::RawString,
144            #[cfg(feature = "schema-registry")]
145            None,
146        )
147        .await
148        .unwrap_err();
149        let m = format!("{err}").to_lowercase();
150        assert!(m.contains("utf-8") || m.contains("utf"));
151    }
152
153    #[cfg(feature = "schema-registry")]
154    #[tokio::test]
155    async fn decode_confluent_avro_without_client_is_config_error() {
156        use faucet_common_kafka::SchemaRegistryConfig;
157        let format = KafkaValueFormat::ConfluentAvro {
158            schema_registry: SchemaRegistryConfig::new("http://localhost:8081"),
159        };
160        let err = decode(Some(b"\x00\x00\x00\x00\x01"), &format, None)
161            .await
162            .unwrap_err();
163        match err {
164            FaucetError::Config(msg) => assert!(
165                msg.contains("ConfluentAvro") && msg.contains("no SchemaRegistryClient"),
166                "unexpected message: {msg}"
167            ),
168            other => panic!("expected Config error, got {other:?}"),
169        }
170    }
171
172    #[cfg(feature = "schema-registry")]
173    #[tokio::test]
174    async fn decode_confluent_protobuf_without_client_is_config_error() {
175        use faucet_common_kafka::SchemaRegistryConfig;
176        let format = KafkaValueFormat::ConfluentProtobuf {
177            schema_registry: SchemaRegistryConfig::new("http://localhost:8081"),
178        };
179        let err = decode(Some(b"\x00\x00\x00\x00\x01"), &format, None)
180            .await
181            .unwrap_err();
182        match err {
183            FaucetError::Config(msg) => assert!(
184                msg.contains("ConfluentProtobuf") && msg.contains("no SchemaRegistryClient"),
185                "unexpected message: {msg}"
186            ),
187            other => panic!("expected Config error, got {other:?}"),
188        }
189    }
190
191    #[cfg(feature = "schema-registry")]
192    #[tokio::test]
193    async fn decode_confluent_json_schema_without_client_is_config_error() {
194        use faucet_common_kafka::SchemaRegistryConfig;
195        let format = KafkaValueFormat::ConfluentJsonSchema {
196            schema_registry: SchemaRegistryConfig::new("http://localhost:8081"),
197            validate: false,
198        };
199        let err = decode(Some(b"\x00\x00\x00\x00\x01"), &format, None)
200            .await
201            .unwrap_err();
202        match err {
203            FaucetError::Config(msg) => assert!(
204                msg.contains("ConfluentJsonSchema") && msg.contains("no SchemaRegistryClient"),
205                "unexpected message: {msg}"
206            ),
207            other => panic!("expected Config error, got {other:?}"),
208        }
209    }
210}
faucet_source_kafka/decode.rs

faucet_source_kafka/
decode.rs
