SupportedFieldTypesValidation implements Validation { @Override public ValidationResult validate(Config config) { for (StructField field : ComponentFactory.create( Schema.class, config.getConfig(this.path), true).getSchema().fields()) { boolean decimalMatch = (field.dataType() instanceof DecimalType && validationTypes.contains(new DecimalType())); if (!validationTypes.contains(field.dataType()) && !decimalMatch) { return new ValidationResult(this, Validity.INVALID, "Schema field type " + field.dataType().simpleString() + " is not supported by this component type."); } } return new ValidationResult(this, Validity.VALID, "Schema field types are valid for this component type."); } SupportedFieldTypesValidation(String path, Set<DataType> validTypes); @Override ValidationResult validate(Config config); @Override Set<String> getKnownPaths();  }
@Test public void testValid() { SupportedFieldTypesValidation v = new SupportedFieldTypesValidation(SCHEMA_CONFIG, new HashSet<DataType>(Arrays.asList(DataTypes.IntegerType, DataTypes.StringType, new DecimalType(), DataTypes.TimestampType, DataTypes.BinaryType))); ValidationResult vr = v.validate(config); assertEquals(vr.getValidity(), Validity.VALID); }
@Test public void testInvalid() { SupportedFieldTypesValidation v = new SupportedFieldTypesValidation(SCHEMA_CONFIG, new HashSet<DataType>(Arrays.asList(DataTypes.IntegerType, new DecimalType(), DataTypes.TimestampType, DataTypes.BinaryType))); ValidationResult vr = v.validate(config); assertEquals(vr.getValidity(), Validity.INVALID); }