@Test public void testConfluentJsonDeserializer() throws IOException { WorkUnitState mockWorkUnitState = getMockWorkUnitState(0L, 10L); mockWorkUnitState.setProp("json.value.type", KafkaRecord.class.getName()); KafkaRecord testKafkaRecord = new KafkaRecord("Hello World"); Serializer<KafkaRecord> kafkaEncoder = new KafkaJsonSerializer<>(); kafkaEncoder.configure(PropertiesUtils.propsToStringKeyMap(mockWorkUnitState.getProperties()), false); Deserializer<KafkaRecord> kafkaDecoder = new KafkaJsonDeserializer<>(); kafkaDecoder.configure(PropertiesUtils.propsToStringKeyMap(mockWorkUnitState.getProperties()), false); ByteBuffer testKafkaRecordByteBuffer = ByteBuffer.wrap(kafkaEncoder.serialize(TEST_TOPIC_NAME, testKafkaRecord)); KafkaSchemaRegistry<?, ?> mockKafkaSchemaRegistry = mock(KafkaSchemaRegistry.class); KafkaDeserializerExtractor kafkaDecoderExtractor = new KafkaDeserializerExtractor(mockWorkUnitState, Optional.fromNullable(Deserializers.CONFLUENT_JSON), kafkaDecoder, mockKafkaSchemaRegistry); ByteArrayBasedKafkaRecord mockMessageAndOffset = getMockMessageAndOffset(testKafkaRecordByteBuffer); Assert.assertEquals(kafkaDecoderExtractor.decodeRecord(mockMessageAndOffset), testKafkaRecord); }
@Override protected Object decodeRecord(ByteArrayBasedKafkaRecord messageAndOffset) throws IOException { Object deserialized = kafkaDeserializer.deserialize(this.topicName, messageAndOffset.getMessageBytes()); return (this.latestSchema == null) ? deserialized : AvroUtils.convertRecordSchema((GenericRecord) deserialized, this.latestSchema); }
KafkaDeserializerExtractor extends KafkaExtractor<Object, Object> { @Override protected Object decodeRecord(ByteArrayBasedKafkaRecord messageAndOffset) throws IOException { Object deserialized = kafkaDeserializer.deserialize(this.topicName, messageAndOffset.getMessageBytes()); return (this.latestSchema == null) ? deserialized : AvroUtils.convertRecordSchema((GenericRecord) deserialized, this.latestSchema); } }
KafkaDeserializerExtractor extends KafkaExtractor<Object, Object> { @Override protected Object decodeRecord(ByteArrayBasedKafkaRecord messageAndOffset) throws IOException { Object deserialized = kafkaDeserializer.deserialize(this.topicName, messageAndOffset.getMessageBytes()); return (this.latestSchema == null) ? deserialized : AvroUtils.convertRecordSchema((GenericRecord) deserialized, this.latestSchema); } 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> { @Override protected Object decodeRecord(ByteArrayBasedKafkaRecord messageAndOffset) throws IOException { Object deserialized = kafkaDeserializer.deserialize(this.topicName, messageAndOffset.getMessageBytes()); return (this.latestSchema == null) ? deserialized : AvroUtils.convertRecordSchema((GenericRecord) deserialized, this.latestSchema); } 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> { @Override protected Object decodeRecord(ByteArrayBasedKafkaRecord messageAndOffset) throws IOException { Object deserialized = kafkaDeserializer.deserialize(this.topicName, messageAndOffset.getMessageBytes()); return (this.latestSchema == null) ? deserialized : AvroUtils.convertRecordSchema((GenericRecord) deserialized, this.latestSchema); } 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; }