Add config for validating payloads against JSON Schema (#2350)

Author: rayokota

src/Confluent.SchemaRegistry.Serdes.Json/JsonDeserializer.cs

@@ -63,6 +63,8 @@ public class JsonDeserializer<T> : AsyncDeserializer<T, JsonSchema> where T : cl
 
         private JsonSchema schema = null;
 
+        private bool validate = true;
+
         private JsonSerializerSettings jsonSchemaGeneratorSettingsSerializerSettings {
             get =>
 #if NET8_0_OR_GREATER
@@ -110,6 +112,7 @@ public JsonDeserializer(ISchemaRegistryClient schemaRegistryClient, JsonDeserial
             if (config.UseLatestVersion != null) { this.useLatestVersion = config.UseLatestVersion.Value; }
             if (config.UseLatestWithMetadata != null) { this.useLatestWithMetadata = config.UseLatestWithMetadata; }
             if (config.SubjectNameStrategy != null) { this.subjectNameStrategy = config.SubjectNameStrategy.Value.ToDelegate(); }
+            if (config.Validate!= null) { this.validate= config.Validate.Value; }
         }
 
         /// <summary>
@@ -228,10 +231,9 @@ public override async Task<T> DeserializeAsync(ReadOnlyMemory<byte> data, bool i
                                 .ContinueWith(t => (JToken)t.Result)
                                 .ConfigureAwait(continueOnCapturedContext: false);
 
-                            if (schema != null)
+                            if (schema != null && validate)
                             {
                                 var validationResult = validator.Validate(json, schema);
-
                                 if (validationResult.Count > 0)
                                 {
                                     throw new InvalidDataException("Schema validation failed for properties: [" +

src/Confluent.SchemaRegistry.Serdes.Json/JsonDeserializerConfig.cs

@@ -53,6 +53,13 @@ public static class PropertyNames
             ///     Possible values: <see cref="Confluent.SchemaRegistry.SubjectNameStrategy" />
             /// </summary>
             public const string SubjectNameStrategy = "json.deserializer.subject.name.strategy";
+
+            /// <summary>
+            ///     Specifies whether to validate payloads against the schema.
+            ///
+            ///     default: true
+            /// </summary>
+            public const string Validate= "json.serializer.validate";
         }
 
 
@@ -124,5 +131,18 @@ public SubjectNameStrategy? SubjectNameStrategy
                 else { this.properties[PropertyNames.SubjectNameStrategy] = value.ToString(); }
             }
         }
+
+
+        /// <summary>
+        ///     Specifies whether or not the JSON serializer should attempt to
+        ///     validate the payload against the schema.
+        ///
+        ///     default: true
+        /// </summary>
+        public bool? Validate
+        {
+            get { return GetBool(PropertyNames.Validate); }
+            set { SetObject(PropertyNames.Validate, value); }
+        }
     }
 }

src/Confluent.SchemaRegistry.Serdes.Json/JsonSerializer.cs

@@ -75,6 +75,8 @@ public class JsonSerializer<T> : AsyncSerializer<T, JsonSchema> where T : class
         private string schemaText;
         private string schemaFullname;
 
+        private bool validate = true;
+
         private JsonSerializerSettings jsonSchemaGeneratorSettingsSerializerSettings {
             get =>
 #if NET8_0_OR_GREATER
@@ -129,6 +131,7 @@ public JsonSerializer(ISchemaRegistryClient schemaRegistryClient, JsonSerializer
             if (config.LatestCompatibilityStrict != null) { this.latestCompatibilityStrict = config.LatestCompatibilityStrict.Value; }
             if (config.UseLatestWithMetadata != null) { this.useLatestWithMetadata = config.UseLatestWithMetadata; }
             if (config.SubjectNameStrategy != null) { this.subjectNameStrategy = config.SubjectNameStrategy.Value.ToDelegate(); }
+            if (config.Validate!= null) { this.validate= config.Validate.Value; }
 
             if (this.useLatestVersion && this.autoRegisterSchema)
             {
@@ -245,10 +248,13 @@ public override async Task<byte[]> SerializeAsync(T value, SerializationContext
                 }
 
                 var serializedString = Newtonsoft.Json.JsonConvert.SerializeObject(value, jsonSchemaGeneratorSettingsSerializerSettings);
-                var validationResult = validator.Validate(serializedString, this.schema);
-                if (validationResult.Count > 0)
+                if (validate)
                 {
-                    throw new InvalidDataException("Schema validation failed for properties: [" + string.Join(", ", validationResult.Select(r => r.Path)) + "]");
+                    var validationResult = validator.Validate(serializedString, this.schema);
+                    if (validationResult.Count > 0)
+                    {
+                        throw new InvalidDataException("Schema validation failed for properties: [" + string.Join(", ", validationResult.Select(r => r.Path)) + "]");
+                    }
                 }
 
                 using (var stream = new MemoryStream(initialBufferSize))

src/Confluent.SchemaRegistry.Serdes.Json/JsonSerializerConfig.cs

@@ -92,6 +92,13 @@ public static class PropertyNames
             ///     Possible values: <see cref="Confluent.SchemaRegistry.SubjectNameStrategy" />
             /// </summary>
             public const string SubjectNameStrategy = "json.serializer.subject.name.strategy";
+
+            /// <summary>
+            ///     Specifies whether to validate payloads against the schema.
+            ///
+            ///     default: true
+            /// </summary>
+            public const string Validate= "json.serializer.validate";
         }
 
 
@@ -220,5 +227,17 @@ public SubjectNameStrategy? SubjectNameStrategy
             }
         }
 
+
+        /// <summary>
+        ///     Specifies whether or not the JSON serializer should attempt to
+        ///     validate the payload against the schema.
+        ///
+        ///     default: true
+        /// </summary>
+        public bool? Validate
+        {
+            get { return GetBool(PropertyNames.Validate); }
+            set { SetObject(PropertyNames.Validate, value); }
+        }
     }
 }