class ConfluentSchemaRegistryStrategy implements SchemaAccessStrategy {

    @Override
    public RecordSchema getSchema(final Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema) throws SchemaNotFoundException, IOException {
        final byte[] buffer = new byte[5];
        try {
            StreamUtils.fillBuffer(contentStream, buffer);
        } catch (final IOException ioe) {
            throw new SchemaNotFoundException("Could not read first 5 bytes from stream", ioe);
        }

        // This encoding follows the pattern that is provided for serializing data by the Confluent Schema Registry serializer
        // as it is provided at:
        // http://docs.confluent.io/current/schema-registry/docs/serializer-formatter.html#wire-format
        // The format consists of the first byte always being 0, to indicate a 'magic byte' followed by 4 bytes
        // representing the schema id.
        final ByteBuffer bb = ByteBuffer.wrap(buffer);
        final int magicByte = bb.get();
        if (magicByte != 0) {
            throw new SchemaNotFoundException("Schema Encoding appears to be of an incompatible version. "
                + "Expected stream to begin with a Magic Byte of 0 but first byte was " + magicByte);
        }

        final int schemaId = bb.getInt();

        final SchemaIdentifier schemaIdentifier = SchemaIdentifier.builder()
                .id(Long.valueOf(schemaId))
                .version(1)
                .build();

        return schemaRegistry.retrieveSchema(schemaIdentifier);
    }

    public  ConfluentSchemaRegistryStrategy(final SchemaRegistry schemaRegistry);

}