If provided with an AVRO schema, validates it and checks if there are
backwards compatible.

TODO should probably place some similar checks for other serializer types
as well?

@param serializerDef
971969
private static void validateIfAvroSchema(SerializerDefinition serializerDef) {
        if(serializerDef.getName().equals(AVRO_GENERIC_VERSIONED_TYPE_NAME)
           || serializerDef.getName().equals(AVRO_GENERIC_TYPE_NAME)) {
            SchemaEvolutionValidator.validateAllAvroSchemas(serializerDef);
            // check backwards compatibility if needed
            if(serializerDef.getName().equals(AVRO_GENERIC_VERSIONED_TYPE_NAME)) {
                SchemaEvolutionValidator.checkSchemaCompatibility(serializerDef);
            }
        }
    }
205918
public boolean isJsonDeserializer( JType type ) {
        return null != type.isClass() && type.isClass().isAssignableTo( jsonDeserializerType );
    }
913163
void comparePrimitiveTypes(Schema oldSchema,
                                                    Schema newSchema,
                                                    List<Message> messages,
                                                    String name) {
        if(oldSchema == null || newSchema == null) {
            throw new IllegalArgumentException("Old schema must both be a primitive type. Name="
                                               + name + ". Type=" + oldSchema);
        }

        Schema.Type oldType = oldSchema.getType();
        Schema.Type newType = newSchema.getType();

        if(oldType != newType) {
            if(((oldType == Schema.Type.INT && (newType == Schema.Type.LONG
                                                || newType == Schema.Type.FLOAT || newType == Schema.Type.DOUBLE))
                || (oldType == Schema.Type.LONG && (newType == Schema.Type.FLOAT || newType == Schema.Type.DOUBLE)) || (oldType == Schema.Type.FLOAT && (newType == Schema.Type.DOUBLE)))) {
                messages.add(new Message(Level.INFO, "Type change from " + oldSchema.getType()
                                                     + " to " + newSchema.getType() + " for field "
                                                     + name));
            } else {
                messages.add(new Message(Level.ERROR, "Illegal type change from "
                                                      + oldSchema.getType() + " to "
                                                      + newSchema.getType() + " for field " + name));
            }
        }
    }
22112
public static boolean isFieldDeprecated(JavacNode field) {
		if (!(field.get() instanceof JCVariableDecl)) return false;
		JCVariableDecl fieldNode = (JCVariableDecl) field.get();
		if ((fieldNode.mods.flags & Flags.DEPRECATED) != 0) {
			return true;
		}
		for (JavacNode child : field.down()) {
			if (annotationTypeMatches(Deprecated.class, child)) {
				return true;
			}
		}
		return false;
	}
135914
private void marshalVersionField() {
    PropertyMetadata versionMetadata = entityMetadata.getVersionMetadata();
    if (versionMetadata != null) {
      long version = (long) IntrospectionUtils.getFieldValue(versionMetadata, entity);
      ValueBuilder<?, ?, ?> valueBuilder = versionMetadata.getMapper().toDatastore(version + 1);
      valueBuilder.setExcludeFromIndexes(!versionMetadata.isIndexed());
      entityBuilder.set(versionMetadata.getMappedName(), valueBuilder.build());
    }
  }
41311
def compat_serializer_check_is_valid(serializer):
    
    if DRFVLIST[0] >= 3:
        serializer.is_valid(raise_exception=True)
    else:
        if not serializer.is_valid():
            serializers.ValidationError('The serializer raises a validation error')
133509
@SuppressWarnings({ "unchecked" })
    @Override
    public void serializeObject(HashGenericRecord rec, String fieldName, String typeName, Object obj) {
        if (obj == null) {
            return;
        }

        getSerializer(typeName).serialize(obj, rec);
    }
