@Test public void testSchemaVersionSet() throws IOException, RestClientException { IndexedRecord avroRecord = createAvroRecord(); int version = schemaRegistry.register("topic", new AvroSchema(avroRecord.getSchema())); byte[] bytes = avroSerializer.serialize("topic", avroRecord); GenericContainerWithVersion genericContainerWithVersion = (GenericContainerWithVersion) deserializer.deserializeWithSchemaAndVersion( "topic", false, bytes); org.apache.avro.Schema avroSchema = genericContainerWithVersion.container().getSchema(); Integer schemaVersion = genericContainerWithVersion.version(); assertThat(schemaVersion, equalTo(version)); }
protected GenericContainerWithVersion deserializeWithSchemaAndVersion( String topic, boolean isKey, byte[] payload) throws SerializationException { if (payload == null) { return null; } DeserializationContext context = new DeserializationContext(topic, isKey, payload); AvroSchema schema = context.schemaForDeserialize(); Object result = context.read(schema.rawSchema(), null); try { Integer version = schemaVersion(topic, isKey, context.getSchemaId(), context.getSubject(), schema, result); if (schema.rawSchema().getType().equals(Schema.Type.RECORD)) { return new GenericContainerWithVersion((GenericContainer) result, version); } else { return new GenericContainerWithVersion(new NonRecordContainer(schema.rawSchema(), result), version); } } catch (RestClientException | IOException e) { throw new SerializationException("Error retrieving Avro " + getSchemaType(isKey) + " schema version for id " + context.getSchemaId(), e); } }
AbstractKafkaAvroDeserializer extends AbstractKafkaSchemaSerDe { protected GenericContainerWithVersion deserializeWithSchemaAndVersion( String topic, boolean isKey, byte[] payload) throws SerializationException { if (payload == null) { return null; } DeserializationContext context = new DeserializationContext(topic, isKey, payload); AvroSchema schema = context.schemaForDeserialize(); Object result = context.read(schema.rawSchema(), null); try { Integer version = schemaVersion(topic, isKey, context.getSchemaId(), context.getSubject(), schema, result); if (schema.rawSchema().getType().equals(Schema.Type.RECORD)) { return new GenericContainerWithVersion((GenericContainer) result, version); } else { return new GenericContainerWithVersion(new NonRecordContainer(schema.rawSchema(), result), version); } } catch (RestClientException | IOException e) { throw new SerializationException("Error retrieving Avro " + getSchemaType(isKey) + " schema version for id " + context.getSchemaId(), e); } } }
AbstractKafkaAvroDeserializer extends AbstractKafkaSchemaSerDe { protected GenericContainerWithVersion deserializeWithSchemaAndVersion( String topic, boolean isKey, byte[] payload) throws SerializationException { if (payload == null) { return null; } DeserializationContext context = new DeserializationContext(topic, isKey, payload); AvroSchema schema = context.schemaForDeserialize(); Object result = context.read(schema.rawSchema(), null); try { Integer version = schemaVersion(topic, isKey, context.getSchemaId(), context.getSubject(), schema, result); if (schema.rawSchema().getType().equals(Schema.Type.RECORD)) { return new GenericContainerWithVersion((GenericContainer) result, version); } else { return new GenericContainerWithVersion(new NonRecordContainer(schema.rawSchema(), result), version); } } catch (RestClientException | IOException e) { throw new SerializationException("Error retrieving Avro " + getSchemaType(isKey) + " schema version for id " + context.getSchemaId(), e); } }  }
AbstractKafkaAvroDeserializer extends AbstractKafkaSchemaSerDe { protected GenericContainerWithVersion deserializeWithSchemaAndVersion( String topic, boolean isKey, byte[] payload) throws SerializationException { if (payload == null) { return null; } DeserializationContext context = new DeserializationContext(topic, isKey, payload); AvroSchema schema = context.schemaForDeserialize(); Object result = context.read(schema.rawSchema(), null); try { Integer version = schemaVersion(topic, isKey, context.getSchemaId(), context.getSubject(), schema, result); if (schema.rawSchema().getType().equals(Schema.Type.RECORD)) { return new GenericContainerWithVersion((GenericContainer) result, version); } else { return new GenericContainerWithVersion(new NonRecordContainer(schema.rawSchema(), result), version); } } catch (RestClientException | IOException e) { throw new SerializationException("Error retrieving Avro " + getSchemaType(isKey) + " schema version for id " + context.getSchemaId(), e); } }   }
AbstractKafkaAvroDeserializer extends AbstractKafkaSchemaSerDe { protected GenericContainerWithVersion deserializeWithSchemaAndVersion( String topic, boolean isKey, byte[] payload) throws SerializationException { if (payload == null) { return null; } DeserializationContext context = new DeserializationContext(topic, isKey, payload); AvroSchema schema = context.schemaForDeserialize(); Object result = context.read(schema.rawSchema(), null); try { Integer version = schemaVersion(topic, isKey, context.getSchemaId(), context.getSubject(), schema, result); if (schema.rawSchema().getType().equals(Schema.Type.RECORD)) { return new GenericContainerWithVersion((GenericContainer) result, version); } else { return new GenericContainerWithVersion(new NonRecordContainer(schema.rawSchema(), result), version); } } catch (RestClientException | IOException e) { throw new SerializationException("Error retrieving Avro " + getSchemaType(isKey) + " schema version for id " + context.getSchemaId(), e); } }    }