ConnectSchema implements Schema { public static void validateValue(Schema schema, Object value) { if (value == null) { if (!schema.isOptional()) throw new DataException("Invalid value: null used for required field"); else return; } List<Class> expectedClasses = LOGICAL_TYPE_CLASSES.get(schema.name()); if (expectedClasses == null) expectedClasses = SCHEMA_TYPE_CLASSES.get(schema.type()); if (expectedClasses == null) throw new DataException("Invalid Java object for schema type " + schema.type() + ": " + value.getClass()); boolean foundMatch = false; for (Class<?> expectedClass : expectedClasses) { if (expectedClass.isInstance(value)) { foundMatch = true; break; } } if (!foundMatch) throw new DataException("Invalid Java object for schema type " + schema.type() + ": " + value.getClass()); switch (schema.type()) { case STRUCT: Struct struct = (Struct) value; if (!struct.schema().equals(schema)) throw new DataException("Struct schemas do not match."); struct.validate(); break; case ARRAY: List<?> array = (List<?>) value; for (Object entry : array) validateValue(schema.valueSchema(), entry); break; case MAP: Map<?, ?> map = (Map<?, ?>) value; for (Map.Entry<?, ?> entry : map.entrySet()) { validateValue(schema.keySchema(), entry.getKey()); validateValue(schema.valueSchema(), entry.getValue()); } break; } } ConnectSchema(Type type, boolean optional, Object defaultValue, String name, Integer version, String doc, Map<String, String> parameters, List<Field> fields, Schema keySchema, Schema valueSchema);  ConnectSchema(Type type, boolean optional, Object defaultValue, String name, Integer version, String doc);  ConnectSchema(Type type); @Override Type type(); @Override boolean isOptional(); @Override Object defaultValue(); @Override String name(); @Override Integer version(); @Override String doc(); @Override Map<String, String> parameters(); @Override List<Field> fields(); Field field(String fieldName); @Override Schema keySchema(); @Override Schema valueSchema(); static void validateValue(Schema schema, Object value); void validateValue(Object value); @Override ConnectSchema schema(); @Override boolean equals(Object o); @Override int hashCode(); @Override String toString(); static Type schemaType(Class<?> klass);  }
@Test(expected = DataException.class) public void testValidateValueMismatchInt8() { ConnectSchema.validateValue(Schema.INT8_SCHEMA, 1); }
@Test(expected = DataException.class) public void testValidateValueMismatchInt16() { ConnectSchema.validateValue(Schema.INT16_SCHEMA, 1); }
@Test(expected = DataException.class) public void testValidateValueMismatchInt32() { ConnectSchema.validateValue(Schema.INT32_SCHEMA, (long) 1); }
@Test(expected = DataException.class) public void testValidateValueMismatchInt64() { ConnectSchema.validateValue(Schema.INT64_SCHEMA, 1); }
@Test(expected = DataException.class) public void testValidateValueMismatchFloat() { ConnectSchema.validateValue(Schema.FLOAT32_SCHEMA, 1.0); }
@Test(expected = DataException.class) public void testValidateValueMismatchDouble() { ConnectSchema.validateValue(Schema.FLOAT64_SCHEMA, 1.f); }
@Test(expected = DataException.class) public void testValidateValueMismatchBoolean() { ConnectSchema.validateValue(Schema.BOOLEAN_SCHEMA, 1.f); }
@Test(expected = DataException.class) public void testValidateValueMismatchString() { CharBuffer cbuf = CharBuffer.wrap("abc"); ConnectSchema.validateValue(Schema.STRING_SCHEMA, cbuf); }
@Test(expected = DataException.class) public void testValidateValueMismatchBytes() { ConnectSchema.validateValue(Schema.BYTES_SCHEMA, new Object[]{1, "foo"}); }
@Test(expected = DataException.class) public void testValidateValueMismatchArray() { ConnectSchema.validateValue(SchemaBuilder.array(Schema.INT32_SCHEMA).build(), Arrays.asList("a", "b", "c")); }
@Test(expected = DataException.class) public void testValidateValueMismatchArraySomeMatch() { ConnectSchema.validateValue(SchemaBuilder.array(Schema.INT32_SCHEMA).build(), Arrays.asList(1, 2, "c")); }
@Test(expected = DataException.class) public void testValidateValueMismatchMapKey() { ConnectSchema.validateValue(MAP_INT_STRING_SCHEMA, Collections.singletonMap("wrong key type", "value")); }
@Test(expected = DataException.class) public void testValidateValueMismatchMapValue() { ConnectSchema.validateValue(MAP_INT_STRING_SCHEMA, Collections.singletonMap(1, 2)); }
@Test(expected = DataException.class) public void testValidateValueMismatchMapSomeKeys() { Map<Object, String> data = new HashMap<>(); data.put(1, "abc"); data.put("wrong", "it's as easy as one two three"); ConnectSchema.validateValue(MAP_INT_STRING_SCHEMA, data); }
@Test(expected = DataException.class) public void testValidateValueMismatchMapSomeValues() { Map<Integer, Object> data = new HashMap<>(); data.put(1, "abc"); data.put(2, "wrong".getBytes()); ConnectSchema.validateValue(MAP_INT_STRING_SCHEMA, data); }
@Test(expected = DataException.class) public void testValidateValueMismatchDate() { ConnectSchema.validateValue(Date.SCHEMA, 1000L); }
@Test(expected = DataException.class) public void testValidateValueMismatchTime() { ConnectSchema.validateValue(Time.SCHEMA, 1000L); }
@Test(expected = DataException.class) public void testValidateValueMismatchTimestamp() { ConnectSchema.validateValue(Timestamp.SCHEMA, 1000L); }
@Test public void testValidateValueMatchingType() { ConnectSchema.validateValue(Schema.INT8_SCHEMA, (byte) 1); ConnectSchema.validateValue(Schema.INT16_SCHEMA, (short) 1); ConnectSchema.validateValue(Schema.INT32_SCHEMA, 1); ConnectSchema.validateValue(Schema.INT64_SCHEMA, (long) 1); ConnectSchema.validateValue(Schema.FLOAT32_SCHEMA, 1.f); ConnectSchema.validateValue(Schema.FLOAT64_SCHEMA, 1.); ConnectSchema.validateValue(Schema.BOOLEAN_SCHEMA, true); ConnectSchema.validateValue(Schema.STRING_SCHEMA, "a string"); ConnectSchema.validateValue(Schema.BYTES_SCHEMA, "a byte array".getBytes()); ConnectSchema.validateValue(Schema.BYTES_SCHEMA, ByteBuffer.wrap("a byte array".getBytes())); ConnectSchema.validateValue(SchemaBuilder.array(Schema.INT32_SCHEMA).build(), Arrays.asList(1, 2, 3)); ConnectSchema.validateValue( SchemaBuilder.map(Schema.INT32_SCHEMA, Schema.STRING_SCHEMA).build(), Collections.singletonMap(1, "value") ); Struct structValue = new Struct(STRUCT_SCHEMA) .put("first", 1) .put("second", "foo") .put("array", Arrays.asList(1, 2, 3)) .put("map", Collections.singletonMap(1, "value")) .put("nested", new Struct(FLAT_STRUCT_SCHEMA).put("field", 12)); ConnectSchema.validateValue(STRUCT_SCHEMA, structValue); }