diff --git a/build.gradle b/build.gradle
index dc34b181b..a8f4905da 100644
--- a/build.gradle
+++ b/build.gradle
@@ -270,7 +270,7 @@ project('serializers:avro') {
         classifier ='all'
         mergeServiceFiles()
     }
-    
+
     tasks.build.dependsOn tasks.shadowJar
     artifacts { archives shadowJar }
     javadoc {
@@ -392,6 +392,18 @@ project('serializers') {
         testCompile group: 'io.pravega', name: 'pravega-client', version: pravegaVersion
     }
     
+    shadowJar {
+        //classifier = 'tests'
+        //from sourceSets.test.output
+        //configurations = [project.configurations.testRuntime]
+        zip64 true
+        relocate 'org.apache.avro', 'io.pravega.schemaregistry.shaded.org.apache.avro'
+        classifier ='all'
+        mergeServiceFiles()
+    }
+    
+    tasks.build.dependsOn tasks.shadowJar
+
     shadowJar {
         zip64 true
         relocate 'org.xerial.snappy' , 'io.pravega.schemaregistry.shaded.org.xerial.snappy'
@@ -409,7 +421,7 @@ project('serializers') {
         classifier ='all'
         mergeServiceFiles()
     }
-    
+
     tasks.build.dependsOn tasks.shadowJar
     artifacts { archives shadowJar }
     javadoc {
@@ -508,7 +520,7 @@ project('test') {
         compile project(':common')
         compile project(':contract')
         compile project(':client')
-        compile project(':server') 
+        compile project(':server')
         compile project(':serializers')
         compile project(':serializers:protobuf')
         compile project(':serializers:avro')
@@ -733,5 +745,4 @@ class DockerPushTask extends Exec {
             return tag
         }
     }
-}
-
+}
\ No newline at end of file
diff --git a/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/ArrayTypeComparator.java b/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/ArrayTypeComparator.java
new file mode 100644
index 000000000..037a63c80
--- /dev/null
+++ b/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/ArrayTypeComparator.java
@@ -0,0 +1,139 @@
+package io.pravega.schemaregistry.rules.jsoncompatibility;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.collect.Iterators;
+
+import java.util.Iterator;
+
+import static io.pravega.schemaregistry.rules.jsoncompatibility.BreakingChangesStore.BreakingChanges;
+
+/**
+ * this class provides methods for comparing array types in 2 schemas.
+ * the simple body check is used for instances when an array type node is found. Not when there is an array type.
+ * being defined
+ */
+public class ArrayTypeComparator {
+
+    JsonCompatibilityChecker jsonCompatibilityChecker;
+
+    public void setJsonTypeComparator() {
+        this.jsonCompatibilityChecker = new JsonCompatibilityChecker();
+    }
+
+    public BreakingChanges compareArrays(JsonNode toCheck, JsonNode toCheckAgainst) {
+        if (toCheck.isArray() && toCheckAgainst.isArray())
+            return arraySimpleBodyComparision(toCheck, toCheckAgainst);
+        else {
+            if (checkUniqueItems(toCheck, toCheckAgainst) != null)
+                return checkUniqueItems(toCheck, toCheckAgainst);
+            BreakingChanges minMaxChanges = minMaxItems(toCheck, toCheckAgainst);
+            if (minMaxChanges != null)
+                return minMaxChanges;
+            BreakingChanges additionalItemsChanges = additionalItems(toCheck, toCheckAgainst);
+            if (additionalItemsChanges != null)
+                return additionalItemsChanges;
+            BreakingChanges itemsValidationChanges = itemValidation(toCheck, toCheckAgainst);
+            if (itemsValidationChanges != null)
+                return itemsValidationChanges;
+            //TODO: Add contains and tupleValidation
+        }
+        return null;
+    }
+
+    private BreakingChanges arraySimpleBodyComparision(JsonNode toCheck, JsonNode toCheckAgainst) {
+        Iterator<String> allNodes = Iterators.concat(toCheck.fieldNames(), toCheckAgainst.fieldNames());
+        while (allNodes.hasNext()) {
+            String item = allNodes.next();
+            if (!toCheck.has(item))
+                return BreakingChanges.ARRAY_SIMPLE_BODY_CHECK_ELEMENT_REMOVED;
+            else if (!toCheckAgainst.has(item))
+                return BreakingChanges.ARRAY_SIMPLE_BODY_CHECK_ELEMENT_ADDED;
+        }
+        return null;
+    }
+
+    private BreakingChanges checkUniqueItems(JsonNode toCheck, JsonNode toCheckAgainst) {
+        if (toCheck.has("uniqueItems")) {
+            if (toCheck.get("uniqueItems").isBoolean() && toCheck.get("uniqueItems").asText() == "true") {
+                if (toCheckAgainst.get("uniqueItems") == null || (toCheckAgainst.get(
+                        "uniqueItems").isBoolean() && toCheckAgainst.get(
+                        "uniqueItems").asText().equals("false")))
+                    return BreakingChanges.ARRAY_UNIQUE_ITEMS_CONDITION_ENABLED;
+            }
+        }
+        return null;
+    }
+
+    private BreakingChanges minMaxItems(JsonNode toCheck, JsonNode toCheckAgainst) {
+        if (toCheck.get("maxItems") != null && toCheckAgainst.get("maxItems") == null)
+            return BreakingChanges.ARRAY_MAX_ITEMS_CONDITION_ADDED;
+        else if (toCheck.get("maxItems") != null && toCheckAgainst.get("maxItems") != null) {
+            int originalMaxLimit = toCheckAgainst.get("maxItems").asInt();
+            int changedMaxLimit = toCheck.get("maxItems").asInt();
+            if (changedMaxLimit < originalMaxLimit)
+                return BreakingChanges.ARRAY_MAX_ITEMS_VALUE_DECREASED;
+        }
+        if (toCheck.get("minItems") != null && toCheckAgainst.get("minItems") == null)
+            return BreakingChanges.ARRAY_MIN_ITEMS_CONDITION_ADDED;
+        else if (toCheck.get("minItems") != null && toCheckAgainst.get("minItems") != null) {
+            int originalMinLimit = toCheckAgainst.get("minItems").asInt();
+            int changedMinLimit = toCheck.get("minItems").asInt();
+            if (changedMinLimit > originalMinLimit)
+                return BreakingChanges.ARRAY_MIN_ITEMS_VALUE_INCREASED;
+        }
+        return null;
+    }
+
+    private BreakingChanges additionalItems(JsonNode toCheck, JsonNode toCheckAgainst) {
+        if (toCheck.get("additionalItems") != null && toCheckAgainst.get("additionalItems") == null) {
+            if (toCheck.get("additionalItems").isBoolean() && toCheck.get("additionalItems").asText() == "false")
+                return BreakingChanges.ARRAY_ADDITIONAL_ITEMS_DISABLED;
+            else if (toCheck.get("additionalItems").isObject())
+                return BreakingChanges.ARRAY_ADDITIONAL_ITEMS_SCOPE_DECREASED;
+        } else if (toCheck.get("additionalItems") != null && toCheckAgainst.get("additionalItems") != null) {
+            if (toCheck.get("additionalItems").isBoolean() && toCheck.get("additionalItems").asText() == "false") {
+                if (!(toCheckAgainst.get("additionalItems").asText() == "false"))
+                    return BreakingChanges.ARRAY_ADDITIONAL_ITEMS_DISABLED;
+            } else if (toCheck.get("additionalItems").isObject()) {
+                if (toCheckAgainst.get("additionalItems").isObject()) {
+                    if (jsonCompatibilityChecker.checkNodeType(toCheck.get("additionalItems"),
+                            toCheckAgainst.get("additionalItems")) != null)
+                        return BreakingChanges.ARRAY_ADDITIONAL_ITEMS_SCOPE_INCOMPATIBLE_CHANGE;
+                } else if (toCheckAgainst.get("additionalItems").isBoolean() && toCheckAgainst.get(
+                        "additionalItems").asText() == "true")
+                    return BreakingChanges.ARRAY_ADDITIONAL_ITEMS_SCOPE_DECREASED;
+            }
+        }
+        return null;
+    }
+
+    private BreakingChanges itemValidation(JsonNode toCheck, JsonNode toCheckAgainst) {
+        if (!toCheck.has("items") && !toCheckAgainst.has("items"))
+            return null;
+        return jsonCompatibilityChecker.checkNodeType(toCheck.get("items"), toCheckAgainst.get("items"));
+    }
+
+    private boolean isDynamicArray(JsonNode node) {
+        if (node.get("additionalItems") == null)
+            return true;
+        else if (node.get("additionalItems").isBoolean()) {
+            if (node.get("additionalItems").asText() == "true")
+                return true;
+        }
+        return false;
+    }
+
+    private boolean isDynamicArrayWithCondition(JsonNode node) {
+        if (node.get("additionalItems").isObject())
+            return true;
+        return false;
+    }
+
+    private boolean isStaticArray(JsonNode node) {
+        if (node.get("additionalItems").isBoolean()) {
+            if (node.get("additionalItems").asText() == "false")
+                return true;
+        }
+        return false;
+    }
+}
diff --git a/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/BreakingChangesStore.java b/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/BreakingChangesStore.java
new file mode 100644
index 000000000..a09688c78
--- /dev/null
+++ b/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/BreakingChangesStore.java
@@ -0,0 +1,92 @@
+package io.pravega.schemaregistry.rules.jsoncompatibility;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+public class BreakingChangesStore {
+    //includes only those changes that lead to incompatibility
+    protected enum BreakingChanges {
+        // data type mismatch
+        DATA_TYPE_MISMATCH,
+        // strings
+        STRING_TYPE_MAX_LENGTH_ADDED,
+        STRING_TYPE_MAX_LENGTH_VALUE_DECREASED,
+        STRING_TYPE_MIN_LENGTH_ADDED,
+        STRING_TYPE_MIN_LENGTH_VALUE_INCREASED,
+        STRING_TYPE_PATTERN_ADDED,
+        STRING_TYPE_PATTERN_MODIFIED,
+        // NUMBERS
+        NUMBER_TYPE_MAXIMUM_VALUE_ADDED,
+        NUMBER_TYPE_MAXIMUM_VALUE_DECREASED,
+        NUMBER_TYPE_EXCLUSIVE_MAXIMUM_VALUE_ADDED,
+        NUMBER_TYPE_EXCLUSIVE_MAXIMUM_VALUE_DECREASED,
+        NUMBER_TYPE_MINIMUM_VALUE_ADDED,
+        NUMBER_TYPE_MINIMUM_VALUE_INCREASED,
+        NUMBER_TYPE_EXCLUSIVE_MINIMUM_VALUE_ADDED,
+        NUMBER_TYPE_EXCLUSIVE_MINIMUM_VALUE_INCREASED,
+        NUMBER_TYPE_MULTIPLE_OF_ADDED,
+        NUMBER_TYPE_MULTIPLE_OF_NON_DIVISIBLE_CHANGE,
+        NUMBER_TYPE_MULTIPLE_OF_INCREASED,
+        NUMBER_TYPE_CHANGED_FROM_NUMBER_TO_INTEGER,
+        //ARRAYS
+        ARRAY_MAX_ITEMS_CONDITION_ADDED,
+        ARRAY_MAX_ITEMS_VALUE_DECREASED,
+        ARRAY_MIN_ITEMS_CONDITION_ADDED,
+        ARRAY_MIN_ITEMS_VALUE_INCREASED,
+        ARRAY_UNIQUE_ITEMS_CONDITION_ENABLED,
+        ARRAY_ADDITIONAL_ITEMS_DISABLED,
+        ARRAY_ADDITIONAL_ITEMS_SCOPE_DECREASED,
+        ARRAY_ADDITIONAL_ITEMS_SCOPE_INCOMPATIBLE_CHANGE,
+        ARRAY_SIMPLE_BODY_CHECK_ELEMENT_REMOVED,
+        ARRAY_SIMPLE_BODY_CHECK_ELEMENT_ADDED,
+        ITEM_REMOVED_NOT_COVERED_BY_PARTIALLY_OPEN_CONTENT_MODEL,
+        ITEM_REMOVED_FROM_CLOSED_CONTENT_MODEL,
+        ITEM_ADDED_TO_OPEN_CONTENT_MODEL,
+        ITEM_ADDED_NOT_COVERED_BY_PARTIALLY_OPEN_CONTENT_MODEL,
+        // PROPERTIES
+        PROPERTIES_SECTION_ADDED,     // this would be taken care of by checking for required and additional properties since that is when incompatibility arises
+        PROPERTY_REMOVED_NOT_PART_OF_DYNAMIC_PROPERTY_SET_WITH_CONDITION, // check on updated schema
+        PROPERTY_REMOVED_FROM_STATIC_PROPERTY_SET,   // check on updated schema
+        PROPERTY_ADDED_TO_DYNAMIC_PROPERTY_SET, // check on original schema
+        PROPERTY_ADDED_NOT_PART_OF_DYNAMIC_PROPERTY_SET_WITH_CONDITION,   // check on original schema
+        REQUIRED_PROPERTY_ADDED_WITHOUT_DEFAULT,
+        REQUIRED_ATTRIBUTE_ADDED, // may not be needed
+        MAX_PROPERTIES_ADDED,
+        MAX_PROPERTIES_LIMIT_DECREASED,
+        MIN_PROPERTIES_ADDED,
+        MIN_PROPERTIES_LIMIT_INCREASED,
+        ADDITIONAL_PROPERTIES_REMOVED,
+        ADDITIONAL_PROPERTIES_NON_COMPATIBLE_CHANGE,
+        // DEPENDENCIES
+        DEPENDENCY_SECTION_ADDED,
+        DEPENDENCY_ADDED_IN_ARRAY_FORM,
+        DEPENDENCY_ARRAY_ELEMENTS_NON_REMOVAL,
+        DEPENDENCY_ADDED_IN_SCHEMA_FORM,
+        DEPENDENCY_IN_SCHEMA_FORM_MODIFIED,
+        // ENUM
+        ENUM_TYPE_ADDED,
+        ENUM_TYPE_ARRAY_CONTENTS_NON_ADDITION_OF_ELEMENTS,
+        // NOT TYPE
+        NOT_TYPE_EXTENDED,
+        // COMBINED
+        SUBSCHEMA_TYPE_ADDED,
+        SUBSCHEMA_TYPE_CHANGED,
+        SUBSCHEMA_TYPE_ALL_OF_SCHEMAS_INCREASED,
+        SUBSCHEMA_TYPE_ALL_OF_SCHEMAS_CHANGED,
+        SUBSCHEMA_TYPE_ONE_OF_SCHEMAS_DECREASED,
+        SUBSCHEMA_TYPE_ONE_OF_SCHEMAS_CHANGED,
+        SUBSCHEMA_TYPE_ANY_OF_SCHEMAS_DECREASED,
+        SUBSCHEMA_TYPE_ANYOF_SCHEMAS_CHANGED
+    }
+    
+    private List<BreakingChanges> breakingChangesList = new ArrayList<>();
+    
+    private void computeBreakingChanges() {
+        breakingChangesList = Arrays.asList(BreakingChanges.values());
+    }
+    
+    public List<BreakingChanges> getBreakingChangesList() {
+        return breakingChangesList;
+    }
+}
diff --git a/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/DependenciesComparator.java b/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/DependenciesComparator.java
new file mode 100644
index 000000000..62dbf13e4
--- /dev/null
+++ b/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/DependenciesComparator.java
@@ -0,0 +1,59 @@
+package io.pravega.schemaregistry.rules.jsoncompatibility;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.google.common.collect.Iterators;
+
+import java.util.Iterator;
+
+import static io.pravega.schemaregistry.rules.jsoncompatibility.BreakingChangesStore.*;
+
+public class DependenciesComparator {
+    JsonCompatibilityCheckerUtils jsonCompatibilityCheckerUtils = new JsonCompatibilityCheckerUtils();
+    PropertiesComparator propertiesComparator = new PropertiesComparator();
+    public BreakingChanges checkDependencies(JsonNode toCheck, JsonNode toCheckAgainst) {
+        propertiesComparator.setJsonCompatibilityChecker();
+        if(toCheck.get("dependencies") != null && toCheckAgainst.get("dependencies") == null)
+            return BreakingChanges.DEPENDENCY_SECTION_ADDED;
+        else if(toCheck.get("dependencies") == null && toCheckAgainst.get("dependencies") == null)
+            return null;
+        else if(toCheck.get("dependencies") == null && toCheckAgainst.get("dependencies") != null)
+            return null;
+        Iterator<String> dependencyFields = Iterators.concat(toCheck.get("dependencies").fieldNames(), toCheckAgainst.get("dependencies").fieldNames());
+        boolean dependencyTypeIsArray = checkDependencyTypeIsArray(toCheck.get("dependencies"), toCheckAgainst.get("dependencies"));
+        while(dependencyFields.hasNext()) {
+            String field = dependencyFields.next();
+               if(toCheck.get("dependencies").get(field) != null && toCheckAgainst.get("dependencies").get(field)==null) {
+                   if(dependencyTypeIsArray)
+                       return BreakingChanges.DEPENDENCY_ADDED_IN_ARRAY_FORM;
+                   else 
+                       return BreakingChanges.DEPENDENCY_ADDED_IN_SCHEMA_FORM;
+               }
+               else if (toCheck.get("dependencies").get(field) != null && toCheckAgainst.get("dependencies").get(field) != null) {
+                   if(dependencyTypeIsArray) {
+                       // check the value returned by the array comparator
+                       if(!jsonCompatibilityCheckerUtils.arrayComparisionOnlyRemoval((ArrayNode) toCheck.get("dependencies").get(field),
+                               (ArrayNode) toCheckAgainst.get("dependencies").get(field)))
+                           return BreakingChanges.DEPENDENCY_ARRAY_ELEMENTS_NON_REMOVAL;
+                   }
+                   else {
+                       if(propertiesComparator.checkProperties(toCheck.get("dependencies").get(field),
+                               toCheckAgainst.get("dependencies").get(field)) != null)
+                           return BreakingChanges.DEPENDENCY_IN_SCHEMA_FORM_MODIFIED;
+                   }
+               }
+        }
+        return null;
+    }
+    
+    private boolean checkDependencyTypeIsArray(JsonNode toCheck, JsonNode toCheckAgainst) {
+        // it is assumed that the dependency type does not change from array to schema or vice versa.
+        Iterator<String> toCheckFields = toCheck.fieldNames();
+        String toCheckSample = toCheckFields.next();
+        Iterator<String> toCheckAgainstFields = toCheckAgainst.fieldNames();
+        String toCheckAgainstSample = toCheckAgainstFields.next();
+        if(toCheck.get(toCheckSample).isArray() && toCheckAgainst.get(toCheckAgainstSample).isArray())
+            return true;
+        return false;
+    }
+}
diff --git a/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/EnumComparator.java b/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/EnumComparator.java
new file mode 100644
index 000000000..12492aafc
--- /dev/null
+++ b/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/EnumComparator.java
@@ -0,0 +1,19 @@
+package io.pravega.schemaregistry.rules.jsoncompatibility;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+
+import static io.pravega.schemaregistry.rules.jsoncompatibility.BreakingChangesStore.*;
+
+public class EnumComparator {
+    JsonCompatibilityCheckerUtils jsonCompatibilityCheckerUtils = new JsonCompatibilityCheckerUtils();
+    public BreakingChanges enumComparator(JsonNode toCheck, JsonNode toCheckAgainst) {
+        if(toCheck.has("enum") && !toCheckAgainst.has("enum"))
+            return BreakingChanges.ENUM_TYPE_ADDED;
+        else if(toCheck.has("enum") && toCheck.has("enum")) {
+            if(!jsonCompatibilityCheckerUtils.arrayComparisionOnlyAddition((ArrayNode)toCheck.get("enum"), (ArrayNode)toCheckAgainst.get("enum")))
+                return BreakingChanges.ENUM_TYPE_ARRAY_CONTENTS_NON_ADDITION_OF_ELEMENTS;
+        }
+        return null;
+    }
+}
diff --git a/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/JsonCompatibilityChecker.java b/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/JsonCompatibilityChecker.java
new file mode 100644
index 000000000..14e472785
--- /dev/null
+++ b/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/JsonCompatibilityChecker.java
@@ -0,0 +1,117 @@
+package io.pravega.schemaregistry.rules.jsoncompatibility;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.pravega.schemaregistry.contract.data.SchemaInfo;
+import io.pravega.schemaregistry.rules.CompatibilityChecker;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static io.pravega.schemaregistry.rules.jsoncompatibility.BreakingChangesStore.BreakingChanges;
+
+public class JsonCompatibilityChecker implements CompatibilityChecker {
+    EnumComparator enumComparator;
+    JsonCompatibilityCheckerUtils jsonCompatibilityCheckerUtils;
+    ObjectTypeComparator objectTypeComparator;
+    NumberComparator numberComparator;
+    StringComparator stringComparator;
+    ArrayTypeComparator arrayTypeComparator;
+    SubSchemaComparator subSchemaComparator;
+    
+    public JsonCompatibilityChecker() {
+        this.enumComparator = new EnumComparator();
+        this.jsonCompatibilityCheckerUtils = new JsonCompatibilityCheckerUtils();
+        this.objectTypeComparator = new ObjectTypeComparator();
+        this.numberComparator = new NumberComparator();
+        this.stringComparator = new StringComparator();
+        this.arrayTypeComparator = new ArrayTypeComparator();
+        this.subSchemaComparator = new SubSchemaComparator();
+    }
+
+    @Override
+    public boolean canRead(SchemaInfo readUsing, List<SchemaInfo> writtenUsing) {
+        try {
+            return canReadChecker(readUsing, writtenUsing);
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+        return false;
+    }
+
+    @Override
+    public boolean canBeRead(SchemaInfo writtenUsing, List<SchemaInfo> readUsing) {
+        return readUsing.stream().map(x -> canRead(x, Collections.singletonList(writtenUsing))).allMatch(x -> x.equals(true));
+    }
+
+    @Override
+    public boolean canMutuallyRead(SchemaInfo schema, List<SchemaInfo> schemaList) {
+        return schemaList.stream().map(x -> canRead(schema, Collections.singletonList(x)) && canBeRead(x,
+                Collections.singletonList(schema))).allMatch(x -> x.equals(true));
+    }
+
+
+    private boolean canReadChecker(SchemaInfo toValidate, List<SchemaInfo> toValidateAgainst) throws IOException {
+        ObjectMapper objectMapper = new ObjectMapper();
+        JsonNode toCheck = objectMapper.readTree(toValidate.getSchemaData().array());
+        List<JsonNode> toCheckAgainst = toValidateAgainst.stream().map(x -> {
+            try {
+                return objectMapper.readTree(x.getSchemaData().array());
+            } catch (IOException e) {
+                e.printStackTrace();
+            }
+            return null;
+        }).collect(
+                Collectors.toList());
+        return !toCheckAgainst.stream().map(x -> checkNodeType(toCheck, x)).anyMatch(x -> x != null);
+    }
+
+    protected BreakingChanges checkNodeType(JsonNode toCheck, JsonNode toCheckAgainst) {
+        if(jsonCompatibilityCheckerUtils.hasSubSchema(toCheck)) {
+            subSchemaComparator.setJsonCompatibilityChecker();
+            BreakingChanges subSchemaChanges = subSchemaComparator.checkSubSchemas(toCheck, toCheckAgainst);
+            if(subSchemaChanges != null)
+                return subSchemaChanges;
+        }
+        if (toCheck.has("enum") || toCheckAgainst.has("enum")) {
+            BreakingChanges enumChanges = enumComparator.enumComparator(toCheck, toCheckAgainst);
+            if (enumChanges != null)
+                return enumChanges;
+        }
+        String nodeType = jsonCompatibilityCheckerUtils.getTypeValue(toCheck).equals(
+                jsonCompatibilityCheckerUtils.getTypeValue(
+                        toCheckAgainst)) ? jsonCompatibilityCheckerUtils.getTypeValue(toCheck) : "mismatch";
+        switch (nodeType) {
+            case "object":
+                return objectTypeComparator.checkAspects(toCheck, toCheckAgainst);
+            case "number":
+            case "integer":
+                return numberComparator.compareNumbers(toCheck, toCheckAgainst);
+            case "string":
+                return stringComparator.stringComparator(toCheck, toCheckAgainst);
+            case "array":
+                arrayTypeComparator.setJsonTypeComparator();
+                return arrayTypeComparator.compareArrays(toCheck, toCheckAgainst);
+            case "boolean":
+                break;
+            case "null":
+                break;
+            case "mismatch":
+                return analyzeMismatch(toCheck, toCheckAgainst);
+        }
+        return null;
+    }
+
+    private BreakingChanges analyzeMismatch(JsonNode toCheck, JsonNode toCheckAgainst) {
+        if ((jsonCompatibilityCheckerUtils.getTypeValue(toCheck).equals(
+                "number") || jsonCompatibilityCheckerUtils.getTypeValue(toCheck).equals(
+                "integer")) && jsonCompatibilityCheckerUtils.getTypeValue(toCheckAgainst).equals(
+                "number") || jsonCompatibilityCheckerUtils.getTypeValue(toCheckAgainst).equals("integer"))
+            return numberComparator.compareNumbers(toCheck, toCheckAgainst);
+        else
+            return BreakingChanges.DATA_TYPE_MISMATCH;
+    }
+
+}
diff --git a/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/JsonCompatibilityCheckerUtils.java b/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/JsonCompatibilityCheckerUtils.java
new file mode 100644
index 000000000..08c64de0b
--- /dev/null
+++ b/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/JsonCompatibilityCheckerUtils.java
@@ -0,0 +1,89 @@
+package io.pravega.schemaregistry.rules.jsoncompatibility;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+
+import java.util.Iterator;
+
+public class JsonCompatibilityCheckerUtils {
+    
+    public String getTypeValue(JsonNode node) {
+        String value = null;
+        Iterator<String> fieldNames = node.fieldNames();
+        while(fieldNames.hasNext()) {
+            if (node.fieldNames().next().equals("type"))
+                value = node.get("type").textValue();
+                break;
+        }
+        return value;
+    }
+    
+    public boolean hasDynamicPropertySet(JsonNode node) {
+        if(node.get("additionalProperties")==null && node.get("patternProperties")==null)
+            return true;
+        else if(node.get("additionalProperties").isBoolean() && node.get("patternProperties") == null) {
+            if(node.get("additionalProperties").asText() == "true") {
+                return true;
+            }
+        }
+        return false;
+    }
+    
+    public boolean hasStaticPropertySet(JsonNode node) {
+        if(node.get("patternProperties") == null && node.get("additionalProperties").asText() == "false")
+            return true;
+        return false;
+    }
+    
+    public boolean isInRequired(String toSearch, JsonNode toSearchIn) {
+        if(toSearchIn.get("required") != null) {
+            ArrayNode arrayToSearch = (ArrayNode) toSearchIn.get("required");
+            for(int i=0;i<arrayToSearch.size();i++) {
+                if(arrayToSearch.get(i).asText().equals(toSearch))
+                    return true;
+            }
+        }
+        return false;
+    }
+    
+    public boolean arrayComparisionOnlyRemoval(ArrayNode toCheck, ArrayNode toCheckAgainst) {
+        // every element in toCheck array must be in toCheckAgainst
+        for(int i=0;i<toCheck.size();i++) {
+            int flag =0;
+            String toSearch = toCheck.get(i).asText();
+            for(int j=0;j<toCheckAgainst.size();j++) {
+                if(toSearch.equals(toCheckAgainst.get(j).asText())) {
+                    flag = 1;
+                    break;
+                }
+            }
+            if(flag==0)
+                return false;
+        }
+        return true;
+    }
+    
+    public boolean arrayComparisionOnlyAddition(ArrayNode toCheck, ArrayNode toCheckAgainst) {
+        // every element in toCheckAgainst array must be in toCheck.
+        for(int i=0;i<toCheckAgainst.size();i++) {
+            int flag =0;
+            String toSearch = toCheckAgainst.get(i).asText();
+            for(int j=0;j<toCheck.size();j++) {
+                if(toSearch.equals(toCheck.get(j).asText())) {
+                    flag = 1;
+                    break;
+                }
+            }
+            if(flag==0)
+                return false;
+        }
+        return true;
+    }
+    
+    public boolean hasSubSchema(JsonNode toCheck) {
+        if(toCheck.has("anyOf") || toCheck.has("allOf") || toCheck.has("oneOf")) {
+            return true;
+        }
+        return false;
+    }
+}
diff --git a/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/NumberComparator.java b/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/NumberComparator.java
new file mode 100644
index 000000000..fbd6c2a2c
--- /dev/null
+++ b/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/NumberComparator.java
@@ -0,0 +1,98 @@
+package io.pravega.schemaregistry.rules.jsoncompatibility;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+import static io.pravega.schemaregistry.rules.jsoncompatibility.BreakingChangesStore.*;
+
+public class NumberComparator {
+    public BreakingChanges compareNumbers(JsonNode toCheck, JsonNode toCheckAgainst) {
+        if(maximumComparator(toCheck, toCheckAgainst) != null)
+            return maximumComparator(toCheck, toCheckAgainst);
+        if(minimumComparator(toCheck, toCheckAgainst) != null)
+            return minimumComparator(toCheck, toCheckAgainst);
+        if(multipleOfComparator(toCheck, toCheckAgainst) != null)
+            return multipleOfComparator(toCheck, toCheckAgainst);
+        if(typeChanged(toCheck, toCheckAgainst) != null)
+            return typeChanged(toCheck, toCheckAgainst);
+        return null;
+        //TO DO - Changes from exclusive max/min to max/min & vice versa.
+    }
+    
+    private BreakingChanges maximumComparator(JsonNode toCheck, JsonNode toCheckAgainst) {
+        if(toCheck.get("maximum") != null && toCheckAgainst.get("maximum") == null)
+            return BreakingChanges.NUMBER_TYPE_MAXIMUM_VALUE_ADDED;
+        else if(toCheck.get("maximum") != null && toCheckAgainst.get("maximum") != null) {
+            int toCheckAgainstMaximum = toCheckAgainst.get("maximum").intValue();
+            int toCheckMaximum = toCheck.get("maximum").intValue();
+            if(toCheckMaximum < toCheckAgainstMaximum) {
+                return BreakingChanges.NUMBER_TYPE_MAXIMUM_VALUE_DECREASED;
+            }
+        }
+        else if(toCheck.get("exclusiveMaximum") != null && toCheckAgainst.get("exclusiveMaximum") == null) {
+            return BreakingChanges.NUMBER_TYPE_EXCLUSIVE_MAXIMUM_VALUE_ADDED;
+        }
+        else if(toCheck.get("exclusiveMaximum") != null && toCheckAgainst.get("exclusiveMaximum") != null) {
+            if(toCheckAgainst.get("exclusiveMaximum").isBoolean()) {
+                if(toCheckAgainst.get("exclusiveMaximum").asText() == "false" && toCheck.get("exclusiveMaximum").asText() == "true")
+                    return BreakingChanges.NUMBER_TYPE_EXCLUSIVE_MAXIMUM_VALUE_ADDED;
+            }
+            else {
+                int toCheckAgainstExclusiveMaximum = toCheckAgainst.get("exclusiveMaximum").asInt();
+                int toCheckExclusiveMaximum = toCheck.get("exclusiveMaximum").asInt();
+                if(toCheckExclusiveMaximum < toCheckAgainstExclusiveMaximum)
+                    return BreakingChanges.NUMBER_TYPE_EXCLUSIVE_MAXIMUM_VALUE_DECREASED;
+            }
+        }
+        return null;
+    }
+    
+    private BreakingChanges minimumComparator(JsonNode toCheck, JsonNode toCheckAgainst) {
+        if(toCheck.get("minimum") != null && toCheckAgainst.get("minimum") == null)
+            return BreakingChanges.NUMBER_TYPE_MINIMUM_VALUE_ADDED;
+        else if(toCheck.get("minimum") != null && toCheckAgainst.get("minimum") != null) {
+            int toCheckAgainstMinimum = toCheckAgainst.get("minimum").intValue();
+            int toCheckMinimum = toCheck.get("minimum").intValue();
+            if(toCheckMinimum > toCheckAgainstMinimum) {
+                return BreakingChanges.NUMBER_TYPE_MINIMUM_VALUE_INCREASED;
+            }
+        }
+        else if(toCheck.get("exclusiveMinimum") != null && toCheckAgainst.get("exclusiveMinimum") == null) {
+            return BreakingChanges.NUMBER_TYPE_EXCLUSIVE_MINIMUM_VALUE_ADDED;
+        }
+        else if(toCheck.get("exclusiveMinimum") != null && toCheckAgainst.get("exclusiveMinimum") != null) {
+            if(toCheckAgainst.get("exclusiveMinimum").isBoolean()) {
+                if(toCheckAgainst.get("exclusiveMinimum").asText() == "false" && toCheck.get("exclusiveMinimum").asText() == "true")
+                    return BreakingChanges.NUMBER_TYPE_EXCLUSIVE_MINIMUM_VALUE_ADDED;
+            }
+            else {
+                int toCheckAgainstExclusiveMinimum = toCheckAgainst.get("exclusiveMinimum").asInt();
+                int toCheckExclusiveMinimum = toCheck.get("exclusiveMinimum").asInt();
+                if(toCheckExclusiveMinimum > toCheckAgainstExclusiveMinimum)
+                    return BreakingChanges.NUMBER_TYPE_EXCLUSIVE_MINIMUM_VALUE_INCREASED;
+            }
+        }
+        return null;
+    }
+    
+    private BreakingChanges multipleOfComparator(JsonNode toCheck, JsonNode toCheckAgainst) {
+        if(toCheck.get("multipleOf") != null && toCheckAgainst.get("multipleOf") == null)
+            return BreakingChanges.NUMBER_TYPE_MULTIPLE_OF_ADDED;
+        else if(toCheck.get("multipleOf") != null && toCheckAgainst.get("multipleOf") != null) {
+            int toCheckMultipleOf = toCheck.get("multipleOf").asInt();
+            int toCheckAgainstMultipleOf = toCheckAgainst.get("multipleOf").asInt();
+            if(toCheckAgainstMultipleOf != toCheckMultipleOf) {
+                if(toCheckMultipleOf%toCheckAgainstMultipleOf == 0)
+                    return BreakingChanges.NUMBER_TYPE_MULTIPLE_OF_INCREASED;
+                else if (toCheckAgainstMultipleOf%toCheckMultipleOf != 0)
+                    return BreakingChanges.NUMBER_TYPE_MULTIPLE_OF_NON_DIVISIBLE_CHANGE;
+            }
+        }
+        return null;
+    }
+    
+    private BreakingChanges typeChanged(JsonNode toCheck, JsonNode toCheckAgainst) {
+        if(toCheck.get("type").asText().equals("integer") && toCheckAgainst.get("type").asText().equals("number"))
+            return BreakingChanges.NUMBER_TYPE_CHANGED_FROM_NUMBER_TO_INTEGER;
+        return null;
+    }
+}
diff --git a/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/ObjectTypeComparator.java b/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/ObjectTypeComparator.java
new file mode 100644
index 000000000..b3e4dee6e
--- /dev/null
+++ b/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/ObjectTypeComparator.java
@@ -0,0 +1,23 @@
+package io.pravega.schemaregistry.rules.jsoncompatibility;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+import static io.pravega.schemaregistry.rules.jsoncompatibility.BreakingChangesStore.*;
+
+public class ObjectTypeComparator {
+    
+    public BreakingChanges checkAspects(JsonNode toCheck, JsonNode toCheckAgainst) {
+        // will check for properties,dependencies, required, additional properties by calling required classes.
+        PropertiesComparator propertiesComparator = new PropertiesComparator();
+        propertiesComparator.setJsonCompatibilityChecker();
+        DependenciesComparator dependenciesComparator = new DependenciesComparator();
+        BreakingChanges propertiesDifference = propertiesComparator.checkProperties(toCheck, toCheckAgainst);    
+        if (propertiesDifference != null)
+            return propertiesDifference;
+        BreakingChanges dependenciesDifference = dependenciesComparator.checkDependencies(toCheck, toCheckAgainst);
+        if(dependenciesDifference != null)
+            return dependenciesDifference;
+        return null;
+    }
+    
+}
diff --git a/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/PropertiesComparator.java b/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/PropertiesComparator.java
new file mode 100644
index 000000000..627ae8868
--- /dev/null
+++ b/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/PropertiesComparator.java
@@ -0,0 +1,123 @@
+package io.pravega.schemaregistry.rules.jsoncompatibility;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.google.common.collect.Iterators;
+import io.pravega.schemaregistry.rules.jsoncompatibility.BreakingChangesStore.BreakingChanges;
+
+import java.util.Iterator;
+
+public class PropertiesComparator {
+    JsonCompatibilityChecker jsonCompatibilityChecker;
+    JsonCompatibilityCheckerUtils jsonCompatibilityCheckerUtils = new JsonCompatibilityCheckerUtils();
+
+    public void setJsonCompatibilityChecker() {
+        this.jsonCompatibilityChecker = new JsonCompatibilityChecker();
+    }
+
+    public BreakingChanges checkProperties(JsonNode toCheck, JsonNode toCheckAgainst) {
+        Iterator<String> propertyFields = Iterators.concat(toCheck.get("properties").fieldNames(),
+                toCheckAgainst.get("properties").fieldNames());
+        while (propertyFields.hasNext()) {
+            String propertyField = propertyFields.next();
+            if (toCheck.get("properties").get(propertyField) == null) {
+                // property has been removed from toCheck
+                if (jsonCompatibilityCheckerUtils.hasStaticPropertySet(toCheck))
+                    return BreakingChanges.PROPERTY_REMOVED_FROM_STATIC_PROPERTY_SET;
+                if (!jsonCompatibilityCheckerUtils.hasStaticPropertySet(
+                        toCheck) && !jsonCompatibilityCheckerUtils.hasDynamicPropertySet(toCheck)) {
+                    // assume that pattern properties always matches
+                    // TODO: add regex check for pattern properties
+                    BreakingChanges breakingChanges = jsonCompatibilityChecker.checkNodeType(
+                            toCheck.get("additionalProperties"),
+                            toCheckAgainst.get("properties").get(propertyField));
+                    if (breakingChanges != null)
+                        return BreakingChanges.PROPERTY_REMOVED_NOT_PART_OF_DYNAMIC_PROPERTY_SET_WITH_CONDITION;
+                }
+            } else if (toCheckAgainst.get("properties").get(propertyField) == null) {
+                // property has been added to toCheck
+                if (jsonCompatibilityCheckerUtils.hasDynamicPropertySet(toCheckAgainst))
+                    return BreakingChanges.PROPERTY_ADDED_TO_DYNAMIC_PROPERTY_SET;
+                //check if required property in toCheck
+                if (toCheck.get("required") != null) {
+                    if (jsonCompatibilityCheckerUtils.isInRequired(propertyField, toCheck)) {
+                        if (toCheck.get("properties").get(propertyField).get("default") == null)
+                            return BreakingChanges.REQUIRED_PROPERTY_ADDED_WITHOUT_DEFAULT;
+                    }
+                }
+                if (!jsonCompatibilityCheckerUtils.hasStaticPropertySet(
+                        toCheckAgainst) && !jsonCompatibilityCheckerUtils.hasDynamicPropertySet(toCheckAgainst)) {
+                    // assume that pattern properties always matches
+                    // TODO: add regex check for pattern properties
+                    BreakingChanges breakingChanges = jsonCompatibilityChecker.checkNodeType(
+                            toCheck.get("properties").get(propertyField),
+                            toCheckAgainst.get("additionalProperties"));
+                    if (breakingChanges != null)
+                        return BreakingChanges.PROPERTY_ADDED_NOT_PART_OF_DYNAMIC_PROPERTY_SET_WITH_CONDITION;
+                }
+            } else if (toCheckAgainst.get("properties").get(propertyField) != null && toCheck.get("properties").get(
+                    propertyField) != null) {
+                 BreakingChanges singlePropertyChanges = jsonCompatibilityChecker.checkNodeType(toCheck.get("properties").get(propertyField),
+                        toCheckAgainst.get("properties").get(propertyField));
+                 if(singlePropertyChanges != null)
+                     return singlePropertyChanges;
+            }
+        }
+        // check for min-max conditions on properties
+        BreakingChanges minMaxBreakingChanges = minMaxProperties(toCheck, toCheckAgainst);
+        if (minMaxBreakingChanges != null)
+            return minMaxBreakingChanges;
+        // check for additional properties
+        BreakingChanges additionalPropsBreakingChanges = additionalProperties(toCheck, toCheckAgainst);
+        if (additionalPropsBreakingChanges != null)
+            return additionalPropsBreakingChanges;
+        // check for required properties
+        BreakingChanges requiredPropsBreakingChanges = requiredProperties(toCheck, toCheckAgainst);
+        if (requiredPropsBreakingChanges != null)
+            return requiredPropsBreakingChanges;
+        return null;
+    }
+
+    private BreakingChanges minMaxProperties(JsonNode toCheck, JsonNode toCheckAgainst) {
+        // minProperties
+        if (toCheck.get("minProperties") != null && toCheckAgainst.get("minProperties") == null)
+            return BreakingChanges.MIN_PROPERTIES_ADDED;
+        else if (toCheck.get("minProperties") != null && toCheckAgainst.get("minProperties") != null) {
+            if (toCheck.get("minProperties").intValue() > toCheckAgainst.get("minProperties").intValue())
+                return BreakingChanges.MIN_PROPERTIES_LIMIT_INCREASED;
+        }
+        // maxProperties
+        if (toCheck.get("maxProperties") != null && toCheckAgainst.get("maxProperties") == null)
+            return BreakingChanges.MAX_PROPERTIES_ADDED;
+        else if (toCheck.get("maxProperties") != null && toCheckAgainst.get("maxProperties") != null) {
+            if (toCheck.get("maxProperties").intValue() < toCheckAgainst.get("maxProperties").intValue())
+                return BreakingChanges.MAX_PROPERTIES_LIMIT_DECREASED;
+        }
+        return null;
+    }
+
+    private BreakingChanges additionalProperties(JsonNode toCheck, JsonNode toCheckAgainst) {
+        if (toCheck.get("additionalProperties") == null && toCheckAgainst.get("additionalProperties") != null)
+            return BreakingChanges.ADDITIONAL_PROPERTIES_REMOVED;
+        else if (toCheck.get("additionalProperties") != null && toCheckAgainst.get("additionalProperties") != null) {
+            BreakingChanges additionalPropertiesBreakingChanges = jsonCompatibilityChecker.checkNodeType(
+                    toCheck.get("additionalProperties"), toCheckAgainst.get("additionalProperties"));
+            if (additionalPropertiesBreakingChanges != null)
+                return BreakingChanges.ADDITIONAL_PROPERTIES_NON_COMPATIBLE_CHANGE;
+        }
+        return null;
+    }
+
+    private BreakingChanges requiredProperties(JsonNode toCheck, JsonNode toCheckAgainst) {
+        ArrayNode arrayNodeToCheck = (ArrayNode) toCheck.get("required");
+        if (arrayNodeToCheck != null) {
+            for (int i = 0; i < arrayNodeToCheck.size(); i++) {
+                if (!jsonCompatibilityCheckerUtils.isInRequired(arrayNodeToCheck.get(i).textValue(), toCheckAgainst)) {
+                    if (toCheck.get("properties").get(arrayNodeToCheck.get(i).textValue()).get("default") == null)
+                        return BreakingChanges.REQUIRED_PROPERTY_ADDED_WITHOUT_DEFAULT;
+                }
+            }
+        }
+        return null;
+    }
+}
diff --git a/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/StringComparator.java b/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/StringComparator.java
new file mode 100644
index 000000000..de514efd3
--- /dev/null
+++ b/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/StringComparator.java
@@ -0,0 +1,52 @@
+package io.pravega.schemaregistry.rules.jsoncompatibility;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+import static io.pravega.schemaregistry.rules.jsoncompatibility.BreakingChangesStore.*;
+
+public class StringComparator {
+    
+    public BreakingChanges stringComparator(JsonNode toCheck, JsonNode toCheckAgainst) {
+        if(minLengthComparator(toCheck, toCheckAgainst) != null)
+            return minLengthComparator(toCheck, toCheckAgainst);
+        if(maxLengthComparator(toCheck, toCheckAgainst) != null)
+            return maxLengthComparator(toCheck, toCheckAgainst);
+        if(patternComparator(toCheck, toCheckAgainst) != null)
+            return patternComparator(toCheck, toCheckAgainst);
+        return null;
+    }
+    
+    private BreakingChanges minLengthComparator(JsonNode toCheck, JsonNode toCheckAgainst) {
+        if(toCheck.get("minLength") != null && toCheckAgainst.get("minLength") == null)
+            return BreakingChanges.STRING_TYPE_MIN_LENGTH_ADDED;
+        else if(toCheck.get("minLength") != null && toCheckAgainst.get("minLength") != null) {
+            int toCheckMinLength = toCheck.get("minLength").asInt();
+            int toCheckAgainstMinLength = toCheckAgainst.get("minLength").asInt();
+            if(toCheckMinLength > toCheckAgainstMinLength)
+                return BreakingChanges.STRING_TYPE_MIN_LENGTH_VALUE_INCREASED;
+        }
+        return null;
+    }
+    
+    private BreakingChanges maxLengthComparator(JsonNode toCheck, JsonNode toCheckAgainst) {
+        if(toCheck.get("maxLength") != null && toCheckAgainst.get("maxLength") == null)
+            return BreakingChanges.STRING_TYPE_MAX_LENGTH_ADDED;
+        else if(toCheck.get("maxLength") != null && toCheckAgainst.get("maxLength") != null) {
+            int toCheckMaxLength = toCheck.get("maxLength").asInt();
+            int toCheckAgainstMaxLength = toCheckAgainst.get("maxLength").asInt();
+            if(toCheckMaxLength < toCheckAgainstMaxLength)
+                return BreakingChanges.STRING_TYPE_MAX_LENGTH_VALUE_DECREASED;
+        }
+        return null;
+    }
+    
+    private BreakingChanges patternComparator(JsonNode toCheck, JsonNode toCheckAgainst) {
+        if(toCheck.get("pattern") != null && toCheckAgainst.get("pattern") == null)
+            return BreakingChanges.STRING_TYPE_PATTERN_ADDED;
+        else if (toCheck.get("pattern") != null && toCheckAgainst.get("pattern") != null) {
+            if(!toCheck.get("pattern").asText().equals(toCheckAgainst.get("pattern").asText()))
+                return BreakingChanges.STRING_TYPE_PATTERN_MODIFIED;
+        }
+        return null;
+    }
+}
diff --git a/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/SubSchemaComparator.java b/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/SubSchemaComparator.java
new file mode 100644
index 000000000..57fd365ae
--- /dev/null
+++ b/server/src/main/java/io/pravega/schemaregistry/rules/jsoncompatibility/SubSchemaComparator.java
@@ -0,0 +1,114 @@
+package io.pravega.schemaregistry.rules.jsoncompatibility;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+
+import static io.pravega.schemaregistry.rules.jsoncompatibility.BreakingChangesStore.*;
+
+public class SubSchemaComparator {
+    JsonCompatibilityChecker jsonCompatibilityChecker;
+    
+    public void setJsonCompatibilityChecker() {
+        this.jsonCompatibilityChecker = new JsonCompatibilityChecker();
+    }
+    
+    public BreakingChanges checkSubSchemas(JsonNode toCheck, JsonNode toCheckAgainst) {
+        if(toCheck.has("anyOf")) {
+            if(toCheckAgainst.has("oneOf") || toCheckAgainst.has("allOf"))
+                return BreakingChanges.SUBSCHEMA_TYPE_CHANGED;
+            else {
+                if(toCheckAgainst.has("anyOf"))
+                    return nonAllOfComparator(toCheck, toCheckAgainst);
+                else
+                    return BreakingChanges.SUBSCHEMA_TYPE_ADDED;
+            }
+        }
+        else if(toCheck.has("oneOf")) {
+            if(toCheckAgainst.has("anyOf") || toCheckAgainst.has("allOf"))
+                return BreakingChanges.SUBSCHEMA_TYPE_CHANGED;
+            else {
+                if(toCheckAgainst.has("oneOf"))
+                    return nonAllOfComparator(toCheck, toCheckAgainst);
+                else
+                    return BreakingChanges.SUBSCHEMA_TYPE_ADDED;
+            }
+        }
+        else if(toCheck.has("allOf")) {
+            if(toCheckAgainst.has("anyOf") || toCheckAgainst.has("oneOf"))
+                return BreakingChanges.SUBSCHEMA_TYPE_CHANGED;
+            else {
+                if(toCheckAgainst.has("allOf"))
+                    return allOfComparator(toCheck, toCheckAgainst);
+                else
+                    return BreakingChanges.SUBSCHEMA_TYPE_ADDED;
+            }
+        }
+        return null;
+    }
+    
+    private BreakingChanges nonAllOfComparator(JsonNode toCheck, JsonNode toCheckAgainst) {
+        if(toCheck.has("oneOf")) {
+            ArrayNode toCheckArray = (ArrayNode) toCheck.get("oneOf");
+            ArrayNode toCheckAgainstArray = (ArrayNode) toCheckAgainst.get("oneOf");
+            if(toCheckArray.size() < toCheckAgainstArray.size())
+                return BreakingChanges.SUBSCHEMA_TYPE_ONE_OF_SCHEMAS_DECREASED;
+            else {
+                if (!nonAllOfCompatibilityComputation(toCheckArray, toCheckAgainstArray))
+                    return BreakingChanges.SUBSCHEMA_TYPE_ONE_OF_SCHEMAS_CHANGED;
+            }
+        }
+        else {
+            ArrayNode toCheckArray = (ArrayNode) toCheck.get("anyOf");
+            ArrayNode toCheckAgainstArray = (ArrayNode) toCheckAgainst.get("anyOf");
+            if(toCheckArray.size() < toCheckAgainstArray.size())
+                return BreakingChanges.SUBSCHEMA_TYPE_ANY_OF_SCHEMAS_DECREASED;
+            else {
+                if (!nonAllOfCompatibilityComputation(toCheckArray, toCheckAgainstArray))
+                    return BreakingChanges.SUBSCHEMA_TYPE_ANYOF_SCHEMAS_CHANGED;
+            }
+        }
+        return null;
+    }
+    
+    private BreakingChanges allOfComparator(JsonNode toCheck, JsonNode toCheckAgainst) {
+        ArrayNode toCheckArray = (ArrayNode) toCheck.get("allOf");
+        ArrayNode toCheckAgainstArray = (ArrayNode) toCheckAgainst.get("allOf");
+        if(!(toCheckArray.size() <= toCheckAgainstArray.size()))
+            return BreakingChanges.SUBSCHEMA_TYPE_ALL_OF_SCHEMAS_INCREASED;
+        else {
+            if (!allOfCompatibilityComputation(toCheckArray, toCheckAgainstArray))
+                return BreakingChanges.SUBSCHEMA_TYPE_ALL_OF_SCHEMAS_CHANGED;
+        }
+        return null;
+    }
+    
+    private boolean allOfCompatibilityComputation(ArrayNode toCheckArray, ArrayNode toCheckAgainstArray) {
+        for(int i=0;i<toCheckArray.size();i++) {
+            int flag = 0;
+            for(int j=0;j<toCheckAgainstArray.size();j++) {
+                if(jsonCompatibilityChecker.checkNodeType(toCheckArray.get(i), toCheckAgainstArray.get(j))==null) {
+                    flag=1;
+                    break;
+                }
+            }
+            if(flag==0)
+                return false;
+        }
+        return true;
+    }
+    
+    private boolean nonAllOfCompatibilityComputation(ArrayNode toCheckArray, ArrayNode toCheckAgainstArray) {
+        for(int i=0;i<toCheckAgainstArray.size();i++) {
+            int flag = 0;
+            for(int j=0;j<toCheckArray.size();j++) {
+                if(jsonCompatibilityChecker.checkNodeType(toCheckAgainstArray.get(i), toCheckArray.get(j))==null) {
+                    flag=1;
+                    break;
+                }
+            }
+            if(flag==0)
+                return false;
+        }
+        return true;
+    }
+}
diff --git a/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/ArrayTypeComparatorTest.java b/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/ArrayTypeComparatorTest.java
new file mode 100644
index 000000000..cd3d5df00
--- /dev/null
+++ b/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/ArrayTypeComparatorTest.java
@@ -0,0 +1,159 @@
+package io.pravega.schemaregistry.rules.jsoncompatibility;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+public class ArrayTypeComparatorTest {
+
+    @Test
+    public void testCheckUniqueItems() throws IOException {
+        ObjectMapper objectMapper = new ObjectMapper();
+        ArrayTypeComparator arrayTypeComparator = new ArrayTypeComparator();
+        arrayTypeComparator.setJsonTypeComparator();
+        String x1 = "{\n" +
+                "\"type\": \"array\" ,\n" +
+                "\"uniqueItems\": true\n" +
+                "}\n";
+        String x2 = "{\n" +
+                "\"type\": \"array\" ,\n" +
+                "\"uniqueItems\": false\n" +
+                "}\n";
+        JsonNode toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        JsonNode toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.ARRAY_UNIQUE_ITEMS_CONDITION_ENABLED,
+                arrayTypeComparator.compareArrays(toCheck, toCheckAgainst));
+        x2 = "{\n" +
+                "\"type\": \"array\"\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.ARRAY_UNIQUE_ITEMS_CONDITION_ENABLED,
+                arrayTypeComparator.compareArrays(toCheck, toCheckAgainst));
+    }
+
+    @Test
+    public void testMinMaxItems() throws IOException {
+        ObjectMapper objectMapper = new ObjectMapper();
+        ArrayTypeComparator arrayTypeComparator = new ArrayTypeComparator();
+        arrayTypeComparator.setJsonTypeComparator();
+        String x1 = "{\n" +
+                "\"type\": \"array\" ,\n" +
+                "\"maxItems\": 3\n" +
+                "}\n";
+        String x2 = "{\n" +
+                "\"type\": \"array\"\n" +
+                "}\n";
+        JsonNode toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        JsonNode toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.ARRAY_MAX_ITEMS_CONDITION_ADDED,
+                arrayTypeComparator.compareArrays(toCheck, toCheckAgainst));
+        x2 = "{\n" +
+                "\"type\": \"array\" ,\n" +
+                "\"maxItems\": 4\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.ARRAY_MAX_ITEMS_VALUE_DECREASED,
+                arrayTypeComparator.compareArrays(toCheck, toCheckAgainst));
+        x1 = "{\n" +
+                "\"type\": \"array\" ,\n" +
+                "\"minItems\": 3\n" +
+                "}\n";
+        x2 = "{\n" +
+                "\"type\": \"array\"\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.ARRAY_MIN_ITEMS_CONDITION_ADDED,
+                arrayTypeComparator.compareArrays(toCheck, toCheckAgainst));
+        x2 = "{\n" +
+                "\"type\": \"array\" ,\n" +
+                "\"minItems\": 2\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.ARRAY_MIN_ITEMS_VALUE_INCREASED,
+                arrayTypeComparator.compareArrays(toCheck, toCheckAgainst));
+    }
+
+    @Test
+    public void testAdditionalItems() throws IOException {
+        ObjectMapper objectMapper = new ObjectMapper();
+        ArrayTypeComparator arrayTypeComparator = new ArrayTypeComparator();
+        arrayTypeComparator.setJsonTypeComparator();
+        String x1 = "{\n" +
+                "\"type\": \"array\" ,\n" +
+                "\"additionalItems\": false\n" +
+                "}\n";
+        String x2 = "{\n" +
+                "\"type\": \"array\"\n" +
+                "}\n";
+        JsonNode toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        JsonNode toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.ARRAY_ADDITIONAL_ITEMS_DISABLED,
+                arrayTypeComparator.compareArrays(toCheck, toCheckAgainst));
+        x1 = "{\n" +
+                "\"type\": \"array\" ,\n" +
+                "\"additionalItems\": { \"type\": \"string\" }\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.ARRAY_ADDITIONAL_ITEMS_SCOPE_DECREASED,
+                arrayTypeComparator.compareArrays(toCheck, toCheckAgainst));
+        x2 = "{\n" +
+                "\"type\": \"array\" ,\n" +
+                "\"additionalItems\": true\n" +
+                "}\n";
+        x1 = "{\n" +
+                "\"type\": \"array\" ,\n" +
+                "\"additionalItems\": false\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.ARRAY_ADDITIONAL_ITEMS_DISABLED,
+                arrayTypeComparator.compareArrays(toCheck, toCheckAgainst));
+        x1 = "{\n" +
+                "\"type\": \"array\" ,\n" +
+                "\"additionalItems\": { \"type\": \"string\" }\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.ARRAY_ADDITIONAL_ITEMS_SCOPE_DECREASED,
+                arrayTypeComparator.compareArrays(toCheck, toCheckAgainst));
+        x2 = "{\n" +
+                "\"type\": \"array\" ,\n" +
+                "\"additionalItems\": { \"type\": \"number\" }\n" +
+                "}\n";
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.ARRAY_ADDITIONAL_ITEMS_SCOPE_INCOMPATIBLE_CHANGE,
+                arrayTypeComparator.compareArrays(toCheck, toCheckAgainst));
+    }
+
+    @Test
+    public void testItemValidation() throws IOException {
+        // node.get(items) is an object
+        ObjectMapper objectMapper = new ObjectMapper();
+        ArrayTypeComparator arrayTypeComparator = new ArrayTypeComparator();
+        arrayTypeComparator.setJsonTypeComparator();
+        String x1 = "{\n" +
+                "\"type\": \"array\" ,\n" +
+                "\"items\": {\n" +
+                "\"type\": \"number\"\n" +
+                "}\n" +
+                "}\n";
+        String x2 = "{\n" +
+                "\"type\": \"array\" ,\n" +
+                "\"items\": {\n" +
+                "\"type\": \"string\"\n" +
+                "}\n" +
+                "}\n";
+        JsonNode toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        JsonNode toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertNotNull(arrayTypeComparator.compareArrays(toCheck, toCheckAgainst));
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.DATA_TYPE_MISMATCH,
+                arrayTypeComparator.compareArrays(toCheck, toCheckAgainst));
+    }
+}
diff --git a/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/DependenciesComparatorTest.java b/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/DependenciesComparatorTest.java
new file mode 100644
index 000000000..2da562a62
--- /dev/null
+++ b/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/DependenciesComparatorTest.java
@@ -0,0 +1,134 @@
+package io.pravega.schemaregistry.rules.jsoncompatibility;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+public class DependenciesComparatorTest {
+
+    @Test
+    public void testCheckDependencies() throws IOException {
+        DependenciesComparator dependenciesComparator = new DependenciesComparator();
+        ObjectMapper objectMapper = new ObjectMapper();
+        String x1 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"name\": { \"type\": \"string\" },\n" +
+                "\"credit_card\": { \"type\": \"number\" },\n" +
+                "\"billing_address\": { \"type\": \"string\" }\n" +
+                "},\n" +
+                "\"required\": [\"name\"],\n" +
+                "\"dependencies\": {\n" +
+                "\"credit_card\": [\"billing_address\"],\n" +
+                "\"billing_address\": [\"credit_card\"]\n" +
+                "}\n" +
+                "}\n";
+        String x2 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"name\": { \"type\": \"string\" },\n" +
+                "\"credit_card\": { \"type\": \"number\" },\n" +
+                "\"billing_address\": { \"type\": \"string\" }\n" +
+                "},\n" +
+                "\"required\": [\"name\"],\n" +
+                "\"dependencies\": {\n" +
+                "\"credit_card\": [\"billing_address\"]\n" +
+                "}\n" +
+                "}\n";
+        JsonNode toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        JsonNode toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.DEPENDENCY_ADDED_IN_ARRAY_FORM,
+                dependenciesComparator.checkDependencies(toCheck, toCheckAgainst));
+        x1 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"name\": { \"type\": \"string\" },\n" +
+                "\"credit_card\": { \"type\": \"number\" },\n" +
+                "\"billing_address\": { \"type\": \"string\" }\n" +
+                "},\n" +
+                "\"required\": [\"name\"],\n" +
+                "\"dependencies\": {\n" +
+                "\"credit_card\": [\"billing_address\", \"name\"]\n" +
+                "}\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.DEPENDENCY_ARRAY_ELEMENTS_NON_REMOVAL,
+                dependenciesComparator.checkDependencies(toCheck, toCheckAgainst));
+        x2 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"}\n" +
+                "}\n" +
+                "}\n";
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.DEPENDENCY_SECTION_ADDED,
+                dependenciesComparator.checkDependencies(toCheck, toCheckAgainst));
+        x1 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"name\": { \"type\": \"string\" },\n" +
+                "\"credit_card\": { \"type\": \"number\" }\n" +
+                "},\n" +
+                "\"required\": [\"name\"],\n" +
+                "\"dependencies\": {\n" +
+                "\"credit_card\": {\n" +
+                "\"properties\": {\n" +
+                "\"billing_address\": { \"type\": \"string\" }\n" +
+                "},\n" +
+                "\"required\": [\"billing_address\"]\n" +
+                "}\n" +
+                "}\n" +
+                "}\n";
+        x2 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"name\": { \"type\": \"string\" },\n" +
+                "\"credit_card\": { \"type\": \"number\" }\n" +
+                "},\n" +
+                "\"required\": [\"name\"],\n" +
+                "\"dependencies\": {\n" +
+                "\"credit_card\": {\n" +
+                "\"properties\": {\n" +
+                "\"card_number\": { \"type\": \"number\" }\n" +
+                "},\n" +
+                "\"required\": [\"card_number\"]\n" +
+                "}\n" +
+                "}\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.DEPENDENCY_IN_SCHEMA_FORM_MODIFIED,
+                dependenciesComparator.checkDependencies(toCheck, toCheckAgainst));
+        x1 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"name\": { \"type\": \"string\" },\n" +
+                "\"credit_card\": { \"type\": \"number\" }\n" +
+                "},\n" +
+                "\"required\": [\"name\"],\n" +
+                "\"dependencies\": {\n" +
+                "\"credit_card\": {\n" +
+                "\"properties\": {\n" +
+                "\"card_number\": { \"type\": \"number\" }\n" +
+                "},\n" +
+                "\"required\": [\"card_number\"]\n" +
+                "},\n" +
+                "\"name\": {\n" +
+                "\"properties\": {\n" +
+                "\"salutation\": { \"type\": \"string\" }\n" +
+                "},\n" +
+                "\"required\": [\"salutation\"]\n" +
+                "}\n" +
+                "}\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.DEPENDENCY_ADDED_IN_SCHEMA_FORM,
+                dependenciesComparator.checkDependencies(toCheck, toCheckAgainst));
+    }
+}
diff --git a/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/EnumComparatorTest.java b/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/EnumComparatorTest.java
new file mode 100644
index 000000000..53b1bba37
--- /dev/null
+++ b/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/EnumComparatorTest.java
@@ -0,0 +1,36 @@
+package io.pravega.schemaregistry.rules.jsoncompatibility;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+public class EnumComparatorTest {
+
+    @Test
+    public void testEnumComparator() throws IOException {
+        EnumComparator enumComparator = new EnumComparator();
+        ObjectMapper objectMapper = new ObjectMapper();
+        String x1 = "{\n" +
+                "\"type\": \"string\" ,\n" +
+                "\"enum\": [\"red\", \"amber\", \"green\"]\n" +
+                "}\n";
+        String x2 = "{\n" +
+                "\"type\": \"string\"\n" +
+                "}\n";
+        JsonNode toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        JsonNode toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.ENUM_TYPE_ADDED,
+                enumComparator.enumComparator(toCheck, toCheckAgainst));
+        x2 = "{\n" +
+                "\"type\": \"string\" ,\n" +
+                "\"enum\": [\"red\", \"amber\", \"yellow\"]\n" +
+                "}\n";
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.ENUM_TYPE_ARRAY_CONTENTS_NON_ADDITION_OF_ELEMENTS,
+                enumComparator.enumComparator(toCheck, toCheckAgainst));
+    }
+}
diff --git a/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/JsonCompatibilityCheckerTest.java b/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/JsonCompatibilityCheckerTest.java
new file mode 100644
index 000000000..c16db4eb9
--- /dev/null
+++ b/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/JsonCompatibilityCheckerTest.java
@@ -0,0 +1,271 @@
+package io.pravega.schemaregistry.rules.jsoncompatibility;
+
+import com.google.common.collect.ImmutableMap;
+import io.pravega.schemaregistry.contract.data.SchemaInfo;
+import io.pravega.schemaregistry.contract.data.SerializationFormat;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+public class JsonCompatibilityCheckerTest {
+
+    @Test
+    public void testEqualCase() {
+        JsonCompatibilityChecker jsonCompatibilityChecker = new JsonCompatibilityChecker();
+        String x = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"name\": { \"type\": \"string\" },\n" +
+                "\"credit_card\": { \"type\": \"number\" }\n" +
+                "},\n" +
+                "\"required\": [\"name\"],\n" +
+                "\"dependencies\": {\n" +
+                "\"credit_card\": {\n" +
+                "\"properties\": {\n" +
+                "\"billing_address\": { \"type\": \"string\" }\n" +
+                "},\n" +
+                "\"required\": [\"billing_address\"]\n" +
+                "}\n" +
+                "}\n" +
+                "}\n";
+        SchemaInfo toValidate = new SchemaInfo("toValidate", SerializationFormat.Json, ByteBuffer.wrap(x.getBytes()),
+                ImmutableMap.of());
+        List<SchemaInfo> toValidateAgainst = new ArrayList<>();
+        SchemaInfo schemaInfo1 = new SchemaInfo("toValidate", SerializationFormat.Json, ByteBuffer.wrap(x.getBytes()),
+                ImmutableMap.of());
+        toValidateAgainst.add(schemaInfo1);
+        Assert.assertTrue(jsonCompatibilityChecker.canRead(toValidate, toValidateAgainst));
+        //CanBeRead
+        Assert.assertTrue(jsonCompatibilityChecker.canBeRead(toValidate, toValidateAgainst));
+    }
+
+    @Test
+    public void testDependencies() {
+        JsonCompatibilityChecker jsonCompatibilityChecker = new JsonCompatibilityChecker();
+        String x = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"name\": { \"type\": \"string\" },\n" +
+                "\"credit_card\": { \"type\": \"number\" }\n" +
+                "},\n" +
+                "\"required\": [\"name\"],\n" +
+                "\"dependencies\": {\n" +
+                "\"credit_card\": {\n" +
+                "\"properties\": {\n" +
+                "\"billing_address\": { \"type\": \"string\" }\n" +
+                "},\n" +
+                "\"required\": [\"billing_address\"]\n" +
+                "}\n" +
+                "}\n" +
+                "}\n";
+        String y = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"name\": { \"type\": \"string\" },\n" +
+                "\"credit_card\": { \"type\": \"number\" }\n" +
+                "},\n" +
+                "\"required\": [\"name\"],\n" +
+                "\"dependencies\": {\n" +
+                "\"credit_card\": {\n" +
+                "\"properties\": {\n" +
+                "\"billing_address\": { \"type\": \"string\" }\n" +
+                "},\n" +
+                "\"required\": [\"billing_address\"]\n" +
+                "},\n" +
+                "\"name\": {\n" +
+                "\"properties\": {\n" +
+                "\"salutation\": { \"type\": \"string\" }\n" +
+                "},\n" +
+                "\"required\": [\"salutation\"]\n" +
+                "}\n" +
+                "}\n" +
+                "}\n";
+        SchemaInfo toValidate = new SchemaInfo("toValidate", SerializationFormat.Json, ByteBuffer.wrap(x.getBytes()),
+                ImmutableMap.of());
+        SchemaInfo schemaInfo1 = new SchemaInfo("toValidate", SerializationFormat.Json, ByteBuffer.wrap(y.getBytes()),
+                ImmutableMap.of());
+        List<SchemaInfo> toValidateAgainst = new ArrayList<>();
+        toValidateAgainst.add(toValidate);
+        Assert.assertTrue(jsonCompatibilityChecker.canRead(toValidate, toValidateAgainst));
+        // canBeRead
+        Assert.assertTrue(jsonCompatibilityChecker.canBeRead(toValidate, toValidateAgainst));
+        toValidateAgainst.add(schemaInfo1);
+        Assert.assertTrue(jsonCompatibilityChecker.canRead(toValidate, toValidateAgainst));
+        Assert.assertFalse(jsonCompatibilityChecker.canBeRead(toValidate, toValidateAgainst));
+        String z = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"name\": { \"type\": \"string\" },\n" +
+                "\"credit_card\": { \"type\": \"number\" }\n" +
+                "},\n" +
+                "\"required\": [\"name\"],\n" +
+                "\"dependencies\": {\n" +
+                "\"credit_card\": {\n" +
+                "\"properties\": {\n" +
+                "\"billing_address\": { \"type\": \"number\" }\n" +
+                "},\n" +
+                "\"required\": [\"billing_address\"]\n" +
+                "}\n" +
+                "}\n" +
+                "}\n";
+        SchemaInfo schemaInfo11 = new SchemaInfo("toValidateAgainst", SerializationFormat.Json,
+                ByteBuffer.wrap(z.getBytes()), ImmutableMap.of());
+        toValidateAgainst.add(schemaInfo11);
+        Assert.assertFalse(jsonCompatibilityChecker.canBeRead(toValidate, toValidateAgainst));
+        Assert.assertFalse(jsonCompatibilityChecker.canBeRead(toValidate, toValidateAgainst));
+    }
+
+    @Test
+    public void testBasicProperties() throws IOException {
+        JsonCompatibilityChecker jsonCompatibilityChecker = new JsonCompatibilityChecker();
+        String x1 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"}\n" +
+                "}\n" +
+                "}\n";
+        String x2 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"}\n" +
+                "}\n" +
+                "}\n";
+
+        SchemaInfo toValidate = new SchemaInfo("toValidate", SerializationFormat.Json, ByteBuffer.wrap(x1.getBytes()),
+                ImmutableMap.of());
+        SchemaInfo toValidateAgainst = new SchemaInfo("toValidateAgainst", SerializationFormat.Json,
+                ByteBuffer.wrap(x2.getBytes()), ImmutableMap.of());
+        List<SchemaInfo> toValidateAgainstList = new ArrayList<>();
+        toValidateAgainstList.add(toValidateAgainst);
+        Assert.assertTrue(jsonCompatibilityChecker.canRead(toValidate, toValidateAgainstList));
+        Assert.assertTrue(jsonCompatibilityChecker.canBeRead(toValidate, toValidateAgainstList));
+        //different properties
+        x2 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name1\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"}\n" +
+                "}\n" +
+                "}\n";
+        SchemaInfo toValidateAgainst1 = new SchemaInfo("toValidateAgainst", SerializationFormat.Json,
+                ByteBuffer.wrap(x2.getBytes()), ImmutableMap.of());
+        toValidateAgainstList.add(toValidateAgainst1);
+        Assert.assertFalse(jsonCompatibilityChecker.canRead(toValidate, toValidateAgainstList));
+        Assert.assertFalse(jsonCompatibilityChecker.canBeRead(toValidate, toValidateAgainstList));
+        //different property values 
+        x2 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"number\" },\n" +
+                "\"street_type\": { \"type\": \"string\"}\n" +
+                "}\n" +
+                "}\n";
+        SchemaInfo toValidateAgainst2 = new SchemaInfo("toValidateAgainst", SerializationFormat.Json,
+                ByteBuffer.wrap(x2.getBytes()), ImmutableMap.of());
+        toValidateAgainstList.add(toValidateAgainst2);
+        Assert.assertFalse(jsonCompatibilityChecker.canRead(toValidate, toValidateAgainstList));
+        Assert.assertFalse(jsonCompatibilityChecker.canBeRead(toValidate, toValidateAgainstList));
+    }
+
+    @Test
+    public void testRequired() throws IOException {
+        JsonCompatibilityChecker jsonCompatibilityChecker = new JsonCompatibilityChecker();
+        // equal test
+        String x1 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"name\": { \"type\": \"string\" },\n" +
+                "\"email\": { \"type\": \"string\" },\n" +
+                "\"address\": { \"type\": \"string\" },\n" +
+                "\"telephone\": { \"type\": \"string\" }\n" +
+                "},\n" +
+                "\"required\": [\"name\", \"email\"]\n" +
+                "}\n";
+        SchemaInfo toValidate = new SchemaInfo("toValidate", SerializationFormat.Json, ByteBuffer.wrap(x1.getBytes()),
+                ImmutableMap.of());
+        SchemaInfo toValidateAgainst = new SchemaInfo("toValidateAgainst", SerializationFormat.Json,
+                ByteBuffer.wrap(x1.getBytes()), ImmutableMap.of());
+        List<SchemaInfo> toValidateAgainstList = new ArrayList<>();
+        toValidateAgainstList.add(toValidateAgainst);
+        Assert.assertTrue(jsonCompatibilityChecker.canRead(toValidate, toValidateAgainstList));
+        Assert.assertTrue(jsonCompatibilityChecker.canBeRead(toValidate, toValidateAgainstList));
+        //remove required array element
+        String x2 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"name\": { \"type\": \"string\" },\n" +
+                "\"email\": { \"type\": \"string\" },\n" +
+                "\"address\": { \"type\": \"string\" },\n" +
+                "\"telephone\": { \"type\": \"string\" }\n" +
+                "},\n" +
+                "\"required\": [\"name\", \"email\", \"address\"]\n" +
+                "}\n";
+        SchemaInfo toValidateAgainst1 = new SchemaInfo("toValidateAgainst", SerializationFormat.Json,
+                ByteBuffer.wrap(x2.getBytes()), ImmutableMap.of());
+        toValidateAgainstList.add(toValidateAgainst1);
+        Assert.assertTrue(jsonCompatibilityChecker.canRead(toValidate, toValidateAgainstList));
+        Assert.assertFalse(jsonCompatibilityChecker.canBeRead(toValidate, toValidateAgainstList));
+        x2 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"name\": { \"type\": \"string\" },\n" +
+                "\"email\": { \"type\": \"string\" },\n" +
+                "\"address\": { \"type\": \"string\" },\n" +
+                "\"telephone\": { \"type\": \"string\" }\n" +
+                "},\n" +
+                "\"required\": [\"name\"]\n" +
+                "}\n";
+        SchemaInfo toValidateAgainst2 = new SchemaInfo("toValidateAgainst", SerializationFormat.Json,
+                ByteBuffer.wrap(x2.getBytes()), ImmutableMap.of());
+        toValidateAgainstList.clear();
+        toValidateAgainstList.add(toValidateAgainst2);
+        Assert.assertFalse(jsonCompatibilityChecker.canRead(toValidate, toValidateAgainstList));
+        Assert.assertTrue(jsonCompatibilityChecker.canBeRead(toValidate, toValidateAgainstList));
+    }
+    
+    @Test
+    public void testDynamicProperties() {
+        JsonCompatibilityChecker jsonCompatibilityChecker = new JsonCompatibilityChecker();
+        String x1 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"name\": { \"type\": \"string\" },\n" +
+                "\"email\": { \"type\": \"string\" },\n" +
+                "\"address\": { \"type\": \"string\" },\n" +
+                "\"telephone\": { \"type\": \"string\" }\n" +
+                "},\n" +
+                "\"additionalProperties\": { \"type\": \"string\" }\n" +
+                "}\n";
+        String x2 =  "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"name\": { \"type\": \"string\" },\n" +
+                "\"email\": { \"type\": \"string\" },\n" +
+                "\"address\": { \"type\": \"string\" },\n" +
+                "\"telephone\": { \"type\": \"string\" },\n" +
+                "\"SSN\": { \"type\": \"number\" }\n" +
+                "},\n" +
+                "\"additionalProperties\": { \"type\": \"string\" }\n" +
+                "}\n";
+        SchemaInfo toValidate = new SchemaInfo("toValidate", SerializationFormat.Json, ByteBuffer.wrap(x1.getBytes()),
+                ImmutableMap.of());
+        SchemaInfo toValidateAgainst = new SchemaInfo("toValidateAgainst", SerializationFormat.Json,
+                ByteBuffer.wrap(x2.getBytes()), ImmutableMap.of());
+        List<SchemaInfo> toValidateAgainstList = new ArrayList<>();
+        toValidateAgainstList.add(toValidateAgainst);
+        Assert.assertFalse(jsonCompatibilityChecker.canRead(toValidate, toValidateAgainstList));
+        Assert.assertFalse(jsonCompatibilityChecker.canBeRead(toValidate, toValidateAgainstList));
+    }
+}
+	
\ No newline at end of file
diff --git a/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/JsonCompatibilityCheckerUtilsTest.java b/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/JsonCompatibilityCheckerUtilsTest.java
new file mode 100644
index 000000000..0f0cd579b
--- /dev/null
+++ b/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/JsonCompatibilityCheckerUtilsTest.java
@@ -0,0 +1,167 @@
+package io.pravega.schemaregistry.rules.jsoncompatibility;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+public class JsonCompatibilityCheckerUtilsTest {
+    ObjectMapper objectMapper = new ObjectMapper();
+    JsonCompatibilityCheckerUtils jsonCompatibilityCheckerUtils = new JsonCompatibilityCheckerUtils();
+    @Test
+    public void testGetTypeValue() throws IOException {
+        String x = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"name\": { \"type\": \"string\" },\n" +
+                "\"credit_card\": { \"type\": \"number\" },\n" +
+                "\"billing_address\": { \"type\": \"string\" }\n" +
+                "},\n" +
+                "\"required\": [\"name\"],\n" +
+                "\"dependencies\": {\n" +
+                "\"credit_card\": [\"billing_address\"]\n" +
+                "}\n" +
+                "}\n";
+        JsonNode node = objectMapper.readTree(ByteBuffer.wrap(x.getBytes()).array());
+        Assert.assertEquals("object", jsonCompatibilityCheckerUtils.getTypeValue(node));
+    }
+    
+    @Test
+    public void testHasDynamicPropertySet() throws IOException {
+        String x = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"name\": { \"type\": \"string\" },\n" +
+                "\"credit_card\": { \"type\": \"number\" },\n" +
+                "\"billing_address\": { \"type\": \"string\" }\n" +
+                "},\n" +
+                "\"required\": [\"name\"],\n" +
+                "\"dependencies\": {\n" +
+                "\"credit_card\": [\"billing_address\"]\n" +
+                "}\n" +
+                "}\n";
+        JsonNode node = objectMapper.readTree(ByteBuffer.wrap(x.getBytes()).array());
+        Assert.assertTrue(jsonCompatibilityCheckerUtils.hasDynamicPropertySet(node));
+        x = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"name\": { \"type\": \"string\" },\n" +
+                "\"credit_card\": { \"type\": \"number\" },\n" +
+                "\"billing_address\": { \"type\": \"string\" }\n" +
+                "},\n" +
+                "\"required\": [\"name\"],\n" +
+                "\"additionalProperties\" : true,\n" +
+                "\"dependencies\": {\n" +
+                "\"credit_card\": [\"billing_address\"]\n" +
+                "}\n" +
+                "}\n";
+        node = objectMapper.readTree(ByteBuffer.wrap(x.getBytes()).array());
+        Assert.assertTrue(jsonCompatibilityCheckerUtils.hasDynamicPropertySet(node));
+    }
+    
+    @Test
+    public void testHasStaticPropertySet() throws IOException {
+        String x = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"name\": { \"type\": \"string\" },\n" +
+                "\"credit_card\": { \"type\": \"number\" },\n" +
+                "\"billing_address\": { \"type\": \"string\" }\n" +
+                "},\n" +
+                "\"required\": [\"name\"],\n" +
+                "\"additionalProperties\" : false,\n" +
+                "\"dependencies\": {\n" +
+                "\"credit_card\": [\"billing_address\"]\n" +
+                "}\n" +
+                "}\n";
+        JsonNode node = objectMapper.readTree(ByteBuffer.wrap(x.getBytes()).array());
+        Assert.assertFalse(jsonCompatibilityCheckerUtils.hasDynamicPropertySet(node));
+    }
+    
+    @Test
+    public void testIsInRequired() throws IOException {
+        String toSearch = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"name\": { \"type\": \"string\" },\n" +
+                "\"credit_card\": { \"type\": \"number\" },\n" +
+                "\"billing_address\": { \"type\": \"string\" }\n" +
+                "},\n" +
+                "\"required\": [\"name\", \"credit_card\"],\n" +
+                "\"additionalProperties\" : false,\n" +
+                "\"dependencies\": {\n" +
+                "\"credit_card\": [\"billing_address\"]\n" +
+                "}\n" +
+                "}\n";
+        JsonNode node = objectMapper.readTree(ByteBuffer.wrap(toSearch.getBytes()).array());
+        Assert.assertTrue(jsonCompatibilityCheckerUtils.isInRequired("name", node));
+    }
+    
+    @Test
+    public void testArrayComparisionOnlyRemoval() throws IOException {
+        String arrayFinal = "[\"item1\", \"item2\"]";
+        String arrayOriginal = "[\"item1\", \"item2\", \"item3\"]";
+        JsonNode finalNode = objectMapper.readTree(ByteBuffer.wrap(arrayFinal.getBytes()).array());
+        ArrayNode finalArray =(ArrayNode) finalNode;
+        JsonNode originalNode = objectMapper.readTree(ByteBuffer.wrap(arrayOriginal.getBytes()).array());
+        ArrayNode originalArray =(ArrayNode) originalNode;
+        Assert.assertTrue(jsonCompatibilityCheckerUtils.arrayComparisionOnlyRemoval(finalArray, originalArray));
+    }
+    
+    @Test
+    public void testArrayComparisionOnlyAddition() throws IOException {
+        String arrayOriginal = "[\"item1\", \"item2\"]";
+        String arrayFinal = "[\"item1\", \"item2\", \"item3\"]";
+        JsonNode finalNode = objectMapper.readTree(ByteBuffer.wrap(arrayFinal.getBytes()).array());
+        ArrayNode finalArray =(ArrayNode) finalNode;
+        JsonNode originalNode = objectMapper.readTree(ByteBuffer.wrap(arrayOriginal.getBytes()).array());
+        ArrayNode originalArray =(ArrayNode) originalNode;
+        Assert.assertTrue(jsonCompatibilityCheckerUtils.arrayComparisionOnlyAddition(finalArray, originalArray));
+    }
+    
+    @Test
+    public void testHasSubSchema() throws IOException {
+        String x = "{\n" +
+                "\"anyOf\": [\n" +
+                "{ \"type\": \"string\" },\n" +
+                "{ \"type\": \"number\" }\n" +
+                "]\n" +
+                "}\n";
+        JsonNode node  = objectMapper.readTree(ByteBuffer.wrap(x.getBytes()).array());
+        Assert.assertTrue(jsonCompatibilityCheckerUtils.hasSubSchema(node));
+        x = "{\n" +
+                "\"oneOf\": [\n" +
+                "{ \"type\": \"string\" },\n" +
+                "{ \"type\": \"number\" }\n" +
+                "]\n" +
+                "}\n";
+        node  = objectMapper.readTree(ByteBuffer.wrap(x.getBytes()).array());
+        Assert.assertTrue(jsonCompatibilityCheckerUtils.hasSubSchema(node));
+        x = "{\n" +
+                "\"allOf\": [\n" +
+                "{ \"type\": \"string\" },\n" +
+                "{ \"type\": \"number\" }\n" +
+                "]\n" +
+                "}\n";
+        node  = objectMapper.readTree(ByteBuffer.wrap(x.getBytes()).array());
+        Assert.assertTrue(jsonCompatibilityCheckerUtils.hasSubSchema(node));
+        x = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"name\": { \"type\": \"string\" },\n" +
+                "\"credit_card\": { \"type\": \"number\" },\n" +
+                "\"billing_address\": { \"type\": \"string\" }\n" +
+                "},\n" +
+                "\"required\": [\"name\"],\n" +
+                "\"dependencies\": {\n" +
+                "\"credit_card\": [\"billing_address\"]\n" +
+                "}\n" +
+                "}\n";
+        node  = objectMapper.readTree(ByteBuffer.wrap(x.getBytes()).array());
+        Assert.assertFalse(jsonCompatibilityCheckerUtils.hasSubSchema(node));
+    }
+}
diff --git a/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/NumberComparatorTest.java b/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/NumberComparatorTest.java
new file mode 100644
index 000000000..7259cee3f
--- /dev/null
+++ b/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/NumberComparatorTest.java
@@ -0,0 +1,171 @@
+package io.pravega.schemaregistry.rules.jsoncompatibility;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+public class NumberComparatorTest {
+
+    @Test
+    public void testMaximumComparator() throws IOException {
+        NumberComparator numberComparator = new NumberComparator();
+        ObjectMapper objectMapper = new ObjectMapper();
+        String x2 = "{\n" +
+                "\"type\": \"number\"\n" +
+                "}\n";
+        String x1 = "{\n" +
+                "\"type\": \"number\" ,\n" +
+                "\"maximum\": 3\n" +
+                "}\n";
+        JsonNode toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        JsonNode toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.NUMBER_TYPE_MAXIMUM_VALUE_ADDED,
+                numberComparator.compareNumbers(toCheck, toCheckAgainst));
+        x1 = "{\n" +
+                "\"type\": \"number\" ,\n" +
+                "\"exclusiveMaximum\": 3\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.NUMBER_TYPE_EXCLUSIVE_MAXIMUM_VALUE_ADDED,
+                numberComparator.compareNumbers(toCheck, toCheckAgainst));
+        x2 = "{\n" +
+                "\"type\": \"number\" ,\n" +
+                "\"exclusiveMaximum\": 4\n" +
+                "}\n";
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.NUMBER_TYPE_EXCLUSIVE_MAXIMUM_VALUE_DECREASED,
+                numberComparator.compareNumbers(toCheck, toCheckAgainst));
+        x1 = "{\n" +
+                "\"type\": \"number\" ,\n" +
+                "\"maximum\": 3\n" +
+                "}\n";
+        x2 = "{\n" +
+                "\"type\": \"number\" ,\n" +
+                "\"maximum\": 4\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.NUMBER_TYPE_MAXIMUM_VALUE_DECREASED,
+                numberComparator.compareNumbers(toCheck, toCheckAgainst));
+        x1 = "{\n" +
+                "\"type\": \"number\" ,\n" +
+                "\"exclusiveMaximum\": true\n" +
+                "}\n";
+        x2 = "{\n" +
+                "\"type\": \"number\" ,\n" +
+                "\"exclusiveMaximum\": false\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.NUMBER_TYPE_EXCLUSIVE_MAXIMUM_VALUE_ADDED,
+                numberComparator.compareNumbers(toCheck, toCheckAgainst));
+    }
+    
+    @Test
+    public void testMinimumComparator() throws IOException {
+        NumberComparator numberComparator = new NumberComparator();
+        ObjectMapper objectMapper = new ObjectMapper();
+        String x2 = "{\n" +
+                "\"type\": \"number\"\n" +
+                "}\n";
+        String x1 = "{\n" +
+                "\"type\": \"number\" ,\n" +
+                "\"minimum\": 3\n" +
+                "}\n";
+        JsonNode toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        JsonNode toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.NUMBER_TYPE_MINIMUM_VALUE_ADDED,
+                numberComparator.compareNumbers(toCheck, toCheckAgainst));
+        x1 = "{\n" +
+                "\"type\": \"number\" ,\n" +
+                "\"exclusiveMinimum\": 3\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.NUMBER_TYPE_EXCLUSIVE_MINIMUM_VALUE_ADDED,
+                numberComparator.compareNumbers(toCheck, toCheckAgainst));
+        x2 = "{\n" +
+                "\"type\": \"number\" ,\n" +
+                "\"exclusiveMinimum\": 2\n" +
+                "}\n";
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.NUMBER_TYPE_EXCLUSIVE_MINIMUM_VALUE_INCREASED,
+                numberComparator.compareNumbers(toCheck, toCheckAgainst));
+        x1 = "{\n" +
+                "\"type\": \"number\" ,\n" +
+                "\"minimum\": 3\n" +
+                "}\n";
+        x2 = "{\n" +
+                "\"type\": \"number\" ,\n" +
+                "\"minimum\": 2\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.NUMBER_TYPE_MINIMUM_VALUE_INCREASED,
+                numberComparator.compareNumbers(toCheck, toCheckAgainst));
+        x1 = "{\n" +
+                "\"type\": \"number\" ,\n" +
+                "\"exclusiveMinimum\": true\n" +
+                "}\n";
+        x2 = "{\n" +
+                "\"type\": \"number\" ,\n" +
+                "\"exclusiveMinimum\": false\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.NUMBER_TYPE_EXCLUSIVE_MINIMUM_VALUE_ADDED,
+                numberComparator.compareNumbers(toCheck, toCheckAgainst));
+    }
+    
+    @Test
+    public void testMultipleOfComparator() throws IOException {
+        NumberComparator numberComparator = new NumberComparator();
+        ObjectMapper objectMapper = new ObjectMapper();
+        String x2 = "{\n" +
+                "\"type\": \"number\"\n" +
+                "}\n";
+        String x1 = "{\n" +
+                "\"type\": \"number\" ,\n" +
+                "\"multipleOf\": 10\n" +
+                "}\n";
+        JsonNode toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        JsonNode toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.NUMBER_TYPE_MULTIPLE_OF_ADDED,
+                numberComparator.compareNumbers(toCheck, toCheckAgainst));
+        x2 = "{\n" +
+                "\"type\": \"number\" ,\n" +
+                "\"multipleOf\": 5\n" +
+                "}\n";
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.NUMBER_TYPE_MULTIPLE_OF_INCREASED,
+                numberComparator.compareNumbers(toCheck, toCheckAgainst));
+        x2 = "{\n" +
+                "\"type\": \"number\" ,\n" +
+                "\"multipleOf\": 3\n" +
+                "}\n";
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.NUMBER_TYPE_MULTIPLE_OF_NON_DIVISIBLE_CHANGE,
+                numberComparator.compareNumbers(toCheck, toCheckAgainst));
+    }
+    
+    @Test
+    public void testTypeChanged() throws IOException {
+        NumberComparator numberComparator = new NumberComparator();
+        ObjectMapper objectMapper = new ObjectMapper();
+        String x2 = "{\n" +
+                "\"type\": \"number\"\n" +
+                "}\n";
+        String x1 = "{\n" +
+                "\"type\": \"integer\"\n" +
+                "}\n";
+        JsonNode toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        JsonNode toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.NUMBER_TYPE_CHANGED_FROM_NUMBER_TO_INTEGER,
+                numberComparator.compareNumbers(toCheck, toCheckAgainst));
+    }
+    
+    
+}
diff --git a/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/PropertiesComparatorTest.java b/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/PropertiesComparatorTest.java
new file mode 100644
index 000000000..57a29ff3a
--- /dev/null
+++ b/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/PropertiesComparatorTest.java
@@ -0,0 +1,327 @@
+package io.pravega.schemaregistry.rules.jsoncompatibility;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+public class PropertiesComparatorTest {
+    @Test
+    public void testBasicProperties() throws IOException {
+        PropertiesComparator propertiesComparator = new PropertiesComparator();
+        propertiesComparator.setJsonCompatibilityChecker();
+        ObjectMapper objectMapper = new ObjectMapper();
+        String x1 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"}\n" +
+                "}\n" +
+                "}\n";
+        String x2 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"},\n" +
+                "\"city\": { \"type\": \"string\"}\n" +
+                "}\n" +
+                "}\n";
+        JsonNode toCheck = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        JsonNode toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.PROPERTY_ADDED_TO_DYNAMIC_PROPERTY_SET,
+                propertiesComparator.checkProperties(toCheck, toCheckAgainst));
+        x1 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"}\n" +
+                "},\n" +
+                "\"additionalProperties\": false\n" +
+                "}\n";
+        x2 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"},\n" +
+                "\"city\": { \"type\": \"string\"}\n" +
+                "},\n" +
+                "\"required\": [\"city\"]\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.REQUIRED_PROPERTY_ADDED_WITHOUT_DEFAULT,
+                propertiesComparator.checkProperties(toCheck, toCheckAgainst));
+        x1 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"}\n" +
+                "},\n" +
+                "\"additionalProperties\": {\"type\": \"number\"}\n" +
+                "}\n";
+        x2 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"},\n" +
+                "\"city\": { \"type\": \"string\"}\n" +
+                "},\n" +
+                "\"additionalProperties\": {\"type\": \"number\"}\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.PROPERTY_ADDED_NOT_PART_OF_DYNAMIC_PROPERTY_SET_WITH_CONDITION,
+                propertiesComparator.checkProperties(toCheck, toCheckAgainst));
+        x2 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"}\n" +
+                "},\n" +
+                "\"additionalProperties\": false\n" +
+                "}\n";
+        x1 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"},\n" +
+                "\"city\": { \"type\": \"string\"}\n" +
+                "}\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.PROPERTY_REMOVED_FROM_STATIC_PROPERTY_SET,
+                propertiesComparator.checkProperties(toCheck, toCheckAgainst));
+        x1 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"},\n" +
+                "\"city\": { \"type\": \"string\"}\n" +
+                "},\n" +
+                "\"additionalProperties\": {\"type\": \"number\"}\n" +
+                "}\n";
+        x2 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"}\n" +
+                "},\n" +
+                "\"additionalProperties\": {\"type\": \"number\"}\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.PROPERTY_REMOVED_NOT_PART_OF_DYNAMIC_PROPERTY_SET_WITH_CONDITION,
+                propertiesComparator.checkProperties(toCheck, toCheckAgainst));
+    }
+    
+    @Test
+    public void testMinMaxProperties() throws IOException {
+        ObjectMapper objectMapper = new ObjectMapper();
+        PropertiesComparator propertiesComparator = new PropertiesComparator();
+        propertiesComparator.setJsonCompatibilityChecker();
+        String x2 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"}\n" +
+                "},\n" +
+                "\"minProperties\": 1\n" +
+                "}\n";
+        String x1 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"}\n" +
+                "}\n" +
+                "}\n";
+        JsonNode toCheck = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        JsonNode toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.MIN_PROPERTIES_ADDED,
+                propertiesComparator.checkProperties(toCheck, toCheckAgainst));
+        x1 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"}\n" +
+                "},\n" +
+                "\"minProperties\": 1\n" +
+                "}\n";
+        x2 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"}\n" +
+                "},\n" +
+                "\"minProperties\": 2\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.MIN_PROPERTIES_LIMIT_INCREASED,
+                propertiesComparator.checkProperties(toCheck, toCheckAgainst));
+        x2 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"}\n" +
+                "},\n" +
+                "\"maxProperties\": 3\n" +
+                "}\n";
+        x1 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"}\n" +
+                "}\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.MAX_PROPERTIES_ADDED,
+                propertiesComparator.checkProperties(toCheck, toCheckAgainst));
+        x1 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"}\n" +
+                "},\n" +
+                "\"maxProperties\": 4\n" +
+                "}\n";
+        x2 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"}\n" +
+                "},\n" +
+                "\"maxProperties\": 3\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.MAX_PROPERTIES_LIMIT_DECREASED,
+                propertiesComparator.checkProperties(toCheck, toCheckAgainst));
+    }
+    
+    @Test
+    public void testAdditionalProperties() throws IOException {
+        ObjectMapper objectMapper = new ObjectMapper();
+        PropertiesComparator propertiesComparator = new PropertiesComparator();
+        propertiesComparator.setJsonCompatibilityChecker();
+        String x1 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"}\n" +
+                "},\n" +
+                "\"additionalProperties\": false\n" +
+                "}\n";
+        String x2 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"}\n" +
+                "}\n" +
+                "}\n";
+        JsonNode toCheck = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        JsonNode toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.ADDITIONAL_PROPERTIES_REMOVED,
+                propertiesComparator.checkProperties(toCheck, toCheckAgainst));
+        x1 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"}\n" +
+                "},\n" +
+                "\"additionalProperties\": {\"type\": \"string\"}\n" +
+                "}\n";
+        x2 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"}\n" +
+                "},\n" +
+                "\"additionalProperties\": {\"type\": \"number\"}\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.ADDITIONAL_PROPERTIES_NON_COMPATIBLE_CHANGE,
+                propertiesComparator.checkProperties(toCheck, toCheckAgainst));
+    }
+    
+    @Test
+    public void testRequiredProperties() throws IOException {
+        ObjectMapper objectMapper = new ObjectMapper();
+        PropertiesComparator propertiesComparator = new PropertiesComparator();
+        propertiesComparator.setJsonCompatibilityChecker();
+        String x1 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"},\n" +
+                "\"city\": { \"type\": \"string\"}\n" +
+                "},\n" +
+                "\"required\": [\"city\"]\n" +
+                "}\n";
+        String x2 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"},\n" +
+                "\"city\": { \"type\": \"string\"}\n" +
+                "}\n" +
+                "}\n";
+        JsonNode toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        JsonNode toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.REQUIRED_PROPERTY_ADDED_WITHOUT_DEFAULT,
+                propertiesComparator.checkProperties(toCheck, toCheckAgainst));
+        x2 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"},\n" +
+                "\"city\": { \"type\": \"string\"}\n" +
+                "},\n" +
+                "\"required\": [\"city\"]\n" +
+                "}\n";
+        x1 = "{\n" +
+                "\"type\": \"object\",\n" +
+                "\"properties\": {\n" +
+                "\"number\": { \"type\": \"number\" },\n" +
+                "\"street_name\": { \"type\": \"string\" },\n" +
+                "\"street_type\": { \"type\": \"string\"},\n" +
+                "\"city\": { \"type\": \"string\"}\n" +
+                "},\n" +
+                "\"required\": [\"city\", \"number\"]\n" +
+                "}\n";
+        toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.REQUIRED_PROPERTY_ADDED_WITHOUT_DEFAULT,
+                propertiesComparator.checkProperties(toCheck, toCheckAgainst));
+    }
+}
diff --git a/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/StringComparatorTest.java b/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/StringComparatorTest.java
new file mode 100644
index 000000000..4023f0d92
--- /dev/null
+++ b/server/src/test/java/io/pravega/schemaregistry/rules/jsoncompatibility/StringComparatorTest.java
@@ -0,0 +1,84 @@
+package io.pravega.schemaregistry.rules.jsoncompatibility;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+public class StringComparatorTest {
+    
+    @Test
+    public void testMinLengthComparator() throws IOException {
+        StringComparator stringComparator = new StringComparator();
+        ObjectMapper objectMapper = new ObjectMapper();
+        String x2 = "{\n" +
+                "\"type\": \"string\"\n" +
+                "}\n";
+        String x1 = "{\n" +
+                "\"type\": \"string\" ,\n" +
+                "\"minLength\": 3\n" +
+                "}\n";
+        JsonNode toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        JsonNode toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.STRING_TYPE_MIN_LENGTH_ADDED,
+                stringComparator.stringComparator(toCheck, toCheckAgainst));
+        x2 = "{\n" +
+                "\"type\": \"string\" ,\n" +
+                "\"minLength\": 2\n" +
+                "}\n";
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.STRING_TYPE_MIN_LENGTH_VALUE_INCREASED,
+                stringComparator.stringComparator(toCheck, toCheckAgainst));
+    }
+    
+    @Test
+    public void testMaxLengthComparator() throws IOException {
+        StringComparator stringComparator = new StringComparator();
+        ObjectMapper objectMapper = new ObjectMapper();
+        String x2 = "{\n" +
+                "\"type\": \"string\"\n" +
+                "}\n";
+        String x1 = "{\n" +
+                "\"type\": \"string\" ,\n" +
+                "\"maxLength\": 3\n" +
+                "}\n";
+        JsonNode toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        JsonNode toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.STRING_TYPE_MAX_LENGTH_ADDED,
+                stringComparator.stringComparator(toCheck, toCheckAgainst));
+        x2 = "{\n" +
+                "\"type\": \"string\" ,\n" +
+                "\"maxLength\": 4\n" +
+                "}\n";
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.STRING_TYPE_MAX_LENGTH_VALUE_DECREASED,
+                stringComparator.stringComparator(toCheck, toCheckAgainst));
+    }
+    
+    @Test
+    public void testPatternComparator() throws IOException {
+        StringComparator stringComparator = new StringComparator();
+        ObjectMapper objectMapper = new ObjectMapper();
+        String x2 = "{\n" +
+                "\"type\": \"string\"\n" +
+                "}\n";
+        String x1 = "{\n" +
+                "\"type\": \"string\" ,\n" +
+                "\"pattern\": \"^(\\\\([0-9]{3}\\\\))?[0-9]{3}-[0-9]{4}$\"\n" +
+                "}\n";
+        JsonNode toCheck = objectMapper.readTree(ByteBuffer.wrap(x1.getBytes()).array());
+        JsonNode toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.STRING_TYPE_PATTERN_ADDED,
+                stringComparator.stringComparator(toCheck, toCheckAgainst));
+        x2 = "{\n" +
+                "\"type\": \"string\" ,\n" +
+                "\"pattern\": \"^(\\\\([0-9]{3}\\\\))?[0-9]{4}-[0-9]{5}$\"\n" +
+                "}\n";
+        toCheckAgainst = objectMapper.readTree(ByteBuffer.wrap(x2.getBytes()).array());
+        Assert.assertEquals(BreakingChangesStore.BreakingChanges.STRING_TYPE_PATTERN_MODIFIED,
+                stringComparator.stringComparator(toCheck, toCheckAgainst));
+    }
+}