protobuf message schema validation

Cargo.toml

@@ -3,7 +3,8 @@ resolver = "2"
 members = [
     "tansu-kafka-model",
     "tansu-kafka-sans-io",
-    "tansu-proxy", "tansu-schema-registry",
+    "tansu-proxy",
+    "tansu-schema-registry",
     "tansu-server",
     "tansu-storage",
 ]
@@ -64,6 +65,8 @@ opentelemetry_sdk = "0.21.2"
 pretty_assertions = "1"
 prettyplease = "0.2.25"
 proc-macro2 = "1.0.92"
+protobuf-json-mapping = "3.7.1"
+protobuf-parse = "3.7.1"
 protobuf = { version = "3.7.1", features = ["with-bytes"] }
 quote = "1.0"
 rand = "0.8"

Dockerfile

@@ -24,7 +24,7 @@ ADD / /usr/src/
 RUN cargo build --package ${PACKAGE} --release
 
 RUN <<EOF
-mkdir /image /image/schema
+mkdir /image /image/schema /image/tmp
 
 # copy any dynamically linked libaries used
 for lib in $(ldd target/release/* 2>/dev/null|grep "=>"|awk '{print $3}'|sort|uniq); do
@@ -48,4 +48,5 @@ EOF
 
 FROM scratch
 COPY --from=builder /image /
+ENV TMP=/tmp
 ENTRYPOINT [ "/tansu-server" ]

tansu-schema-registry/Cargo.toml

@@ -5,19 +5,24 @@ version.workspace = true
 license.workspace = true
 
 [dependencies]
+anyhow.workspace = true
 bytes.workspace = true
 jsonschema.workspace = true
 object_store.workspace = true
+protobuf-parse.workspace = true
+protobuf.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 tansu-kafka-sans-io = { path = "../tansu-kafka-sans-io" }
+tempfile.workspace = true
 thiserror.workspace = true
 tokio.workspace = true
 tracing.workspace = true
 url.workspace = true
 
 [dev-dependencies]
 pretty_assertions.workspace = true
+protobuf-json-mapping.workspace = true
 tracing-subscriber.workspace = true
 
 

tansu-schema-registry/src/lib.rs

@@ -29,9 +29,11 @@ use url::Url;
 use tracing_subscriber::filter::ParseError;
 
 mod json;
+mod proto;
 
 #[derive(thiserror::Error, Debug)]
 pub enum Error {
+    Anyhow(#[from] anyhow::Error),
     Api(ErrorCode),
     Io(#[from] io::Error),
     KafkaSansIo(#[from] tansu_kafka_sans_io::Error),
@@ -41,6 +43,13 @@ pub enum Error {
     #[cfg(test)]
     ParseFilter(#[from] ParseError),
 
+    #[cfg(test)]
+    ProtobufJsonMapping(#[from] protobuf_json_mapping::ParseError),
+
+    Protobuf(#[from] protobuf::Error),
+
+    ProtobufFileDescriptorMissing(Bytes),
+
     SchemaValidation,
     SerdeJson(#[from] serde_json::Error),
     UnsupportedSchemaRegistryUrl(Url),
@@ -104,18 +113,34 @@ impl Registry {
     }
 
     pub async fn validate(&self, topic: &str, batch: &Batch) -> Result<()> {
-        json::Schema::try_from(Schema {
-            key: self
-                .get(&Path::from(format!("{topic}/key.json")))
+        let list_result = self.object_store.list_with_delimiter(None).await?;
+        debug!(common_prefixes = ?list_result.common_prefixes, objects = ?list_result.objects);
+
+        if list_result
+            .objects
+            .iter()
+            .any(|meta| meta.location == Path::from(format!("{topic}.proto")))
+        {
+            self.get(&Path::from(format!("{topic}.proto")))
                 .await
-                .ok(),
-
-            value: self
-                .get(&Path::from(format!("{topic}/value.json")))
-                .await
-                .ok(),
-        })
-        .and_then(|schema| schema.validate(batch))
+                .and_then(proto::Schema::try_from)
+                .and_then(|schema| schema.validate(batch))
+        } else if list_result.common_prefixes.contains(&Path::from(topic)) {
+            json::Schema::try_from(Schema {
+                key: self
+                    .get(&Path::from(format!("{topic}/key.json")))
+                    .await
+                    .ok(),
+
+                value: self
+                    .get(&Path::from(format!("{topic}/value.json")))
+                    .await
+                    .ok(),
+            })
+            .and_then(|schema| schema.validate(batch))
+        } else {
+            Ok(())
+        }
     }
 }
 
@@ -199,50 +224,62 @@ mod tests {
         ))
     }
 
-    #[tokio::test]
-    async fn valid() -> Result<()> {
+    static DEF_PROTO: &[u8] = br#"
+      syntax = 'proto3';
+
+      message Key {
+        int32 id = 1;
+      }
+
+      message Value {
+        string name = 1;
+        string email = 2;
+      }
+    "#;
+
+    async fn populate() -> Result<Registry> {
         let _guard = init_tracing()?;
 
-        let object_store = Arc::new(InMemory::new());
-        let location = Path::from("test/key.json");
-        let payload = serde_json::to_vec(&json!({"type": "number",
-            "multipleOf": 10}))
+        let object_store = InMemory::new();
+
+        let location = Path::from("abc/key.json");
+        let payload = serde_json::to_vec(&json!({
+            "type": "number",
+            "multipleOf": 10
+        }))
         .map(Bytes::from)
         .map(PutPayload::from)?;
 
         _ = object_store.put(&location, payload).await?;
 
-        let registry = Registry {
-            object_store: object_store.clone(),
-        };
+        let location = Path::from("def.proto");
+        let payload = PutPayload::from(Bytes::from_static(DEF_PROTO));
+        _ = object_store.put(&location, payload).await?;
+
+        Ok(Registry::new(object_store))
+    }
+
+    #[tokio::test]
+    async fn abc_valid() -> Result<()> {
+        let _guard = init_tracing()?;
+
+        let registry = populate().await?;
 
         let key = Bytes::from_static(b"5450");
 
         let batch = Batch::builder()
             .record(Record::builder().key(key.clone().into()))
             .build()?;
 
-        registry.validate("test", &batch).await?;
+        registry.validate("abc", &batch).await?;
 
         Ok(())
     }
 
     #[tokio::test]
-    async fn invalid() -> Result<()> {
+    async fn abc_invalid() -> Result<()> {
         let _guard = init_tracing()?;
-
-        let object_store = Arc::new(InMemory::new());
-        let location = Path::from("test/key.json");
-        let payload = serde_json::to_vec(&json!({"type": "number",
-            "multipleOf": 10}))
-        .map(Bytes::from)
-        .map(PutPayload::from)?;
-
-        _ = object_store.put(&location, payload).await?;
-
-        let registry = Registry {
-            object_store: object_store.clone(),
-        };
+        let registry = populate().await?;
 
         let key = Bytes::from_static(b"545");
 
@@ -252,7 +289,7 @@ mod tests {
 
         assert!(matches!(
             registry
-                .validate("test", &batch)
+                .validate("abc", &batch)
                 .await
                 .inspect_err(|err| error!(?err)),
             Err(Error::Api(ErrorCode::InvalidRecord))