@Test public void testGetSchemaWithValidAttributes() throws IOException, SchemaNotFoundException { final long schemaId = 123456; final int version = 2; final int protocol = 1; final Map<String,String> attributes = new HashMap<>(); attributes.put(HortonworksAttributeSchemaReferenceStrategy.SCHEMA_ID_ATTRIBUTE, String.valueOf(schemaId)); attributes.put(HortonworksAttributeSchemaReferenceStrategy.SCHEMA_VERSION_ATTRIBUTE, String.valueOf(version)); attributes.put(HortonworksAttributeSchemaReferenceStrategy.SCHEMA_PROTOCOL_VERSION_ATTRIBUTE, String.valueOf(protocol)); final SchemaAccessStrategy schemaAccessStrategy = new HortonworksAttributeSchemaReferenceStrategy(schemaRegistry); final SchemaIdentifier expectedSchemaIdentifier = SchemaIdentifier.builder() .id(schemaId) .version(version) .build(); when(schemaRegistry.retrieveSchema(argThat(new SchemaIdentifierMatcher(expectedSchemaIdentifier)))) .thenReturn(recordSchema); final RecordSchema retrievedSchema = schemaAccessStrategy.getSchema(attributes, null, recordSchema); assertNotNull(retrievedSchema); }
@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; }
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; } }
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; } HortonworksAttributeSchemaReferenceStrategy(final SchemaRegistry schemaRegistry); }
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; } HortonworksAttributeSchemaReferenceStrategy(final SchemaRegistry schemaRegistry); boolean isFlowFileRequired(); @Override RecordSchema getSchema(Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema); @Override Set<SchemaField> getSuppliedSchemaFields(); }
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; } HortonworksAttributeSchemaReferenceStrategy(final SchemaRegistry schemaRegistry); boolean isFlowFileRequired(); @Override RecordSchema getSchema(Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema); @Override Set<SchemaField> getSuppliedSchemaFields(); static final String SCHEMA_ID_ATTRIBUTE; static final String SCHEMA_VERSION_ATTRIBUTE; static final String SCHEMA_PROTOCOL_VERSION_ATTRIBUTE; }