faucet_sink_kafka/
encode.rs

1//! Encode Kafka message bytes from a serde_json::Value, 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#[derive(Default, Clone)]
15pub struct SchemaContext {
16    /// Subject name used for ConfluentAvro/Protobuf/JsonSchema. Usually `{topic}-value`.
17    pub subject: String,
18    /// Schema text to register (Avro JSON, JSON Schema JSON, Protobuf source).
19    /// Required for the Confluent variants on encode.
20    pub schema_text: Option<String>,
21}
22
23pub async fn encode(
24    value: &Value,
25    format: &KafkaValueFormat,
26    #[cfg(feature = "schema-registry")] sr_client: Option<&SchemaRegistryClient>,
27    #[cfg(feature = "schema-registry")] schema_ctx: &SchemaContext,
28) -> Result<Vec<u8>, FaucetError> {
29    match format {
30        KafkaValueFormat::Json => serde_json::to_vec(value)
31            .map_err(|e| FaucetError::Sink(format!("kafka json encode: {e}"))),
32        KafkaValueFormat::RawString => match value {
33            Value::String(s) => Ok(s.as_bytes().to_vec()),
34            other => Ok(other.to_string().into_bytes()),
35        },
36        KafkaValueFormat::Bytes => match value {
37            Value::String(s) => base64::engine::general_purpose::STANDARD
38                .decode(s)
39                .map_err(|e| FaucetError::Sink(format!("kafka bytes base64 decode: {e}"))),
40            _ => Err(FaucetError::Sink(
41                "kafka Bytes format requires the record value to be a base64-encoded string".into(),
42            )),
43        },
44        #[cfg(feature = "schema-registry")]
45        KafkaValueFormat::ConfluentAvro { .. } => {
46            let client = sr_client.ok_or_else(|| {
47                FaucetError::Config("ConfluentAvro selected but no SchemaRegistryClient".into())
48            })?;
49            let schema_text = schema_ctx
50                .schema_text
51                .as_deref()
52                .ok_or_else(|| FaucetError::Config("ConfluentAvro requires schema_text".into()))?;
53            avro::encode(client, &schema_ctx.subject, schema_text, value).await
54        }
55        #[cfg(feature = "schema-registry")]
56        KafkaValueFormat::ConfluentProtobuf { .. } => {
57            let client = sr_client.ok_or_else(|| {
58                FaucetError::Config("ConfluentProtobuf selected but no SchemaRegistryClient".into())
59            })?;
60            let schema_text = schema_ctx.schema_text.as_deref().ok_or_else(|| {
61                FaucetError::Config("ConfluentProtobuf requires schema_text".into())
62            })?;
63            protobuf::encode(client, &schema_ctx.subject, schema_text, value).await
64        }
65        #[cfg(feature = "schema-registry")]
66        KafkaValueFormat::ConfluentJsonSchema { .. } => {
67            let client = sr_client.ok_or_else(|| {
68                FaucetError::Config(
69                    "ConfluentJsonSchema selected but no SchemaRegistryClient".into(),
70                )
71            })?;
72            let schema_text = schema_ctx.schema_text.as_deref().ok_or_else(|| {
73                FaucetError::Config("ConfluentJsonSchema requires schema_text".into())
74            })?;
75            json_schema::encode(client, &schema_ctx.subject, schema_text, value).await
76        }
77    }
78}
79
80#[cfg(test)]
81mod tests {
82    use super::*;
83    use serde_json::json;
84
85    #[tokio::test]
86    async fn encode_json_object() {
87        let bytes = encode(
88            &json!({"a": 1}),
89            &KafkaValueFormat::Json,
90            #[cfg(feature = "schema-registry")]
91            None,
92            #[cfg(feature = "schema-registry")]
93            &SchemaContext::default(),
94        )
95        .await
96        .unwrap();
97        assert_eq!(bytes, br#"{"a":1}"#);
98    }
99
100    #[tokio::test]
101    async fn encode_raw_string_passes_through() {
102        let bytes = encode(
103            &Value::String("hello".into()),
104            &KafkaValueFormat::RawString,
105            #[cfg(feature = "schema-registry")]
106            None,
107            #[cfg(feature = "schema-registry")]
108            &SchemaContext::default(),
109        )
110        .await
111        .unwrap();
112        assert_eq!(bytes, b"hello");
113    }
114
115    #[tokio::test]
116    async fn encode_raw_string_stringifies_non_string() {
117        let bytes = encode(
118            &json!(42),
119            &KafkaValueFormat::RawString,
120            #[cfg(feature = "schema-registry")]
121            None,
122            #[cfg(feature = "schema-registry")]
123            &SchemaContext::default(),
124        )
125        .await
126        .unwrap();
127        assert_eq!(bytes, b"42");
128    }
129
130    #[tokio::test]
131    async fn encode_bytes_decodes_base64() {
132        let bytes = encode(
133            &Value::String("3q2+7w==".into()),
134            &KafkaValueFormat::Bytes,
135            #[cfg(feature = "schema-registry")]
136            None,
137            #[cfg(feature = "schema-registry")]
138            &SchemaContext::default(),
139        )
140        .await
141        .unwrap();
142        assert_eq!(bytes, vec![0xDE, 0xAD, 0xBE, 0xEF]);
143    }
144
145    #[tokio::test]
146    async fn encode_bytes_errors_on_non_string() {
147        let err = encode(
148            &json!({"x": 1}),
149            &KafkaValueFormat::Bytes,
150            #[cfg(feature = "schema-registry")]
151            None,
152            #[cfg(feature = "schema-registry")]
153            &SchemaContext::default(),
154        )
155        .await
156        .unwrap_err();
157        assert!(format!("{err}").contains("base64"));
158    }
159
160    #[cfg(feature = "schema-registry")]
161    mod schema_registry {
162        use super::*;
163        use faucet_common_kafka::SchemaRegistryConfig;
164        use faucet_common_kafka::schema_registry::client::SchemaRegistryClient;
165
166        fn avro_format() -> KafkaValueFormat {
167            KafkaValueFormat::ConfluentAvro {
168                schema_registry: SchemaRegistryConfig::new("http://localhost:8081"),
169            }
170        }
171
172        fn protobuf_format() -> KafkaValueFormat {
173            KafkaValueFormat::ConfluentProtobuf {
174                schema_registry: SchemaRegistryConfig::new("http://localhost:8081"),
175            }
176        }
177
178        fn json_schema_format() -> KafkaValueFormat {
179            KafkaValueFormat::ConfluentJsonSchema {
180                schema_registry: SchemaRegistryConfig::new("http://localhost:8081"),
181                validate: false,
182            }
183        }
184
185        /// A `SchemaRegistryClient::new` only builds the HTTP client and
186        /// validates the URL — no network I/O — so it is safe to construct
187        /// offline for the "client present but schema_text missing" branch.
188        fn offline_client() -> SchemaRegistryClient {
189            SchemaRegistryClient::new(&SchemaRegistryConfig::new("http://localhost:8081"))
190                .expect("offline client builds")
191        }
192
193        #[tokio::test]
194        async fn encode_confluent_avro_without_client_is_config_error() {
195            let err = encode(
196                &json!({"a": 1}),
197                &avro_format(),
198                None,
199                &SchemaContext::default(),
200            )
201            .await
202            .unwrap_err();
203            match err {
204                FaucetError::Config(msg) => assert!(
205                    msg.contains("ConfluentAvro") && msg.contains("no SchemaRegistryClient"),
206                    "unexpected message: {msg}"
207                ),
208                other => panic!("expected Config error, got {other:?}"),
209            }
210        }
211
212        #[tokio::test]
213        async fn encode_confluent_protobuf_without_client_is_config_error() {
214            let err = encode(
215                &json!({"a": 1}),
216                &protobuf_format(),
217                None,
218                &SchemaContext::default(),
219            )
220            .await
221            .unwrap_err();
222            match err {
223                FaucetError::Config(msg) => assert!(
224                    msg.contains("ConfluentProtobuf") && msg.contains("no SchemaRegistryClient"),
225                    "unexpected message: {msg}"
226                ),
227                other => panic!("expected Config error, got {other:?}"),
228            }
229        }
230
231        #[tokio::test]
232        async fn encode_confluent_json_schema_without_client_is_config_error() {
233            let err = encode(
234                &json!({"a": 1}),
235                &json_schema_format(),
236                None,
237                &SchemaContext::default(),
238            )
239            .await
240            .unwrap_err();
241            match err {
242                FaucetError::Config(msg) => assert!(
243                    msg.contains("ConfluentJsonSchema") && msg.contains("no SchemaRegistryClient"),
244                    "unexpected message: {msg}"
245                ),
246                other => panic!("expected Config error, got {other:?}"),
247            }
248        }
249
250        #[tokio::test]
251        async fn encode_confluent_avro_without_schema_text_is_config_error() {
252            let client = offline_client();
253            // schema_text defaults to None — the schema_text guard fires before
254            // any registry network call.
255            let err = encode(
256                &json!({"a": 1}),
257                &avro_format(),
258                Some(&client),
259                &SchemaContext::default(),
260            )
261            .await
262            .unwrap_err();
263            match err {
264                FaucetError::Config(msg) => assert!(
265                    msg.contains("ConfluentAvro") && msg.contains("schema_text"),
266                    "unexpected message: {msg}"
267                ),
268                other => panic!("expected Config error, got {other:?}"),
269            }
270        }
271
272        #[tokio::test]
273        async fn encode_confluent_protobuf_without_schema_text_is_config_error() {
274            let client = offline_client();
275            let err = encode(
276                &json!({"a": 1}),
277                &protobuf_format(),
278                Some(&client),
279                &SchemaContext::default(),
280            )
281            .await
282            .unwrap_err();
283            match err {
284                FaucetError::Config(msg) => assert!(
285                    msg.contains("ConfluentProtobuf") && msg.contains("schema_text"),
286                    "unexpected message: {msg}"
287                ),
288                other => panic!("expected Config error, got {other:?}"),
289            }
290        }
291
292        #[tokio::test]
293        async fn encode_confluent_json_schema_without_schema_text_is_config_error() {
294            let client = offline_client();
295            let err = encode(
296                &json!({"a": 1}),
297                &json_schema_format(),
298                Some(&client),
299                &SchemaContext::default(),
300            )
301            .await
302            .unwrap_err();
303            match err {
304                FaucetError::Config(msg) => assert!(
305                    msg.contains("ConfluentJsonSchema") && msg.contains("schema_text"),
306                    "unexpected message: {msg}"
307                ),
308                other => panic!("expected Config error, got {other:?}"),
309            }
310        }
311    }
312}
faucet_sink_kafka/encode.rs

faucet_sink_kafka/
encode.rs
