@Test public void testNameAndBlankBranch() throws SchemaNotFoundException, IOException { final PropertyValue nameValue = new MockPropertyValue("person"); final PropertyValue branchValue = new MockPropertyValue(" "); final PropertyValue versionValue = new MockPropertyValue(null); final SchemaNamePropertyStrategy schemaNamePropertyStrategy = new SchemaNamePropertyStrategy( schemaRegistry, nameValue, branchValue, versionValue); final SchemaIdentifier expectedSchemaIdentifier = SchemaIdentifier.builder() .name(nameValue.getValue()) .build(); when(schemaRegistry.retrieveSchema(argThat(new SchemaIdentifierMatcher(expectedSchemaIdentifier)))) .thenReturn(recordSchema); final RecordSchema retrievedSchema = schemaNamePropertyStrategy.getSchema(Collections.emptyMap(), null, recordSchema); assertNotNull(retrievedSchema); }
@Override public RecordSchema getSchema(final Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema) throws SchemaNotFoundException { final String schemaName = schemaNamePropertyValue.evaluateAttributeExpressions(variables).getValue(); if (schemaName.trim().isEmpty()) { throw new SchemaNotFoundException(String.format("%s did not provide appropriate Schema Name", schemaNamePropertyValue)); } try { final String schemaBranchName = schemaBranchNamePropertyValue.evaluateAttributeExpressions(variables).getValue(); final String schemaVersion = schemaVersionPropertyValue.evaluateAttributeExpressions(variables).getValue(); final SchemaIdentifier.Builder identifierBuilder = SchemaIdentifier.builder(); identifierBuilder.name(schemaName); if (!StringUtils.isBlank(schemaBranchName)) { identifierBuilder.branch(schemaBranchName); } if (!StringUtils.isBlank(schemaVersion)) { try { identifierBuilder.version(Integer.valueOf(schemaVersion)); } catch (NumberFormatException nfe) { throw new SchemaNotFoundException("Could not retrieve schema with name '" + schemaName + "' because a non-numeric version was supplied '" + schemaVersion + "'", nfe); } } final RecordSchema recordSchema = schemaRegistry.retrieveSchema(identifierBuilder.build()); if (recordSchema == null) { throw new SchemaNotFoundException("Could not find a schema with name '" + schemaName + "' in the configured Schema Registry"); } return recordSchema; } catch (final SchemaNotFoundException snf) { throw snf; } catch (final Exception e) { throw new SchemaNotFoundException("Could not retrieve schema with name '" + schemaName + "' from the configured Schema Registry", e); } }
SchemaNamePropertyStrategy implements SchemaAccessStrategy { @Override public RecordSchema getSchema(final Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema) throws SchemaNotFoundException { final String schemaName = schemaNamePropertyValue.evaluateAttributeExpressions(variables).getValue(); if (schemaName.trim().isEmpty()) { throw new SchemaNotFoundException(String.format("%s did not provide appropriate Schema Name", schemaNamePropertyValue)); } try { final String schemaBranchName = schemaBranchNamePropertyValue.evaluateAttributeExpressions(variables).getValue(); final String schemaVersion = schemaVersionPropertyValue.evaluateAttributeExpressions(variables).getValue(); final SchemaIdentifier.Builder identifierBuilder = SchemaIdentifier.builder(); identifierBuilder.name(schemaName); if (!StringUtils.isBlank(schemaBranchName)) { identifierBuilder.branch(schemaBranchName); } if (!StringUtils.isBlank(schemaVersion)) { try { identifierBuilder.version(Integer.valueOf(schemaVersion)); } catch (NumberFormatException nfe) { throw new SchemaNotFoundException("Could not retrieve schema with name '" + schemaName + "' because a non-numeric version was supplied '" + schemaVersion + "'", nfe); } } final RecordSchema recordSchema = schemaRegistry.retrieveSchema(identifierBuilder.build()); if (recordSchema == null) { throw new SchemaNotFoundException("Could not find a schema with name '" + schemaName + "' in the configured Schema Registry"); } return recordSchema; } catch (final SchemaNotFoundException snf) { throw snf; } catch (final Exception e) { throw new SchemaNotFoundException("Could not retrieve schema with name '" + schemaName + "' from the configured Schema Registry", e); } } }
SchemaNamePropertyStrategy implements SchemaAccessStrategy { @Override public RecordSchema getSchema(final Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema) throws SchemaNotFoundException { final String schemaName = schemaNamePropertyValue.evaluateAttributeExpressions(variables).getValue(); if (schemaName.trim().isEmpty()) { throw new SchemaNotFoundException(String.format("%s did not provide appropriate Schema Name", schemaNamePropertyValue)); } try { final String schemaBranchName = schemaBranchNamePropertyValue.evaluateAttributeExpressions(variables).getValue(); final String schemaVersion = schemaVersionPropertyValue.evaluateAttributeExpressions(variables).getValue(); final SchemaIdentifier.Builder identifierBuilder = SchemaIdentifier.builder(); identifierBuilder.name(schemaName); if (!StringUtils.isBlank(schemaBranchName)) { identifierBuilder.branch(schemaBranchName); } if (!StringUtils.isBlank(schemaVersion)) { try { identifierBuilder.version(Integer.valueOf(schemaVersion)); } catch (NumberFormatException nfe) { throw new SchemaNotFoundException("Could not retrieve schema with name '" + schemaName + "' because a non-numeric version was supplied '" + schemaVersion + "'", nfe); } } final RecordSchema recordSchema = schemaRegistry.retrieveSchema(identifierBuilder.build()); if (recordSchema == null) { throw new SchemaNotFoundException("Could not find a schema with name '" + schemaName + "' in the configured Schema Registry"); } return recordSchema; } catch (final SchemaNotFoundException snf) { throw snf; } catch (final Exception e) { throw new SchemaNotFoundException("Could not retrieve schema with name '" + schemaName + "' from the configured Schema Registry", e); } } SchemaNamePropertyStrategy(final SchemaRegistry schemaRegistry,
                                      final PropertyValue schemaNamePropertyValue,
                                      final PropertyValue schemaBranchNamePropertyValue,
                                      final PropertyValue schemaVersionPropertyValue); }
