Repository: gklijs/schema_registry_converter (language: rust)

Issue:
Error using encode_with_struct with fixed value in schema
**Describe the bug**
When using a fixed value in the schema it will create a value of Value::Array, which will give an a error. Ideally this should be fixed in the avro-rs library but we can work around it,

**To Reproduce**
Steps to reproduce the behavior:
Having a schema like
```rust
r#"{"type":"record","name":"ConfirmAccountCreation","namespace":"nl.openweb.data","fields":[{"name":"id","type":{"type":"fixed","name":"Uuid","size":16}},{"name":"a_type","type":{"type":"enum","name":"Atype","symbols":["AUTO","MANUAL"]}}]}"#;
```



Discussion / hints:

Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index 4a25ca4..98dfb62 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -39,7 +39,7 @@ version = ">=0.8.0, <2.0"
 version = ">= 2.1.1"
 
 [dev-dependencies]
-mockito = "0.22.0"
+mockito = "0.23.3"
 rdkafka = { version = "0.23", features = ["cmake-build"] }
 futures = "^0.3.1"
 rand = "0.6.5"
diff --git a/src/lib.rs b/src/lib.rs
index b65a0fc..55c7263 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -25,7 +25,9 @@
 #[macro_use] extern crate failure;
 
 pub mod schema_registry;
+mod value_fixed_fix;
 
+use crate::value_fixed_fix::FixedFixer;
 use avro_rs::schema::Name;
 use avro_rs::to_value;
 use avro_rs::types::{Record, ToAvro, Value};
