@Test public void testKafkaAvroSerializerSupportsSchemaEvolution() throws IOException, RestClientException { final String fieldToDelete = "fieldToDelete"; final String newOptionalField = "newOptionalField"; Schema schemaV1 = SchemaBuilder .record("SchemaEvolution") .namespace("example.avro") .fields() .requiredString(fieldToDelete) .endRecord(); Schema schemaV2 = SchemaBuilder .record("SchemaEvolution") .namespace("example.avro") .fields() .nullableString(newOptionalField, "optional") .endRecord(); AvroSchema avroSchemaV1 = new AvroSchema(schemaV1); AvroSchema avroSchemaV2 = new AvroSchema(schemaV2); assertTrue("Schema V2 should be backwards compatible", avroSchemaV2.isBackwardCompatible(avroSchemaV1).isEmpty()); GenericRecord recordV1 = new GenericData.Record(avroSchemaV1.rawSchema()); recordV1.put(fieldToDelete, "present"); byte[] bytes = avroSerializer.serialize(topic, recordV1); GenericRecord genericRecordV2 = (GenericRecord) avroDeserializer.deserialize(topic, bytes, avroSchemaV2.rawSchema()); assertNotNull("Optional field should have a non-null default value", genericRecordV2.get(newOptionalField)); assertNull("Field was removed in schema V2 but is still present", genericRecordV2.get(fieldToDelete)); }
@Override public byte[] serialize(String topic, Object record) { if (record == null) { return null; } AvroSchema schema = new AvroSchema(AvroSchemaUtils.getSchema(record, useSchemaReflection)); return serializeImpl(getSubjectName(topic, isKey, record, schema), record, schema); }
KafkaAvroSerializer extends AbstractKafkaAvroSerializer implements Serializer<Object> { @Override public byte[] serialize(String topic, Object record) { if (record == null) { return null; } AvroSchema schema = new AvroSchema(AvroSchemaUtils.getSchema(record, useSchemaReflection)); return serializeImpl(getSubjectName(topic, isKey, record, schema), record, schema); } }
KafkaAvroSerializer extends AbstractKafkaAvroSerializer implements Serializer<Object> { @Override public byte[] serialize(String topic, Object record) { if (record == null) { return null; } AvroSchema schema = new AvroSchema(AvroSchemaUtils.getSchema(record, useSchemaReflection)); return serializeImpl(getSubjectName(topic, isKey, record, schema), record, schema); } KafkaAvroSerializer();  KafkaAvroSerializer(SchemaRegistryClient client);  KafkaAvroSerializer(SchemaRegistryClient client, Map<String, ?> props); }
KafkaAvroSerializer extends AbstractKafkaAvroSerializer implements Serializer<Object> { @Override public byte[] serialize(String topic, Object record) { if (record == null) { return null; } AvroSchema schema = new AvroSchema(AvroSchemaUtils.getSchema(record, useSchemaReflection)); return serializeImpl(getSubjectName(topic, isKey, record, schema), record, schema); } KafkaAvroSerializer();  KafkaAvroSerializer(SchemaRegistryClient client);  KafkaAvroSerializer(SchemaRegistryClient client, Map<String, ?> props); @Override void configure(Map<String, ?> configs, boolean isKey); @Override byte[] serialize(String topic, Object record); @Override void close(); }
KafkaAvroSerializer extends AbstractKafkaAvroSerializer implements Serializer<Object> { @Override public byte[] serialize(String topic, Object record) { if (record == null) { return null; } AvroSchema schema = new AvroSchema(AvroSchemaUtils.getSchema(record, useSchemaReflection)); return serializeImpl(getSubjectName(topic, isKey, record, schema), record, schema); } KafkaAvroSerializer();  KafkaAvroSerializer(SchemaRegistryClient client);  KafkaAvroSerializer(SchemaRegistryClient client, Map<String, ?> props); @Override void configure(Map<String, ?> configs, boolean isKey); @Override byte[] serialize(String topic, Object record); @Override void close();  }