class HortonworksAttributeSchemaReferenceStrategy implements SchemaAccessStrategy {

    @Override
    public RecordSchema getSchema(Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema) throws SchemaNotFoundException, IOException {
        final String schemaIdentifier = variables.get(SCHEMA_ID_ATTRIBUTE);
        final String schemaVersion = variables.get(SCHEMA_VERSION_ATTRIBUTE);
        final String schemaProtocol = variables.get(SCHEMA_PROTOCOL_VERSION_ATTRIBUTE);
        if (schemaIdentifier == null || schemaVersion == null || schemaProtocol == null) {
            throw new SchemaNotFoundException("Could not determine Schema for " + variables + " because it is missing one of the following three required attributes: "
                + SCHEMA_ID_ATTRIBUTE + ", " + SCHEMA_VERSION_ATTRIBUTE + ", " + SCHEMA_PROTOCOL_VERSION_ATTRIBUTE);
        }

        if (!isNumber(schemaProtocol)) {
            throw new SchemaNotFoundException("Could not determine Schema for " + variables + " because the " + SCHEMA_PROTOCOL_VERSION_ATTRIBUTE + " has a value of '"
                + schemaProtocol + "', which is not a valid Protocol Version number");
        }

        final int protocol = Integer.parseInt(schemaProtocol);
        if (protocol != 1) {
            throw new SchemaNotFoundException("Could not determine Schema for " + variables + " because the " + SCHEMA_PROTOCOL_VERSION_ATTRIBUTE + " has a value of '"
                + schemaProtocol + "', which is not a valid Protocol Version number. Expected Protocol Version to be 1.");
        }

        if (!isNumber(schemaIdentifier)) {
            throw new SchemaNotFoundException("Could not determine Schema for " + variables + " because the " + SCHEMA_ID_ATTRIBUTE + " has a value of '"
                + schemaProtocol + "', which is not a valid Schema Identifier number");
        }

        if (!isNumber(schemaVersion)) {
            throw new SchemaNotFoundException("Could not determine Schema for " + variables + " because the " + SCHEMA_VERSION_ATTRIBUTE + " has a value of '"
                + schemaProtocol + "', which is not a valid Schema Version number");
        }

        final long schemaId = Long.parseLong(schemaIdentifier);
        final int version = Integer.parseInt(schemaVersion);

        final SchemaIdentifier identifier = SchemaIdentifier.builder().id(schemaId).version(version).build();
        final RecordSchema schema = schemaRegistry.retrieveSchema(identifier);
        if (schema == null) {
            throw new SchemaNotFoundException("Could not find a Schema in the Schema Registry with Schema Identifier '" + schemaId + "' and Version '" + version + "'");
        }

        return schema;
    }

    public  HortonworksAttributeSchemaReferenceStrategy(final SchemaRegistry schemaRegistry);

}