@Test public void testBuiltInStringDeserializer() throws ReflectiveOperationException { WorkUnitState mockWorkUnitState = getMockWorkUnitState(0L, 10L); mockWorkUnitState.setProp(KafkaDeserializerExtractor.KAFKA_DESERIALIZER_TYPE, KafkaDeserializerExtractor.Deserializers.STRING.name()); KafkaDeserializerExtractor kafkaDecoderExtractor = new KafkaDeserializerExtractor(mockWorkUnitState); Assert.assertEquals(kafkaDecoderExtractor.getKafkaDeserializer().getClass(), KafkaDeserializerExtractor.Deserializers.STRING.getDeserializerClass()); Assert.assertEquals(kafkaDecoderExtractor.getKafkaSchemaRegistry().getClass(), KafkaDeserializerExtractor.Deserializers.STRING.getSchemaRegistryClass()); }
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); }
KafkaDeserializerExtractor extends KafkaExtractor<Object, Object> { 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); } }
KafkaDeserializerExtractor extends KafkaExtractor<Object, Object> { 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); } KafkaDeserializerExtractor(WorkUnitState state); private  KafkaDeserializerExtractor(WorkUnitState state, Optional<Deserializers> deserializerType); @VisibleForTesting  KafkaDeserializerExtractor(WorkUnitState state, Optional<Deserializers> deserializerType,
      Deserializer<?> kafkaDeserializer, KafkaSchemaRegistry<?, ?> kafkaSchemaRegistry); }
KafkaDeserializerExtractor extends KafkaExtractor<Object, Object> { 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); } KafkaDeserializerExtractor(WorkUnitState state); private  KafkaDeserializerExtractor(WorkUnitState state, Optional<Deserializers> deserializerType); @VisibleForTesting  KafkaDeserializerExtractor(WorkUnitState state, Optional<Deserializers> deserializerType,
      Deserializer<?> kafkaDeserializer, KafkaSchemaRegistry<?, ?> kafkaSchemaRegistry); @Override Object getSchema(); }
KafkaDeserializerExtractor extends KafkaExtractor<Object, Object> { 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); } KafkaDeserializerExtractor(WorkUnitState state); private  KafkaDeserializerExtractor(WorkUnitState state, Optional<Deserializers> deserializerType); @VisibleForTesting  KafkaDeserializerExtractor(WorkUnitState state, Optional<Deserializers> deserializerType,
      Deserializer<?> kafkaDeserializer, KafkaSchemaRegistry<?, ?> kafkaSchemaRegistry); @Override Object getSchema(); static final String KAFKA_DESERIALIZER_TYPE; }