SchemaNamePropertyStrategy implements SchemaAccessStrategy { @Override public RecordSchema getSchema(final Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema) throws SchemaNotFoundException { final String schemaName = schemaNamePropertyValue.evaluateAttributeExpressions(variables).getValue(); if (schemaName.trim().isEmpty()) { throw new SchemaNotFoundException(String.format("%s did not provide appropriate Schema Name", schemaNamePropertyValue)); } try { final String schemaBranchName = schemaBranchNamePropertyValue.evaluateAttributeExpressions(variables).getValue(); final String schemaVersion = schemaVersionPropertyValue.evaluateAttributeExpressions(variables).getValue(); final SchemaIdentifier.Builder identifierBuilder = SchemaIdentifier.builder(); identifierBuilder.name(schemaName); if (!StringUtils.isBlank(schemaBranchName)) { identifierBuilder.branch(schemaBranchName); } if (!StringUtils.isBlank(schemaVersion)) { try { identifierBuilder.version(Integer.valueOf(schemaVersion)); } catch (NumberFormatException nfe) { throw new SchemaNotFoundException("Could not retrieve schema with name '" + schemaName + "' because a non-numeric version was supplied '" + schemaVersion + "'", nfe); } } final RecordSchema recordSchema = schemaRegistry.retrieveSchema(identifierBuilder.build()); if (recordSchema == null) { throw new SchemaNotFoundException("Could not find a schema with name '" + schemaName + "' in the configured Schema Registry"); } return recordSchema; } catch (final SchemaNotFoundException snf) { throw snf; } catch (final Exception e) { throw new SchemaNotFoundException("Could not retrieve schema with name '" + schemaName + "' from the configured Schema Registry", e); } } SchemaNamePropertyStrategy(final SchemaRegistry schemaRegistry,
                                      final PropertyValue schemaNamePropertyValue,
                                      final PropertyValue schemaBranchNamePropertyValue,
                                      final PropertyValue schemaVersionPropertyValue); @Override RecordSchema getSchema(final Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema); @Override Set<SchemaField> getSuppliedSchemaFields(); }
SchemaNamePropertyStrategy implements SchemaAccessStrategy { @Override public RecordSchema getSchema(final Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema) throws SchemaNotFoundException { final String schemaName = schemaNamePropertyValue.evaluateAttributeExpressions(variables).getValue(); if (schemaName.trim().isEmpty()) { throw new SchemaNotFoundException(String.format("%s did not provide appropriate Schema Name", schemaNamePropertyValue)); } try { final String schemaBranchName = schemaBranchNamePropertyValue.evaluateAttributeExpressions(variables).getValue(); final String schemaVersion = schemaVersionPropertyValue.evaluateAttributeExpressions(variables).getValue(); final SchemaIdentifier.Builder identifierBuilder = SchemaIdentifier.builder(); identifierBuilder.name(schemaName); if (!StringUtils.isBlank(schemaBranchName)) { identifierBuilder.branch(schemaBranchName); } if (!StringUtils.isBlank(schemaVersion)) { try { identifierBuilder.version(Integer.valueOf(schemaVersion)); } catch (NumberFormatException nfe) { throw new SchemaNotFoundException("Could not retrieve schema with name '" + schemaName + "' because a non-numeric version was supplied '" + schemaVersion + "'", nfe); } } final RecordSchema recordSchema = schemaRegistry.retrieveSchema(identifierBuilder.build()); if (recordSchema == null) { throw new SchemaNotFoundException("Could not find a schema with name '" + schemaName + "' in the configured Schema Registry"); } return recordSchema; } catch (final SchemaNotFoundException snf) { throw snf; } catch (final Exception e) { throw new SchemaNotFoundException("Could not retrieve schema with name '" + schemaName + "' from the configured Schema Registry", e); } } SchemaNamePropertyStrategy(final SchemaRegistry schemaRegistry,
                                      final PropertyValue schemaNamePropertyValue,
                                      final PropertyValue schemaBranchNamePropertyValue,
                                      final PropertyValue schemaVersionPropertyValue); @Override RecordSchema getSchema(final Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema); @Override Set<SchemaField> getSuppliedSchemaFields();  }