1135257
@SuppressWarnings({ "rawtypes", "unchecked" })
  public static Schema toPangoolSchema(org.apache.avro.Schema avroSchema) {
		List<Field> fields = new ArrayList<Field>();
		for(org.apache.avro.Schema.Field avroField : avroSchema.getFields()) {
			org.apache.avro.Schema.Type type = avroField.schema().getType();
			Field pangoolField;
			switch(type){
			case INT: pangoolField = Field.create(avroField.name(),Type.INT); break;
			case LONG: pangoolField = Field.create(avroField.name(),Type.LONG); break;
			case FLOAT: pangoolField = Field.create(avroField.name(),Type.FLOAT); break;
			case DOUBLE: pangoolField = Field.create(avroField.name(),Type.DOUBLE); break;
			case BOOLEAN: pangoolField = Field.create(avroField.name(),Type.BOOLEAN); break;
			case STRING: pangoolField = Field.create(avroField.name(),Type.STRING); break;
			case BYTES:
				if (avroField.getProp(Field.METADATA_BYTES_AS_OBJECT) == null){
					pangoolField = Field.create(avroField.name(),Type.BYTES);
				} else {
					try {
					String objectClazz = avroField.getProp(Field.METADATA_OBJECT_CLASS);
					pangoolField = Field.createObject(avroField.name(),Class.forName(objectClazz));
					String serializationString = avroField.getProp(Field.METADATA_OBJECT_SERIALIZATION);
					if (serializationString != null){
							Class<? extends Serialization> ser=
		             (serializationString == null) ? null : 
			          	(Class<? extends Serialization>)Class.forName(serializationString);
							pangoolField.setObjectSerialization(ser);
					}
					
					} catch(ClassNotFoundException e) {
	          throw new PangoolRuntimeException(e);
          }
				}
				break;
			case ENUM:
				String objectClazz = avroField.getProp(Field.METADATA_OBJECT_CLASS);
				try{
					pangoolField = Field.createEnum(avroField.name(),Class.forName(objectClazz));
					} catch(ClassNotFoundException e){
						throw new PangoolRuntimeException(e);
					}
				break;
			default:
				throw new PangoolRuntimeException("Avro type:" + type + " can't be converted to Pangool Schema type");
			}
			//add properties 
			for(Map.Entry<String,String> entry : avroField.props().entrySet()){
				if (!Field.RESERVED_KEYWORDS.contains(entry.getKey())){
					pangoolField.addProp(entry.getKey(),entry.getValue());
				}
			}
			fields.add(pangoolField);
			
		}
		Schema schema = new Schema(avroSchema.getFullName(), fields);
		return schema;
	}
980695
@SuppressWarnings("rawtypes")
	public void registerTypeWithKryoSerializer(Class<?> type, Class<? extends Serializer> serializerClass) {
		if (type == null || serializerClass == null) {
			throw new NullPointerException("Cannot register null class or serializer.");
		}

		@SuppressWarnings("unchecked")
		Class<? extends Serializer<?>> castedSerializerClass = (Class<? extends Serializer<?>>) serializerClass;
		registeredTypesWithKryoSerializerClasses.put(type, castedSerializerClass);
	}
1193277
public static boolean isPrimitive(Field field )
	{
		if(field == null)
			return false;
		
		String fieldName = field.getName();
		
		if("serialVersionUID".equals(fieldName))
			return false;
		
		return isPrimitive(field.getType());
		
	
	}
364228
private void validateString(Class<?> clazz) throws SerializerException {
        if (null != clazz && !clazz.isAssignableFrom(String.class)) {
            throw new SerializerException("String serializer is able to work only with data types assignable from java.lang.String");
        }
    }
461473
public final void register(Class type, Serializer serializer) {
        if (type == null) {
            throw new IllegalArgumentException("Class type information is required!");
        }
        if (serializer.getTypeId() <= 0) {
            throw new IllegalArgumentException(
                    "Type ID must be positive! Current: " + serializer.getTypeId() + ", Serializer: " + serializer);
        }
        safeRegister(type, createSerializerAdapter(serializer, this));
    }
833112
@Deprecated
    @Override
    @SuppressWarnings("unchecked")
    public void serializeObject(GenericObject rec, String fieldName, String typeName, Object obj) {
        if( obj == null ){
            rec.add(fieldName, null);
        } else if (isPrimitive(obj.getClass())){
            serializePrimitive(rec, fieldName, obj);
            return;
        } else {
            GenericObject subObject = new GenericObject(typeName, obj);
            getSerializer(typeName).serialize(obj, subObject);
            rec.add(fieldName, subObject);
        }
    }
1127696
private void _serializeSerializable(Serializable serializable, StringBuilder sb) throws ConverterException {

	sb.append(goIn());
	sb.append("evaluateJava(").append(QUOTE_CHR);
	try {
	    sb.append(JavaConverter.serialize(serializable));
	}
	catch (IOException e) {
	    throw toConverterException(e);
	}
	sb.append(QUOTE_CHR).append(')');
    }
1165133
private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
		/*
		Please see FLINK-11436 for details on why manual deserialization is required.

		During the release of Flink 1.7, the value of serialVersionUID was uptick to 2L (was 1L before)
		And although the AvroSerializer (along with it's snapshot class) were migrated to the new serialization
		abstraction (hence free from Java serialization), there were composite serializers that were not migrated
		and were serialized with Java serialization. In case that one of the nested serializers were Avro we would
		bump into deserialization exception due to a wrong serialVersionUID. Unfortunately it is not possible to revert
		the serialVersionUID back to 1L, because users might have snapshots with 2L present already.
		To overcome this we first need to make sure that the AvroSerializer is being Java deserialized with
		FailureTolerantObjectInputStream, and then we determine the serialized layout by looking at the fields.

		From: https://docs.oracle.com/javase/8/docs/platform/serialization/spec/class.html#a5421
		-------------------------------------------------------------------------------------------------------------
		The descriptors for primitive typed fields are written first
		sorted by field name followed by descriptors for the object typed fields sorted by field name.
		The names are sorted using String.compareTo.
		-------------------------------------------------------------------------------------------------------------

		pre 1.6		field order:   	[type]
		pre 1.7 	field order:   	[schemaString, 		type]
		post 1.7 	field order:	[previousSchema,	schema,		type]

		We would use the first field to distinguish between the three different layouts.
		To complicate things even further in pre 1.7, the field @schemaString could be
		null or a string, but, in post 1.7, the field @previousSchema was never set to null, therefore
		we can use the first field to determine the version.

		this logic should stay here as long as we support Flink 1.6 (along with Java serialized
		TypeSerializers)
		*/
		final Object firstField = in.readObject();

		if (firstField == null) {
			// first field can only be NULL in 1.6 (schemaString)
			read16Layout(null, in);
		}
		else if (firstField instanceof String) {
			// first field is a String only in 1.6 (schemaString)
			read16Layout((String) firstField, in);
		}
		else if (firstField instanceof Class<?>) {
			// first field is a Class<?> only in 1.5 (type)
			@SuppressWarnings("unchecked") Class<T> type = (Class<T>) firstField;
			read15Layout(type);
		}
		else if (firstField instanceof SerializableAvroSchema) {
			readCurrentLayout((SerializableAvroSchema) firstField, in);
		}
		else {
			throw new IllegalStateException("Failed to Java-Deserialize an AvroSerializer instance. " +
				"Was expecting a first field to be either a String or SerializableAvroSchema, but got: " +
				"" + firstField.getClass());
		}
	}
1313378
void compareRecordTypes(Schema oldSchema,
                                                 Schema newSchema,
                                                 List<Message> messages,
                                                 String name) {
        if(oldSchema == null || newSchema == null || oldSchema.getType() != Schema.Type.RECORD) {
            throw new IllegalArgumentException("Old schema must be RECORD type. Name=" + name
                                               + ". Type=" + oldSchema);
        }

        if(newSchema.getType() != Schema.Type.RECORD) {
            messages.add(new Message(Level.ERROR, "Illegal type change from " + oldSchema.getType()
                                                  + " to " + newSchema.getType() + " for field "
                                                  + name));
            return;
        }

        // Check all of the fields in the new schema
        for(Field newField: newSchema.getFields()) {
            String fieldName = newField.name();
            Field oldField = oldSchema.getField(fieldName);

            if(oldField == null) {
                // This is a new field that did not exist in the original
                // schema.
                // Check if it is optional or has a default value.
                if(isOptional(newField)) {
                    if(newField.defaultValue() == null) {
                        messages.add(new Message(Level.INFO, "Added optional field " + name + "."
                                                             + fieldName
                                                             + " with no default value."));
                    } else {
                        messages.add(new Message(Level.INFO, "Added optional field " + name + "."
                                                             + fieldName + " with default value: "
                                                             + newField.defaultValue()));
                    }
                } else {
                    if(newField.defaultValue() == null) {
                        messages.add(new Message(Level.ERROR, "Added required field " + name + "."
                                                              + fieldName
                                                              + " with no default value."));
                    } else {
                        messages.add(new Message(Level.INFO, "Added required field " + name + "."
                                                             + fieldName + " with default value: "
                                                             + newField.defaultValue()));
                    }
                }
            } else {
                // This is a field that existed in the original schema.

                // Check if the field was changed from optional to required or
                // vice versa.
                boolean newFieldIsOptional = isOptional(newField);
                boolean oldFieldIsOptional = isOptional(oldField);

                if(oldFieldIsOptional != newFieldIsOptional) {
                    if(oldFieldIsOptional) {
                        messages.add(new Message(Level.ERROR,
                                                 "Existing field " + name + "." + fieldName
                                                         + " was optional and is now required."));
                    } else {
                        messages.add(new Message(Level.WARN, "Existing field " + name + "."
                                                             + fieldName
                                                             + " was required and is now optional."));
                    }
                }

                // Recursively compare the nested field types
                compareTypes(oldField.schema(), newField.schema(), messages, name + "." + fieldName);

                // Check if the default value has been changed
                if(newField.defaultValue() == null) {
                    if(oldField.defaultValue() != null) {
                        messages.add(new Message(Level.WARN,
                                                 "Removed default value for existing field " + name
                                                         + "." + fieldName
                                                         + ". The old default was: "
                                                         + oldField.defaultValue()));
                    }
                } else // newField.defaultValue() != null
                {
                    if(oldField.defaultValue() == null) {
                        messages.add(new Message(Level.WARN,
                                                 "Added a default value for existing field " + name
                                                         + "." + fieldName
                                                         + ". The new default is: "
                                                         + newField.defaultValue()));
                    } else if(!newField.defaultValue().equals(oldField.defaultValue())) {
                        messages.add(new Message(Level.INFO,
                                                 "Changed the default value for existing field "
                                                         + name + "." + fieldName
                                                         + ". The old default was: "
                                                         + oldField.defaultValue()
                                                         + ". The new default is: "
                                                         + newField.defaultValue()));
                    }
                }
            }

            // For all fields in the new schema (whether or not it existed in
            // the old schema), if there is a default value for this field, make
            // sure it is legal.
            if(newField.defaultValue() != null) {
                checkDefaultValueIsLegal(newField, messages, name + "." + fieldName);
            }
        }

        // Check if any fields were removed.
        for(Field oldField: newSchema.getFields()) {
            String fieldName = oldField.name();
            Field newField = newSchema.getField(fieldName);

            if(newField == null) {
                if(isOptional(oldField)) {
                    messages.add(new Message(Level.INFO, "Removed optional field " + name + "."
                                                         + fieldName));
                } else {
                    messages.add(new Message(Level.WARN, "Removed required field " + name + "."
                                                         + fieldName));
                }
            }
        }

        // Check if indexing was modified or added to any old fields.
        for(Field oldField: oldSchema.getFields()) {
            if(newSchema.getField(oldField.name()) != null) {
                String oldIndexType = oldField.getProp("indexType");
                String newIndexType = newSchema.getField(oldField.name()).getProp("indexType");

                // Check if added indexing.
                if(oldIndexType == null && newIndexType != null) {
                    messages.add(new Message(Level.ERROR,
                                             "Cannot add indexing to "
                                                     + oldField.name()
                                                     + ". Adding indexing to fields created without indexing is not supported."));
                }

                // Check if changed indexing
                if(oldIndexType != null && !oldIndexType.equals(newIndexType)) {
                    messages.add(new Message(Level.ERROR,
                                             "Cannot change indexing from "
                                                     + oldIndexType
                                                     + " to "
                                                     + newIndexType
                                                     + " for "
                                                     + oldField.name()
                                                     + ". Changing indexing method is not supported."));
                }
            }
        }
    }
896881
function typeFromSchemaResponse(schema, parseOptions) {
  var schemaType = avro.parse(schema);

  //check if the schema has been previouisly parsed and added to the registry
  if(typeof parseOptions.registry === 'object' && typeof parseOptions.registry[schemaType.name] !== 'undefined'){
    return parseOptions.registry[schemaType.name];
  }

  return avro.parse(schema, parseOptions);
}