ConfluentSchemaRegistryDeserializer extends KafkaAvroDeserializer { @Override public Object deserialize(String s, byte[] bytes) { return this.deserialize(bytes, readerSchema); } ConfluentSchemaRegistryDeserializer(SchemaRegistryClient client, Schema readerSchema); @Override Object deserialize(String s, byte[] bytes);  }
@Test public void testDeserialize() { String schemaRegistryUrl = "mock: String subject = "mytopic"; SchemaRegistryClient mockRegistryClient = mockSchemaRegistryClient(schemaRegistryUrl, subject); Map<String, Object> map = new HashMap<>(); map.put(AbstractKafkaAvroSerDeConfig.AUTO_REGISTER_SCHEMAS, true); map.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl); Serializer<GenericRecord> serializer = (Serializer) new KafkaAvroSerializer(mockRegistryClient); serializer.configure(map, true); byte[] bytes = serializer.serialize( subject, new GenericRecordBuilder(AVRO_SCHEMA_V1) .set("name", "KeyName") .set("age", 1) .set("favorite_number", 2) .set("favorite_color", "color3") .build()); Object deserialized = mockDeserializerProvider(schemaRegistryUrl, subject, null) .getDeserializer(new HashMap<>(), true) .deserialize(subject, bytes); GenericRecord expected = new GenericRecordBuilder(AVRO_SCHEMA) .set("name", "KeyName") .set("favorite_number", 2) .set("favorite_color", "color3") .build(); assertEquals(expected, deserialized); }