ruststream_rdkafka/schema_registry/client.rs
1//! What a registry answers: the seam, and the HTTP implementation this crate ships.
2
3use std::fmt;
4use std::sync::Arc;
5use std::time::Duration;
6
7use futures::future::BoxFuture;
8use serde::Deserialize;
9
10use super::{RegisteredSchema, SchemaType};
11use crate::error::KafkaError;
12
13/// The deadline one registry request gets unless the client names another.
14///
15/// Ten seconds sits above anything a healthy registry needs, a cold one and a slow hop included,
16/// and far below the stall an absent deadline allows on the delivery path.
17pub(super) const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
18
19/// The async half of a [`SchemaRegistry`](super::SchemaRegistry): the conversation with the
20/// registry itself.
21///
22/// This is the seam a service reaches for when the client that ships here is not the one it
23/// wants - a different HTTP stack, a published client crate, a binding to a non-Rust one, a fake
24/// in a test. Everything above it (the codecs, the byte-lane subjects, the prefetch, the
25/// transcoding middleware) is written against [`SchemaRegistry`](super::SchemaRegistry), which is a facade over this
26/// trait and a [`SchemaCache`](super::SchemaCache), so an implementation here reaches all of it.
27///
28/// Every method is async and returns a boxed future, because the trait is used as
29/// `Arc<dyn RegistryClient>`: a virtual call and one allocation per *registry round trip* is
30/// nothing beside the round trip, and erasing the type here is what keeps the codecs from
31/// growing a type parameter that would then have to be written out at every mount site.
32///
33/// # Examples
34///
35/// A client that answers every question with one fixed schema, behind the registry a service
36/// consumes through:
37///
38/// ```
39/// use std::sync::Arc;
40///
41/// use futures::future::BoxFuture;
42/// use ruststream_rdkafka::KafkaError;
43/// use ruststream_rdkafka::prelude::*;
44/// use ruststream_rdkafka::schema_registry::{
45/// RegisteredSchema, RegistryClient, SchemaRegistry, SchemaType,
46/// };
47/// # #[derive(serde::Deserialize)]
48/// # struct Order { id: u64 }
49/// # #[subscriber("orders")]
50/// # async fn place(order: &Order) -> HandlerOutcome {
51/// # tracing::info!(order.id, "accepted");
52/// # HandlerOutcome::ack()
53/// # }
54///
55/// const ORDER_SCHEMA: &str =
56/// r#"{"type": "record", "name": "Order", "fields": [{"name": "id", "type": "long"}]}"#;
57///
58/// struct Fixed(Arc<RegisteredSchema>);
59///
60/// impl RegistryClient for Fixed {
61/// fn schema_by_id(&self, _id: u32) -> BoxFuture<'_, Result<Arc<RegisteredSchema>, KafkaError>> {
62/// let schema = Arc::clone(&self.0);
63/// Box::pin(async move { Ok(schema) })
64/// }
65///
66/// fn latest(
67/// &self,
68/// _subject: &str,
69/// ) -> BoxFuture<'_, Result<Option<Arc<RegisteredSchema>>, KafkaError>> {
70/// let schema = Arc::clone(&self.0);
71/// Box::pin(async move { Ok(Some(schema)) })
72/// }
73///
74/// fn register(
75/// &self,
76/// _subject: &str,
77/// _schema_type: SchemaType,
78/// _definition: String,
79/// ) -> BoxFuture<'_, Result<u32, KafkaError>> {
80/// let id = self.0.id();
81/// Box::pin(async move { Ok(id) })
82/// }
83///
84/// fn lookup_id(
85/// &self,
86/// subject: &str,
87/// schema_type: SchemaType,
88/// definition: String,
89/// ) -> BoxFuture<'_, Result<u32, KafkaError>> {
90/// self.register(subject, schema_type, definition)
91/// }
92/// }
93///
94/// fn app() -> RustStream {
95/// let schema = RegisteredSchema::new(1, SchemaType::Avro, ORDER_SCHEMA);
96/// let registry = SchemaRegistry::with_client(Arc::new(Fixed(Arc::new(schema))));
97/// let broker = KafkaBroker::new(["localhost:9092"])
98/// .default_group("orders-svc")
99/// .schema_registry(registry);
100/// RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
101/// b.include(place);
102/// })
103/// }
104/// # let _ = app;
105/// ```
106#[diagnostic::on_unimplemented(
107 message = "`{Self}` is not a schema registry client",
108 note = "a client answers the four registry questions asynchronously: a schema by id, a \
109 subject's latest version, registering a schema, and looking one up without \
110 registering it"
111)]
112pub trait RegistryClient: Send + Sync + 'static {
113 /// The schema the registry holds under `id`.
114 ///
115 /// # Errors
116 ///
117 /// Returns [`KafkaError::SchemaRegistry`] when the registry is unreachable, rejects the
118 /// request, or does not know the id.
119 fn schema_by_id(&self, id: u32) -> BoxFuture<'_, Result<Arc<RegisteredSchema>, KafkaError>>;
120
121 /// `subject`'s latest version, or `None` when the registry has no such subject.
122 ///
123 /// A missing subject is `None` rather than an error because the framing middleware reads it
124 /// as "this topic is not registry-backed", which is an ordinary topology, not a failure.
125 ///
126 /// # Errors
127 ///
128 /// Returns [`KafkaError::SchemaRegistry`] when the registry is unreachable or rejects the
129 /// request.
130 fn latest(
131 &self,
132 subject: &str,
133 ) -> BoxFuture<'_, Result<Option<Arc<RegisteredSchema>>, KafkaError>>;
134
135 /// Registers `definition` under `subject` and returns the id, which is idempotent
136 /// registry-side: an identical schema keeps the id it already had.
137 ///
138 /// # Errors
139 ///
140 /// Returns [`KafkaError::SchemaRegistry`] when the registry is unreachable, rejects the
141 /// request, or refuses the schema as incompatible with the subject's history.
142 fn register(
143 &self,
144 subject: &str,
145 schema_type: SchemaType,
146 definition: String,
147 ) -> BoxFuture<'_, Result<u32, KafkaError>>;
148
149 /// The id `definition` already has under `subject`, registering nothing.
150 ///
151 /// # Errors
152 ///
153 /// Returns [`KafkaError::SchemaRegistry`] when the registry is unreachable or rejects the
154 /// request, and [`KafkaError::InvalidOptions`] when the subject does not hold this schema.
155 fn lookup_id(
156 &self,
157 subject: &str,
158 schema_type: SchemaType,
159 definition: String,
160 ) -> BoxFuture<'_, Result<u32, KafkaError>>;
161
162 /// Whether the registry would register `definition` under `subject`, judged by the
163 /// compatibility level the subject is configured with.
164 ///
165 /// The level decides which versions take part: the latest one for `BACKWARD`, `FORWARD` and
166 /// `FULL`, every version for their `_TRANSITIVE` forms. The default answers `None`,
167 /// meaning "this client cannot tell", and the caller then skips the check rather than reading
168 /// silence as either answer - so a client written before this method existed keeps working.
169 ///
170 /// # Errors
171 ///
172 /// Returns [`KafkaError::SchemaRegistry`] when the registry is unreachable or rejects the
173 /// request.
174 fn is_compatible(
175 &self,
176 subject: &str,
177 schema_type: SchemaType,
178 definition: String,
179 ) -> BoxFuture<'_, Result<Option<bool>, KafkaError>> {
180 let _ = (subject, schema_type, definition);
181 Box::pin(async { Ok(None) })
182 }
183}
184
185// No `Debug`: the variants hold credentials, and the registry's own `Debug` leaves them out.
186#[derive(Clone)]
187pub(super) enum Auth {
188 None,
189 Basic { user: String, password: String },
190 Bearer(String),
191}
192
193/// Everything the shipped HTTP client is configured with, kept whole so that changing one
194/// setting rebuilds the client without dropping the others.
195#[derive(Clone)]
196pub(super) struct HttpConfig {
197 pub(super) base_url: String,
198 pub(super) auth: Auth,
199 pub(super) timeout: Duration,
200}
201
202impl HttpConfig {
203 pub(super) fn new(base_url: String) -> Self {
204 Self {
205 base_url,
206 auth: Auth::None,
207 timeout: DEFAULT_REQUEST_TIMEOUT,
208 }
209 }
210}
211
212#[derive(Deserialize)]
213struct SchemaByIdResponse {
214 schema: String,
215 #[serde(rename = "schemaType")]
216 schema_type: Option<String>,
217}
218
219#[derive(Deserialize)]
220struct RegisterResponse {
221 id: u32,
222}
223
224#[derive(Deserialize)]
225struct CompatibilityResponse {
226 is_compatible: bool,
227 /// Present with `?verbose=true`: the registry's own account of what differs.
228 #[serde(default)]
229 messages: Vec<String>,
230}
231
232#[derive(Deserialize)]
233struct LatestVersionResponse {
234 id: u32,
235 schema: String,
236 #[serde(rename = "schemaType")]
237 schema_type: Option<String>,
238}
239
240/// The [`RegistryClient`] this crate ships: Confluent's HTTP API over `reqwest`, rustls only.
241pub struct HttpRegistryClient {
242 config: HttpConfig,
243 http: reqwest::Client,
244}
245
246impl fmt::Debug for HttpRegistryClient {
247 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
248 f.debug_struct("HttpRegistryClient")
249 .field("base_url", &self.config.base_url)
250 .field("request_timeout", &self.config.timeout)
251 .finish_non_exhaustive()
252 }
253}
254
255impl HttpRegistryClient {
256 pub(super) fn new(config: HttpConfig) -> Self {
257 Self {
258 config,
259 http: reqwest::Client::new(),
260 }
261 }
262
263 fn request(&self, method: reqwest::Method, path: &str) -> reqwest::RequestBuilder {
264 let url = format!("{}{path}", self.config.base_url);
265 let request = self.http.request(method, url).timeout(self.config.timeout);
266 match &self.config.auth {
267 Auth::None => request,
268 Auth::Basic { user, password } => request.basic_auth(user, Some(password)),
269 Auth::Bearer(token) => request.bearer_auth(token),
270 }
271 }
272
273 /// Turns one request failure into a crate error, telling a silent registry apart from a
274 /// failing one: only the expiry of this client's deadline becomes the timeout error.
275 fn failed(&self, path: &str, err: reqwest::Error) -> KafkaError {
276 if err.is_timeout() {
277 return KafkaError::SchemaRegistryTimeout {
278 request: path.to_owned(),
279 timeout: self.config.timeout,
280 };
281 }
282 KafkaError::schema_registry(err)
283 }
284
285 async fn get_json<T: serde::de::DeserializeOwned>(&self, path: &str) -> Result<T, KafkaError> {
286 let response = self
287 .request(reqwest::Method::GET, path)
288 .send()
289 .await
290 .map_err(|err| self.failed(path, err))?
291 .error_for_status()
292 .map_err(KafkaError::schema_registry)?;
293 response.json().await.map_err(|err| self.failed(path, err))
294 }
295
296 async fn post_schema(
297 &self,
298 path: String,
299 schema_type: SchemaType,
300 definition: String,
301 missing: Option<String>,
302 ) -> Result<u32, KafkaError> {
303 let body = serde_json::json!({
304 "schema": definition,
305 "schemaType": schema_type.as_api(),
306 });
307 let response = self
308 .request(reqwest::Method::POST, &path)
309 .json(&body)
310 .send()
311 .await
312 .map_err(|err| self.failed(&path, err))?;
313 if let Some(message) = missing
314 && response.status() == reqwest::StatusCode::NOT_FOUND
315 {
316 return Err(KafkaError::InvalidOptions(message));
317 }
318 let response = response
319 .error_for_status()
320 .map_err(KafkaError::schema_registry)?;
321 let registered: RegisterResponse = response
322 .json()
323 .await
324 .map_err(|err| self.failed(&path, err))?;
325 Ok(registered.id)
326 }
327}
328
329impl RegistryClient for HttpRegistryClient {
330 fn schema_by_id(&self, id: u32) -> BoxFuture<'_, Result<Arc<RegisteredSchema>, KafkaError>> {
331 Box::pin(async move {
332 let fetched: SchemaByIdResponse = self.get_json(&format!("/schemas/ids/{id}")).await?;
333 Ok(Arc::new(RegisteredSchema::new(
334 id,
335 SchemaType::from_api(fetched.schema_type.as_deref()),
336 fetched.schema,
337 )))
338 })
339 }
340
341 fn latest(
342 &self,
343 subject: &str,
344 ) -> BoxFuture<'_, Result<Option<Arc<RegisteredSchema>>, KafkaError>> {
345 let path = format!("/subjects/{subject}/versions/latest");
346 Box::pin(async move {
347 let response = self
348 .request(reqwest::Method::GET, &path)
349 .send()
350 .await
351 .map_err(|err| self.failed(&path, err))?;
352 if response.status() == reqwest::StatusCode::NOT_FOUND {
353 return Ok(None);
354 }
355 let response = response
356 .error_for_status()
357 .map_err(KafkaError::schema_registry)?;
358 let fetched: LatestVersionResponse = response
359 .json()
360 .await
361 .map_err(|err| self.failed(&path, err))?;
362 Ok(Some(Arc::new(RegisteredSchema::new(
363 fetched.id,
364 SchemaType::from_api(fetched.schema_type.as_deref()),
365 fetched.schema,
366 ))))
367 })
368 }
369
370 fn register(
371 &self,
372 subject: &str,
373 schema_type: SchemaType,
374 definition: String,
375 ) -> BoxFuture<'_, Result<u32, KafkaError>> {
376 let path = format!("/subjects/{subject}/versions");
377 Box::pin(self.post_schema(path, schema_type, definition, None))
378 }
379
380 fn lookup_id(
381 &self,
382 subject: &str,
383 schema_type: SchemaType,
384 definition: String,
385 ) -> BoxFuture<'_, Result<u32, KafkaError>> {
386 let path = format!("/subjects/{subject}");
387 let missing = format!(
388 "the registry holds no such schema under subject {subject:?}; register it there \
389 first (the typed shorthands do it in one call) or point the producer at the \
390 subject that carries this schema",
391 );
392 Box::pin(self.post_schema(path, schema_type, definition, Some(missing)))
393 }
394
395 fn is_compatible(
396 &self,
397 subject: &str,
398 schema_type: SchemaType,
399 definition: String,
400 ) -> BoxFuture<'_, Result<Option<bool>, KafkaError>> {
401 // `/versions` runs the check registration runs, against every version the subject's
402 // level names; `/versions/latest` compares with the latest alone even under a
403 // `_TRANSITIVE` level, and passes a schema the registry then refuses to register.
404 let path = format!("/compatibility/subjects/{subject}/versions?verbose=true");
405 let subject = subject.to_owned();
406 Box::pin(async move {
407 let body = serde_json::json!({
408 "schema": definition,
409 "schemaType": schema_type.as_api(),
410 });
411 let response = self
412 .request(reqwest::Method::POST, &path)
413 .json(&body)
414 .send()
415 .await
416 .map_err(|err| self.failed(&path, err))?;
417 // A subject with no versions cannot be checked against one; that is the missing
418 // subject case, which the caller has already settled by its own policy. The API
419 // documents a 404 for it; Confluent 7.7.1 answers "compatible" instead, which reads
420 // the same way here because nothing stands against the schema.
421 if response.status() == reqwest::StatusCode::NOT_FOUND {
422 return Ok(None);
423 }
424 let response = response
425 .error_for_status()
426 .map_err(KafkaError::schema_registry)?;
427 let verdict: CompatibilityResponse = response
428 .json()
429 .await
430 .map_err(|err| self.failed(&path, err))?;
431 if !verdict.is_compatible {
432 // The registry names the offending field; passing it through is the difference
433 // between a usable startup failure and "incompatible".
434 return Err(KafkaError::SchemaRegistry(
435 format!(
436 "the schema a codec publishes under subject {subject:?} is not \
437 compatible with the versions the subject's compatibility level \
438 checks it against: {}",
439 verdict.messages.join(" "),
440 )
441 .into(),
442 ));
443 }
444 Ok(Some(true))
445 })
446 }
447}