private static void validateValue(Schema schema, Object value, String location) {
        if (value == null) {
            if (!schema.isOptional())
                throw new DataException("Invalid value: null used for required " + location
                        + ", schema type: " + schema.type());
            return;
        }

        List<Class<?>> expectedClasses = expectedClassesFor(schema);
        boolean foundMatch = false;
        for (Class<?> expectedClass : expectedClasses) {
            if (expectedClass.isInstance(value)) {
                foundMatch = true;
                break;
            }
        }

        if (!foundMatch) {
            StringBuilder exceptionMessage = new StringBuilder("Invalid Java object for schema");
            if (schema.name() != null) {
                exceptionMessage.append(" \"").append(schema.name()).append("\"");
            }
            exceptionMessage.append(" with type ").append(schema.type()).append(": ").append(value.getClass());
            if (location != null) {
                exceptionMessage.append(" for ").append(location);
            }
            throw new DataException(exceptionMessage.toString());
        }

        switch (schema.type()) {
            case STRUCT:
                Struct struct = (Struct) value;
                if (!struct.schema().equals(schema))
                    throw new DataException("Struct schemas do not match.");
                struct.validate();
                break;
            case ARRAY:
                List<?> array = (List<?>) value;
                String entryLocation = "element of array " + location;
                Schema arrayValueSchema = assertSchemaNotNull(schema.valueSchema(), entryLocation);
                for (Object entry : array) {
                    validateValue(arrayValueSchema, entry, entryLocation);
                }
                break;
            case MAP:
                Map<?, ?> map = (Map<?, ?>) value;
                String keyLocation = "key of map " + location;
                String valueLocation = "value of map " + location;
                Schema mapKeySchema = assertSchemaNotNull(schema.keySchema(), keyLocation);
                Schema mapValueSchema = assertSchemaNotNull(schema.valueSchema(), valueLocation);
                for (Map.Entry<?, ?> entry : map.entrySet()) {
                    validateValue(mapKeySchema, entry.getKey(), keyLocation);
                    validateValue(mapValueSchema, entry.getValue(), valueLocation);
                }
                break;
        }
    }
Validate that the value can be used with the schema, i.e. that its type matches the schema type and nullability
requirements. Throws a {@link DataException} if the value is invalid.
@param schema Schema to test
@param value value to test
connect/api/src/main/java/org/apache/kafka/connect/data/ConnectSchema.java