@Test public void testConfluentAvroDeserializer() throws IOException, RestClientException { WorkUnitState mockWorkUnitState = getMockWorkUnitState(0L,10L); mockWorkUnitState.setProp("schema.registry.url", TEST_URL); Schema schema = SchemaBuilder.record(TEST_RECORD_NAME) .namespace(TEST_NAMESPACE).fields() .name(TEST_FIELD_NAME).type().stringType().noDefault() .endRecord(); GenericRecord testGenericRecord = new GenericRecordBuilder(schema).set(TEST_FIELD_NAME, "testValue").build(); SchemaRegistryClient mockSchemaRegistryClient = mock(SchemaRegistryClient.class); when(mockSchemaRegistryClient.getByID(any(Integer.class))).thenReturn(schema); Serializer<Object> kafkaEncoder = new KafkaAvroSerializer(mockSchemaRegistryClient); Deserializer<Object> kafkaDecoder = new KafkaAvroDeserializer(mockSchemaRegistryClient); ByteBuffer testGenericRecordByteBuffer = ByteBuffer.wrap(kafkaEncoder.serialize(TEST_TOPIC_NAME, testGenericRecord)); KafkaSchemaRegistry<Integer, Schema> mockKafkaSchemaRegistry = mock(KafkaSchemaRegistry.class); KafkaDeserializerExtractor kafkaDecoderExtractor = new KafkaDeserializerExtractor(mockWorkUnitState, Optional.fromNullable(Deserializers.CONFLUENT_AVRO), kafkaDecoder, mockKafkaSchemaRegistry); ByteArrayBasedKafkaRecord mockMessageAndOffset = getMockMessageAndOffset(testGenericRecordByteBuffer); Assert.assertEquals(kafkaDecoderExtractor.decodeRecord(mockMessageAndOffset), testGenericRecord); }
@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; }