@@ -157,10 +159,9 @@ impl Decoder {
         match schema {
             Ok(v) => match from_avro_datum(&v, &mut reader, None) {
                 Ok(v) => Ok(v),
-                Err(e) => Err(SRCError::new(
+                Err(e) => Err(SRCError::non_retryable_from_err(
+                    e,
                     "Could not transform bytes using schema",
-                    Some(&e.to_string()),
-                    false,
                 )),
             },
             Err(e) => Err(e.clone()),
@@ -410,16 +411,15 @@ fn to_payload<T: ToAvro>(schema: &Schema, id: u32, record: T) -> Result<Vec<u8>,
     BigEndian::write_u32(&mut buf, id);
     payload.extend_from_slice(&buf);
     match to_avro_datum(schema, record) {
-        Ok(v) => payload.extend_from_slice(v.as_slice()),
-        Err(e) => {
-            return Err(SRCError::new(
-                "Could not get avro bytes",
-                Some(&e.to_string()),
-                false,
-            ));
+        Ok(v) => {
+            payload.extend_from_slice(v.as_slice());
+            Ok(payload)
         }
+        Err(e) => Err(SRCError::non_retryable_from_err(
+            e,
+            "Could not get avro bytes",
+        )),
     }
-    Ok(payload)
 }
 
 /// Using the schema with a vector of values the values will be correctly deserialized according to
@@ -436,7 +436,7 @@ fn to_bytes(
                 "Could not create record from schema",
                 None,
                 false,
-            ));
+            ))
         }
     };
     for value in values {
@@ -448,15 +448,14 @@ fn to_bytes(
 /// Using the schema with an item implementing serialize the item will be correctly deserialized
 /// according to the avro specification.
 fn item_to_bytes(schema: &Schema, id: u32, item: impl Serialize) -> Result<Vec<u8>, SRCError> {
-    let record = match to_value(item) {
-        Ok(v) => v,
-        Err(e) => {
-            return Err(SRCError::new(
-                "Could not get avro record from struct",
-                Some(&e.to_string()),
-                false,
-            ));
-        }
+    let record = match to_value(item)
+        .map_err(|e| SRCError::non_retryable_from_err(e, "Could not transform to avro_rs value"))
+        .and_then(|r| r.fix_fixed(schema))
+        .map(|r| r.resolve(schema))
+    {
+        Ok(Ok(v)) => v,
+        Ok(Err(e)) => return Err(SRCError::non_retryable_from_err(e, "Failed to resolve")),
+        Err(e) => return Err(e),
     };
     to_payload(schema, id, record)
 }
@@ -498,17 +497,11 @@ fn to_bytes_no_transfer_wrong() {
 
 #[cfg(test)]
 mod tests {
-    use crate::schema_registry::{SRCError, SubjectNameStrategy, SuppliedSchema};
-    use crate::Decoder;
-    use crate::Encoder;
+    use super::*;
+    use crate::schema_registry::SuppliedSchema;
     use avro_rs::from_value;
-    use avro_rs::schema::Name;
-    use avro_rs::types::Value;
-    use avro_rs::Schema;
     use mockito::{mock, server_address};
-    use serde::Deserialize;
-    use serde::Serialize;
-    use std::collections::HashMap;
+    use serde::{Deserialize, Serialize};
 
     #[derive(Debug, Deserialize, Serialize)]
     struct Heartbeat {
@@ -520,6 +513,38 @@ mod tests {
         map: HashMap<u32, u32>,
     }
 
+    #[derive(Debug, PartialEq, Eq, Hash, PartialOrd, Ord, Clone, Deserialize, Serialize)]
+    pub enum Atype {
+        #[serde(rename = "AUTO")]
+        Auto,
+        #[serde(rename = "MANUAL")]
+        Manual,
+    }
+
+    impl Default for Atype {
+        fn default() -> Self {
+            Atype::Auto
+        }
+    }
+
+    pub type Uuid = [u8; 16];
+
+    #[serde(default)]
+    #[derive(Debug, PartialEq, Clone, Deserialize, Serialize)]
+    pub struct ConfirmAccountCreation {
+        pub id: Uuid,
+        pub a_type: Atype,
+    }
+
+    impl Default for ConfirmAccountCreation {
+        fn default() -> ConfirmAccountCreation {
+            ConfirmAccountCreation {
+                id: Uuid::default(),
+                a_type: Atype::Auto,
+            }
+        }
+    }
+
     #[test]
     fn display_decoder() {
         let decoder = Decoder::new(server_address().to_string());
@@ -762,6 +787,20 @@ mod tests {
         )
     }
 
+    #[test]
+    fn test_decoder_fixed_with_enum() {
+        let _m = mock("GET", "/schemas/ids/6")
+            .with_status(200)
+            .with_header("content-type", "application/vnd.schemaregistry.v1+json")
+            .with_body(r#"{"schema":"{\"type\":\"record\",\"name\":\"ConfirmAccountCreation\",\"namespace\":\"nl.openweb.data\",\"fields\":[{\"name\":\"id\",\"type\":{\"type\":\"fixed\",\"name\":\"Uuid\",\"size\":16}},{\"name\":\"a_type\",\"type\":{\"type\":\"enum\",\"name\":\"Atype\",\"symbols\":[\"AUTO\",\"MANUAL\"]}}]}"}"#)
+            .create();
+
+        let mut decoder = Decoder::new(format!("http://{}", server_address()));
+        let cac = decoder.decode(Some(&[0, 0, 0, 0, 6, 204, 240, 237, 74, 227, 188, 75, 46, 183, 163, 122, 214, 178, 72, 118, 162, 2]));
+
+        assert_eq!(cac, Ok(Value::Record(vec!(("id".to_string(), Value::Fixed(16, vec!(204, 240, 237, 74, 227, 188, 75, 46, 183, 163, 122, 214, 178, 72, 118, 162))), ("a_type".to_string(), Value::Enum(1, "MANUAL".to_string()))))));
+    }
+
     #[test]
     fn test_decoder_cache() {
         let mut decoder = Decoder::new(format!("http://{}", server_address()));
@@ -1122,8 +1161,8 @@ mod tests {
         assert_eq!(
             result,
             Err(SRCError::new(
-                "Could not get avro bytes",
-                Some("Validation error: value does not match schema"),
+                "error fixing record",
+                Some("missing field name in record"),
                 false,
             ))
         )
@@ -1138,10 +1177,26 @@ mod tests {
         assert_eq!(
             result,
             Err(SRCError::new(
-                "Could not get avro record from struct",
+                "Could not transform to avro_rs value",
                 Some("map key is not a string"),
                 false,
             ))
         )
     }
+
+    #[test]
+    fn item_to_bytes_proper_bytes_record_with_fixed() {
+        let schema = Schema::parse_str(r#"{"type":"record","name":"ConfirmAccountCreation","namespace":"nl.openweb.data","fields":[{"name":"id","type":{"type":"fixed","name":"Uuid","size":16}},{"name":"a_type","type":{"type":"enum","name":"Atype","symbols":["AUTO","MANUAL"]}}]}"#).unwrap();
+        let item = ConfirmAccountCreation {
+            id: [
+                204, 240, 237, 74, 227, 188, 75, 46, 183, 163, 122, 214, 178, 72, 118, 162,
+            ],
+            a_type: Atype::Manual,
+        };
+        let result = crate::item_to_bytes(&schema, 6, item);
+        assert_eq!(
+            result,
+            Ok(vec!(0, 0, 0, 0, 6, 204, 240, 237, 74, 227, 188, 75, 46, 183, 163, 122, 214, 178, 72, 118, 162, 2))
+        )
+    }
 }
diff --git a/src/schema_registry.rs b/src/schema_registry.rs
index 787d013..549727b 100644
--- a/src/schema_registry.rs
+++ b/src/schema_registry.rs
@@ -5,6 +5,7 @@ use core::fmt;
 use curl::easy::{Easy2, Handler, List, WriteError};
 use serde_json::{Map as JsonMap, Value as JsonValue};
 use std::error::Error;
+use std::fmt::Display;
 use std::ops::Deref;
 use std::str;
 use url::Url;
@@ -198,10 +199,9 @@ fn schema_from_url(url: &str, id: Option<u32>) -> Result<(Schema, u32), SRCError
     let schema = match Schema::parse_str(raw_schema) {
         Ok(v) => v,
         Err(e) => {
-            return Err(SRCError::new(
+            return Err(SRCError::non_retryable_from_err(
+                e,
                 "Could not parse schema",
-                Some(&e.to_string()),
-                false,
             ))
         }
     };
@@ -283,7 +283,6 @@ fn to_json(mut easy: Easy2<Collector>) -> Result<JsonValue, SRCError> {
                 false,
             ))
         }
-
         Err(e) => {
             return Err(SRCError::new(
                 format!("Encountered error getting http response: {}", e).as_str(),
@@ -299,10 +298,9 @@ fn to_json(mut easy: Easy2<Collector>) -> Result<JsonValue, SRCError> {
     let body = match str::from_utf8(data.as_ref()) {
         Ok(v) => v,
         Err(e) => {
-            return Err(SRCError::new(
+            return Err(SRCError::non_retryable_from_err(
+                e,
                 "Invalid UTF-8 sequence",
-                Some(e.description()),
-                false,
             ))
         }
     };
@@ -386,6 +384,9 @@ impl SRCError {
             cached: false,
         }
     }
+    pub fn non_retryable_from_err<T: Display>(cause: T, error: &str) -> SRCError {
+        SRCError::new(error, Some(&cause.to_string()), false)
+    }
     /// Should be called before putting the error in the cache
     pub fn into_cache(self) -> SRCError {
         SRCError {
@@ -488,7 +489,7 @@ fn display_erro_no_cause() {
 }
 
 #[test]
-fn display_erro_with_cause() {
+fn display_error_with_cause() {
     let err = SRCError::new(
         "Could not get id from response",
         Some("error in response"),
diff --git a/src/value_fixed_fix.rs b/src/value_fixed_fix.rs
new file mode 100644
index 0000000..a8d064b
--- /dev/null
+++ b/src/value_fixed_fix.rs
@@ -0,0 +1,178 @@
+//! This module fixes the incorrect handling of fixed values in the schema
+
+use crate::schema_registry::SRCError;
+use avro_rs::schema::{RecordField, UnionSchema};
+use avro_rs::types::{ToAvro, Value};
+use avro_rs::Schema;
+use std::collections::HashMap;
+
+pub trait FixedFixer {
+    fn fix_fixed(self, schema: &Schema) -> Result<Self, SRCError>
+    where
+        Self: std::marker::Sized;
+    fn fix_fixed_fixed(self, size: usize) -> Result<Self, SRCError>
+    where
+        Self: std::marker::Sized;
+    fn fix_fixed_union(self, schema: &UnionSchema) -> Result<Self, SRCError>
+    where
+        Self: std::marker::Sized;
+    fn fix_fixed_array(self, schema: &Schema) -> Result<Self, SRCError>
+    where
+        Self: std::marker::Sized;
+    fn fix_fixed_map(self, schema: &Schema) -> Result<Self, SRCError>
+    where
+        Self: std::marker::Sized;
+    fn fix_fixed_record(self, schema: &[RecordField]) -> Result<Self, SRCError>
+    where
+        Self: std::marker::Sized;
+}
+
+fn vec_to_fixed(vec: Vec<Value>) -> Result<Value, SRCError> {
+    let mut errors: Vec<Value> = vec![];
+    let mut hits: Vec<u8> = vec![];
+    vec.iter().for_each(|part| match part {
+        Value::Int(v) => {
+            if *v >= 0 && *v <= 255 {
+                hits.push((v & 0xff) as u8)
+            } else {
+                errors.push(Value::Int(*v))
+            }
+        }
+        other => errors.push(other.clone()),
+    });
+    if errors.is_empty() {
+        Ok(Value::Fixed(vec.len(), hits))
+    } else {
+        Err(SRCError::non_retryable_from_err(
+            format!(
+                "Found non-int values and/or ints that can't be converted to bytes: {:#?}",
+                errors
+            ),
+            "could not fix fixed",
+        ))
+    }
+}
+
+impl FixedFixer for Value {
+    /// Will go  through the schema and value, and convert arrays of the correct size to fixed
+    /// equivalents in order to fix serialization for structs.
+    fn fix_fixed(self, schema: &Schema) -> Result<Self, SRCError> {
+        match *schema {
+            Schema::Fixed { size, .. } => self.fix_fixed_fixed(size),
+            Schema::Union(ref inner) => self.fix_fixed_union(inner),
+            Schema::Array(ref inner) => self.fix_fixed_array(inner),
+            Schema::Map(ref inner) => self.fix_fixed_map(inner),
+            Schema::Record { ref fields, .. } => self.fix_fixed_record(fields),
+            _ => Ok(self),
+        }
+    }
+
+    fn fix_fixed_fixed(self, size: usize) -> Result<Self, SRCError> {
+        match self {
+            Value::Fixed(n, bytes) => {
+                if n == size {
+                    Ok(Value::Fixed(n, bytes))
+                } else {
+                    Err(SRCError::non_retryable_from_err(
+                        format!("Fixed size mismatch, {} expected, got {}", size, n),
+                        "could not fix fixed",
+                    ))
+                }
+            }
+            Value::Array(vec) => {
+                if vec.len() == size {
+                    vec_to_fixed(vec)
+                } else {
+                    Err(SRCError::non_retryable_from_err(
+                        format!("Array size mismatch, {} expected, got {}", size, vec.len()),
+                        "could not fix fixed",
+                    ))
+                }
+            }
+            other => Err(SRCError::non_retryable_from_err(
+                format!("Fixed or vector expected, got {:?}", other),
+                "could not fix fixed",
+            )),
+        }
+    }
+
+    fn fix_fixed_union(self, schema: &UnionSchema) -> Result<Self, SRCError> {
+        let v = match self {
+            // Both are unions case.
+            Value::Union(v) => *v,
+            // Reader is a union, but writer is not.
+            v => v,
+        };
+        // Find the first match in the reader schema.
+        let (_, inner) = schema.find_schema(&v).ok_or_else(|| {
+            SRCError::non_retryable_from_err(
+                "Could not find matching type in union",
+                "could not fix fixed",
+            )
+        })?;
+        v.fix_fixed(inner)
+    }
+
+    fn fix_fixed_array(self, schema: &Schema) -> Result<Self, SRCError> {
+        match self {
+            Value::Array(items) => Ok(Value::Array(
+                items
+                    .into_iter()
+                    .map(|item| item.fix_fixed(schema))
+                    .collect::<Result<Vec<_>, _>>()?,
+            )),
+            other => Err(SRCError::non_retryable_from_err(
+                format!("Array({:?}) expected, got {:?}", schema, other),
+                "could not fix fixed",
+            )),
+        }
+    }
+
+    fn fix_fixed_map(self, schema: &Schema) -> Result<Self, SRCError> {
+        match self {
+            Value::Map(items) => Ok(Value::Map(
+                items
+                    .into_iter()
+                    .map(|(key, value)| value.fix_fixed(schema).map(|value| (key, value)))
+                    .collect::<Result<HashMap<_, _>, _>>()?,
+            )),
+            other => Err(SRCError::non_retryable_from_err(
+                format!("Map({:?}) expected, got {:?}", schema, other),
+                "could not fix fixed",
+            )),
+        }
+    }
+
+    fn fix_fixed_record(self, fields: &[RecordField]) -> Result<Self, SRCError> {
+        let mut items = match self {
+            Value::Map(items) => Ok(items),
+            Value::Record(fields) => Ok(fields.into_iter().collect::<HashMap<_, _>>()),
+            other => Err(SRCError::non_retryable_from_err(
+                format!("Record({:?}) expected, got {:?}", fields, other),
+                "Error creating record",
+            )),
+        }?;
+
+        let new_fields = fields
+            .iter()
+            .map(|field| {
+                let value = match items.remove(&field.name) {
+                    Some(value) => value,
+                    None => match field.default {
+                        Some(ref value) => value.clone().avro(),
+                        _ => {
+                            return Err(SRCError::non_retryable_from_err(
+                                format!("missing field {} in record", field.name),
+                                "error fixing record",
+                            ))
+                        }
+                    },
+                };
+                value
+                    .fix_fixed(&field.schema)
+                    .map(|value| (field.name.clone(), value))
+            })
+            .collect::<Result<Vec<_>, _>>()?;
+        Ok(Value::Record(new_fields))
+    }
+}


Test patch: