private static KafkaSchemaRegistry<?, ?> getKafkaSchemaRegistry(Properties props)
      throws ReflectiveOperationException {

    Optional<Deserializers> deserializerType =
        Enums.getIfPresent(Deserializers.class, props.getProperty(KAFKA_DESERIALIZER_TYPE).toUpperCase());

    if (deserializerType.isPresent()) {
      return ConstructorUtils.invokeConstructor(deserializerType.get().getSchemaRegistryClass(), props);
    }
    if (props.containsKey(KafkaSchemaRegistry.KAFKA_SCHEMA_REGISTRY_CLASS)) {
      return KafkaSchemaRegistry.get(props);
    }
    return new SimpleKafkaSchemaRegistry(props);
  }
private static KafkaSchemaRegistry < ? , ? > getKafkaSchemaRegistry ( Properties props ) throws ReflectiveOperationException { Optional < Deserializers > deserializerType = Enums . getIfPresent ( Deserializers . class , props . getProperty ( KAFKA_DESERIALIZER_TYPE ) . toUpperCase ( ) ) ; if ( deserializerType . isPresent ( ) ) { return ConstructorUtils . invokeConstructor ( deserializerType . get ( ) . getSchemaRegistryClass ( ) , props ) ; } if ( props . containsKey ( KafkaSchemaRegistry . KAFKA_SCHEMA_REGISTRY_CLASS ) ) { return KafkaSchemaRegistry . get ( props ) ; } return new SimpleKafkaSchemaRegistry ( props ) ; }
Constructs a {@link KafkaSchemaRegistry} using the value of {@link #KAFKA_DESERIALIZER_TYPE}, if not set it
defaults to {@link SimpleKafkaSchemaRegistry}.