Fix JSON Schema validation

rayokota · rayokota · commit 34e8cccd05be · 2025-01-28T17:28:11.000-08:00
diff --git a/src/Confluent.SchemaRegistry.Serdes.Json/JsonDeserializer.cs b/src/Confluent.SchemaRegistry.Serdes.Json/JsonDeserializer.cs
@@ -61,6 +61,8 @@ public class JsonDeserializer<T> : AsyncDeserializer<T, JsonSchema> where T : cl
 
         private JsonSchemaValidator validator = new JsonSchemaValidator();
 
+        private Schema schemaJson = null;
+
         private JsonSchema schema = null;
 
         private bool validate = true;
@@ -142,6 +144,7 @@ public JsonDeserializer(ISchemaRegistryClient schemaRegistryClient, Schema schem
             JsonSchemaResolver utils = new JsonSchemaResolver(
                 schemaRegistryClient, schema, this.jsonSchemaGeneratorSettings);
             JsonSchema jsonSchema = utils.GetResolvedSchema().Result;
+            this.schemaJson = schema;
             this.schema = jsonSchema;
         }
 
@@ -184,8 +187,10 @@ public override async Task<T> DeserializeAsync(ReadOnlyMemory<byte> data, bool i
 
             try
             {
-                Schema writerSchema = null;
-                JsonSchema writerSchemaJson = null;
+                Schema writerSchemaJson = null;
+                JsonSchema writerSchema = null;
+                Schema readerSchemaJson;
+                JsonSchema readerSchema;
                 T value;
                 IList<Migration> migrations = new List<Migration>();
                 using (var stream = new MemoryStream(array))
@@ -201,22 +206,34 @@ public override async Task<T> DeserializeAsync(ReadOnlyMemory<byte> data, bool i
 
                     if (schemaRegistryClient != null)
                     {
-                        (writerSchema, writerSchemaJson) = await GetSchema(subject, writerId);
+                        (writerSchemaJson, writerSchema) = await GetSchema(subject, writerId);
                         if (subject == null)
                         {
-                            subject = GetSubjectName(topic, isKey, writerSchemaJson.Title);
+                            subject = GetSubjectName(topic, isKey, writerSchema.Title);
                             if (subject != null)
                             {
                                 latestSchema = await GetReaderSchema(subject)
                                     .ConfigureAwait(continueOnCapturedContext: false);
                             }
                         }
                     }
-                    
+
                     if (latestSchema != null)
                     {
-                        migrations = await GetMigrations(subject, writerSchema, latestSchema)
+                        migrations = await GetMigrations(subject, writerSchemaJson, latestSchema)
                             .ConfigureAwait(continueOnCapturedContext: false);
+                        readerSchemaJson = latestSchema;
+                        readerSchema = await GetParsedSchema(latestSchema);
+                    }
+                    else if (schema != null)
+                    {
+                        readerSchemaJson = schemaJson;
+                        readerSchema = schema;
+                    }
+                    else
+                    {
+                        readerSchemaJson = writerSchemaJson;
+                        readerSchema = writerSchema;
                     }
 
                     if (migrations.Count > 0)
@@ -231,9 +248,9 @@ public override async Task<T> DeserializeAsync(ReadOnlyMemory<byte> data, bool i
                                 .ContinueWith(t => (JToken)t.Result)
                                 .ConfigureAwait(continueOnCapturedContext: false);
 
-                            if (schema != null && validate)
+                            if (readerSchema != null && validate)
                             {
-                                var validationResult = validator.Validate(json, schema);
+                                var validationResult = validator.Validate(json, readerSchema);
                                 if (validationResult.Count > 0)
                                 {
                                     throw new InvalidDataException("Schema validation failed for properties: [" +
@@ -250,10 +267,10 @@ public override async Task<T> DeserializeAsync(ReadOnlyMemory<byte> data, bool i
                         using (var jsonReader = new StreamReader(jsonStream, Encoding.UTF8))
                         {
                             string serializedString = jsonReader.ReadToEnd();
-                    
-                            if (schema != null && validate)
+
+                            if (readerSchema != null && validate)
                             {
-                                var validationResult = validator.Validate(serializedString, schema);
+                                var validationResult = validator.Validate(serializedString, readerSchema);
 
                                 if (validationResult.Count > 0)
                                 {
@@ -266,14 +283,10 @@ public override async Task<T> DeserializeAsync(ReadOnlyMemory<byte> data, bool i
                                 jsonSchemaGeneratorSettingsSerializerSettings);
                         }
                     }
-
-                    // A schema is not required to deserialize json messages.
-                    // TODO: add validation capability.
                 }
-                if (writerSchema != null)
+
+                if (readerSchema != null)
                 {
-                    Schema readerSchemaJson = latestSchema ?? writerSchema;
-                    JsonSchema readerSchema = latestSchema != null ? await GetParsedSchema(latestSchema) : writerSchemaJson;
                     FieldTransformer fieldTransformer = async (ctx, transform, message) =>
                     {
                         return await JsonUtils.Transform(ctx, readerSchema, "$", message, transform).ConfigureAwait(false);
@@ -283,8 +296,8 @@ public override async Task<T> DeserializeAsync(ReadOnlyMemory<byte> data, bool i
                             readerSchemaJson, value, fieldTransformer)
                         .ContinueWith(t => (T)t.Result)
                         .ConfigureAwait(continueOnCapturedContext: false);
-                } 
-                    
+                }
+
                 return value;
             }
             catch (AggregateException e)
diff --git a/src/Confluent.SchemaRegistry.Serdes.Json/JsonSerializer.cs b/src/Confluent.SchemaRegistry.Serdes.Json/JsonSerializer.cs
@@ -232,25 +232,30 @@ public override async Task<byte[]> SerializeAsync(T value, SerializationContext
                 {
                     serdeMutex.Release();
                 }
-                
+
+                JsonSchema parsedSchema;
                 if (latestSchema != null)
                 {
-                    var latestSchemaJson = await GetParsedSchema(latestSchema).ConfigureAwait(false);
+                    parsedSchema = await GetParsedSchema(latestSchema).ConfigureAwait(false);
                     FieldTransformer fieldTransformer = async (ctx, transform, message) =>
                     {
-                        return await JsonUtils.Transform(ctx, latestSchemaJson, "$", message, transform).ConfigureAwait(false);
+                        return await JsonUtils.Transform(ctx, parsedSchema, "$", message, transform).ConfigureAwait(false);
                     };
                     value = await ExecuteRules(context.Component == MessageComponentType.Key, subject,
                             context.Topic, context.Headers, RuleMode.Write, null,
                             latestSchema, value, fieldTransformer)
                         .ContinueWith(t => (T)t.Result)
                         .ConfigureAwait(continueOnCapturedContext: false);
                 }
+                else
+                {
+                    parsedSchema = schema;
+                }
 
                 var serializedString = Newtonsoft.Json.JsonConvert.SerializeObject(value, jsonSchemaGeneratorSettingsSerializerSettings);
                 if (validate)
                 {
-                    var validationResult = validator.Validate(serializedString, this.schema);
+                    var validationResult = validator.Validate(serializedString, parsedSchema);
                     if (validationResult.Count > 0)
                     {
                         throw new InvalidDataException("Schema validation failed for properties: [" + string.Join(", ", validationResult.Select(r => r.Path)) + "]");

Original file line number	Diff line number	Diff line change
`@@ -61,6 +61,8 @@ public class JsonDeserializer<T> : AsyncDeserializer<T, JsonSchema> where T : cl`
`61`	`61`
`62`	`62`	`private JsonSchemaValidator validator = new JsonSchemaValidator();`
`63`	`63`
	`64`	`+ private Schema schemaJson = null;`
	`65`	`+`
`64`	`66`	`private JsonSchema schema = null;`
`65`	`67`
`66`	`68`	`private bool validate = true;`
`@@ -142,6 +144,7 @@ public JsonDeserializer(ISchemaRegistryClient schemaRegistryClient, Schema schem`
`142`	`144`	`JsonSchemaResolver utils = new JsonSchemaResolver(`
`143`	`145`	`schemaRegistryClient, schema, this.jsonSchemaGeneratorSettings);`
`144`	`146`	`JsonSchema jsonSchema = utils.GetResolvedSchema().Result;`
	`147`	`+ this.schemaJson = schema;`
`145`	`148`	`this.schema = jsonSchema;`
`146`	`149`	`}`
`147`	`150`
`@@ -184,8 +187,10 @@ public override async Task<T> DeserializeAsync(ReadOnlyMemory<byte> data, bool i`
`184`	`187`
`185`	`188`	`try`
`186`	`189`	`{`
`187`		`- Schema writerSchema = null;`
`188`		`- JsonSchema writerSchemaJson = null;`
	`190`	`+ Schema writerSchemaJson = null;`
	`191`	`+ JsonSchema writerSchema = null;`
	`192`	`+ Schema readerSchemaJson;`
	`193`	`+ JsonSchema readerSchema;`
`189`	`194`	`T value;`
`190`	`195`	`IList<Migration> migrations = new List<Migration>();`
`191`	`196`	`using (var stream = new MemoryStream(array))`
`@@ -201,22 +206,34 @@ public override async Task<T> DeserializeAsync(ReadOnlyMemory<byte> data, bool i`
`201`	`206`
`202`	`207`	`if (schemaRegistryClient != null)`
`203`	`208`	`{`
`204`		`- (writerSchema, writerSchemaJson) = await GetSchema(subject, writerId);`
	`209`	`+ (writerSchemaJson, writerSchema) = await GetSchema(subject, writerId);`
`205`	`210`	`if (subject == null)`
`206`	`211`	`{`
`207`		`- subject = GetSubjectName(topic, isKey, writerSchemaJson.Title);`
	`212`	`+ subject = GetSubjectName(topic, isKey, writerSchema.Title);`
`208`	`213`	`if (subject != null)`
`209`	`214`	`{`
`210`	`215`	`latestSchema = await GetReaderSchema(subject)`
`211`	`216`	`.ConfigureAwait(continueOnCapturedContext: false);`
`212`	`217`	`}`
`213`	`218`	`}`
`214`	`219`	`}`
`215`		`-`
	`220`	`+`
`216`	`221`	`if (latestSchema != null)`
`217`	`222`	`{`
`218`		`- migrations = await GetMigrations(subject, writerSchema, latestSchema)`
	`223`	`+ migrations = await GetMigrations(subject, writerSchemaJson, latestSchema)`
`219`	`224`	`.ConfigureAwait(continueOnCapturedContext: false);`
	`225`	`+ readerSchemaJson = latestSchema;`
	`226`	`+ readerSchema = await GetParsedSchema(latestSchema);`
	`227`	`+ }`
	`228`	`+ else if (schema != null)`
	`229`	`+ {`
	`230`	`+ readerSchemaJson = schemaJson;`
	`231`	`+ readerSchema = schema;`
	`232`	`+ }`
	`233`	`+ else`
	`234`	`+ {`
	`235`	`+ readerSchemaJson = writerSchemaJson;`
	`236`	`+ readerSchema = writerSchema;`
`220`	`237`	`}`
`221`	`238`
`222`	`239`	`if (migrations.Count > 0)`
`@@ -231,9 +248,9 @@ public override async Task<T> DeserializeAsync(ReadOnlyMemory<byte> data, bool i`
`231`	`248`	`.ContinueWith(t => (JToken)t.Result)`
`232`	`249`	`.ConfigureAwait(continueOnCapturedContext: false);`
`233`	`250`
`234`		`- if (schema != null && validate)`
	`251`	`+ if (readerSchema != null && validate)`
`235`	`252`	`{`
`236`		`- var validationResult = validator.Validate(json, schema);`
	`253`	`+ var validationResult = validator.Validate(json, readerSchema);`
`237`	`254`	`if (validationResult.Count > 0)`
`238`	`255`	`{`
`239`	`256`	`throw new InvalidDataException("Schema validation failed for properties: [" +`
`@@ -250,10 +267,10 @@ public override async Task<T> DeserializeAsync(ReadOnlyMemory<byte> data, bool i`
`250`	`267`	`using (var jsonReader = new StreamReader(jsonStream, Encoding.UTF8))`
`251`	`268`	`{`
`252`	`269`	`string serializedString = jsonReader.ReadToEnd();`
`253`		`-`
`254`		`- if (schema != null && validate)`
	`270`	`+`
	`271`	`+ if (readerSchema != null && validate)`
`255`	`272`	`{`
`256`		`- var validationResult = validator.Validate(serializedString, schema);`
	`273`	`+ var validationResult = validator.Validate(serializedString, readerSchema);`
`257`	`274`
`258`	`275`	`if (validationResult.Count > 0)`
`259`	`276`	`{`
`@@ -266,14 +283,10 @@ public override async Task<T> DeserializeAsync(ReadOnlyMemory<byte> data, bool i`
`266`	`283`	`jsonSchemaGeneratorSettingsSerializerSettings);`
`267`	`284`	`}`
`268`	`285`	`}`
`269`		`-`
`270`		`- // A schema is not required to deserialize json messages.`
`271`		`- // TODO: add validation capability.`
`272`	`286`	`}`
`273`		`- if (writerSchema != null)`
	`287`	`+`
	`288`	`+ if (readerSchema != null)`
`274`	`289`	`{`
`275`		`- Schema readerSchemaJson = latestSchema ?? writerSchema;`
`276`		`- JsonSchema readerSchema = latestSchema != null ? await GetParsedSchema(latestSchema) : writerSchemaJson;`
`277`	`290`	`FieldTransformer fieldTransformer = async (ctx, transform, message) =>`
`278`	`291`	`{`
`279`	`292`	`return await JsonUtils.Transform(ctx, readerSchema, "$", message, transform).ConfigureAwait(false);`
`@@ -283,8 +296,8 @@ public override async Task<T> DeserializeAsync(ReadOnlyMemory<byte> data, bool i`
`283`	`296`	`readerSchemaJson, value, fieldTransformer)`
`284`	`297`	`.ContinueWith(t => (T)t.Result)`
`285`	`298`	`.ConfigureAwait(continueOnCapturedContext: false);`
`286`		`- }`
`287`		`-`
	`299`	`+ }`
	`300`	`+`
`288`	`301`	`return value;`
`289`	`302`	`}`
`290`	`303`	`catch (AggregateException e)`
Original file line number	Diff line number	Diff line change
`@@ -232,25 +232,30 @@ public override async Task<byte[]> SerializeAsync(T value, SerializationContext`
`232`	`232`	`{`
`233`	`233`	`serdeMutex.Release();`
`234`	`234`	`}`
`235`		`-`
	`235`	`+`
	`236`	`+ JsonSchema parsedSchema;`
`236`	`237`	`if (latestSchema != null)`
`237`	`238`	`{`
`238`		`- var latestSchemaJson = await GetParsedSchema(latestSchema).ConfigureAwait(false);`
	`239`	`+ parsedSchema = await GetParsedSchema(latestSchema).ConfigureAwait(false);`
`239`	`240`	`FieldTransformer fieldTransformer = async (ctx, transform, message) =>`
`240`	`241`	`{`
`241`		`- return await JsonUtils.Transform(ctx, latestSchemaJson, "$", message, transform).ConfigureAwait(false);`
	`242`	`+ return await JsonUtils.Transform(ctx, parsedSchema, "$", message, transform).ConfigureAwait(false);`
`242`	`243`	`};`
`243`	`244`	`value = await ExecuteRules(context.Component == MessageComponentType.Key, subject,`
`244`	`245`	`context.Topic, context.Headers, RuleMode.Write, null,`
`245`	`246`	`latestSchema, value, fieldTransformer)`
`246`	`247`	`.ContinueWith(t => (T)t.Result)`
`247`	`248`	`.ConfigureAwait(continueOnCapturedContext: false);`
`248`	`249`	`}`
	`250`	`+ else`
	`251`	`+ {`
	`252`	`+ parsedSchema = schema;`
	`253`	`+ }`
`249`	`254`
`250`	`255`	`var serializedString = Newtonsoft.Json.JsonConvert.SerializeObject(value, jsonSchemaGeneratorSettingsSerializerSettings);`
`251`	`256`	`if (validate)`
`252`	`257`	`{`
`253`		`- var validationResult = validator.Validate(serializedString, this.schema);`
	`258`	`+ var validationResult = validator.Validate(serializedString, parsedSchema);`
`254`	`259`	`if (validationResult.Count > 0)`
`255`	`260`	`{`
`256`	`261`	`throw new InvalidDataException("Schema validation failed for properties: [" + string.Join(", ", validationResult.Select(r => r.Path)) + "]");`