@Test public void testSchemaCompatiability() { Schema readerSchema = new Schema.Parser().parse("{\"type\":\"record\",\"name\":\"GobblinTrackingEvent_GaaS2\",\"namespace\":\"gobblin.metrics\",\"fields\":[{\"name\":\"timestamp\",\"type\":\"long\",\"doc\":\"Time at which event was created.\",\"default\":0},{\"name\":\"namespace\",\"type\":[{\"type\":\"string\",\"avro.java.string\":\"String\"},\"null\"],\"doc\":\"Namespace used for filtering of events.\"},{\"name\":\"name\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"},\"doc\":\"Event name.\"},{\"name\":\"metadata\",\"type\":{\"type\":\"map\",\"values\":{\"type\":\"string\",\"avro.java.string\":\"String\"},\"avro.java.string\":\"String\"},\"doc\":\"Event metadata.\",\"default\":{}}]}"); Schema writerSchema1 = new Schema.Parser().parse("{\"type\":\"record\",\"name\":\"GobblinTrackingEvent\",\"namespace\":\"org.apache.gobblin.metrics\",\"fields\":[{\"name\":\"timestamp\",\"type\":\"long\",\"doc\":\"Time at which event was created.\",\"default\":0},{\"name\":\"namespace\",\"type\":[\"string\",\"null\"],\"doc\":\"Namespace used for filtering of events.\"},{\"name\":\"name\",\"type\":\"string\",\"doc\":\"Event name.\"},{\"name\":\"metadata\",\"type\":{\"type\":\"map\",\"values\":\"string\"},\"doc\":\"Event metadata.\",\"default\":{}}]}"); Schema writerSchema2 = new Schema.Parser().parse("{\"type\":\"record\",\"name\":\"GobblinTrackingEvent\",\"namespace\":\"org.apache.gobblin.metrics\",\"fields\":[{\"name\":\"timestamp\",\"type\":\"long\",\"doc\":\"Time at which event was created.\",\"default\":0},{\"name\":\"namespace\",\"type\":[\"string\",\"null\"],\"doc\":\"Namespace used for filtering of events.\"},{\"name\":\"name2\",\"type\":\"string\",\"doc\":\"Event name.\"},{\"name\":\"metadata\",\"type\":{\"type\":\"map\",\"values\":\"string\"},\"doc\":\"Event metadata.\",\"default\":{}}]}"); Assert.assertTrue(AvroUtils.checkReaderWriterCompatibility(readerSchema, writerSchema1, true)); Assert.assertFalse(AvroUtils.checkReaderWriterCompatibility(readerSchema, writerSchema1, false)); Assert.assertFalse(AvroUtils.checkReaderWriterCompatibility(readerSchema, writerSchema2, true)); }
public static boolean checkReaderWriterCompatibility(Schema readerSchema, Schema writerSchema, boolean ignoreNamespace) { if (ignoreNamespace) { List<Schema.Field> fields = deepCopySchemaFields(readerSchema); readerSchema = Schema.createRecord(writerSchema.getName(), writerSchema.getDoc(), writerSchema.getNamespace(), readerSchema.isError()); readerSchema.setFields(fields); } return SchemaCompatibility.checkReaderWriterCompatibility(readerSchema, writerSchema).getType().equals(SchemaCompatibility.SchemaCompatibilityType.COMPATIBLE); }
AvroUtils { public static boolean checkReaderWriterCompatibility(Schema readerSchema, Schema writerSchema, boolean ignoreNamespace) { if (ignoreNamespace) { List<Schema.Field> fields = deepCopySchemaFields(readerSchema); readerSchema = Schema.createRecord(writerSchema.getName(), writerSchema.getDoc(), writerSchema.getNamespace(), readerSchema.isError()); readerSchema.setFields(fields); } return SchemaCompatibility.checkReaderWriterCompatibility(readerSchema, writerSchema).getType().equals(SchemaCompatibility.SchemaCompatibilityType.COMPATIBLE); } }
AvroUtils { public static boolean checkReaderWriterCompatibility(Schema readerSchema, Schema writerSchema, boolean ignoreNamespace) { if (ignoreNamespace) { List<Schema.Field> fields = deepCopySchemaFields(readerSchema); readerSchema = Schema.createRecord(writerSchema.getName(), writerSchema.getDoc(), writerSchema.getNamespace(), readerSchema.isError()); readerSchema.setFields(fields); } return SchemaCompatibility.checkReaderWriterCompatibility(readerSchema, writerSchema).getType().equals(SchemaCompatibility.SchemaCompatibilityType.COMPATIBLE); }  }
AvroUtils { public static boolean checkReaderWriterCompatibility(Schema readerSchema, Schema writerSchema, boolean ignoreNamespace) { if (ignoreNamespace) { List<Schema.Field> fields = deepCopySchemaFields(readerSchema); readerSchema = Schema.createRecord(writerSchema.getName(), writerSchema.getDoc(), writerSchema.getNamespace(), readerSchema.isError()); readerSchema.setFields(fields); } return SchemaCompatibility.checkReaderWriterCompatibility(readerSchema, writerSchema).getType().equals(SchemaCompatibility.SchemaCompatibilityType.COMPATIBLE); }  static boolean checkReaderWriterCompatibility(Schema readerSchema, Schema writerSchema, boolean ignoreNamespace); static Schema addSchemaCreationTime(Schema inputSchema, Schema outputSchema); static String getSchemaCreationTime(Schema inputSchema); static Schema setSchemaCreationTime(Schema inputSchema, String creationTime); static List<Field> deepCopySchemaFields(Schema readerSchema); static void convertFieldToSchemaWithProps(Map<String,JsonNode> fieldProps, Schema targetSchemaObj); static Optional<Schema> getFieldSchema(Schema schema, String fieldLocation); static Optional<Field> getField(Schema schema, String fieldLocation); static Optional<Object> getFieldValue(GenericRecord record, String fieldLocation); static Map<String, Object> getMultiFieldValue(GenericRecord record, String fieldLocation); @SuppressWarnings("unchecked") static Map<String, String> toStringMap(Object map); static GenericRecord convertRecordSchema(GenericRecord record, Schema newSchema); static byte[] recordToByteArray(GenericRecord record); static Schema getSchemaFromDataFile(Path dataFile, FileSystem fs); static Schema parseSchemaFromFile(Path filePath, FileSystem fs); static void writeSchemaToFile(Schema schema, Path filePath, FileSystem fs, boolean overwrite); static void writeSchemaToFile(Schema schema, Path filePath, Path tempFilePath, FileSystem fs, boolean overwrite); static void writeSchemaToFile(Schema schema, Path filePath, FileSystem fs, boolean overwrite, FsPermission perm); static void writeSchemaToFile(Schema schema, Path filePath, Path tempFilePath, FileSystem fs, boolean overwrite,
      FsPermission perm); static Schema getDirectorySchema(Path directory, FileSystem fs, boolean latest); static Schema getDirectorySchema(Path directory, Configuration conf, boolean latest); static Schema nullifyFieldsForSchemaMerge(Schema oldSchema, Schema newSchema); static Optional<Schema> removeUncomparableFields(Schema schema); static Schema switchName(Schema schema, String newName); static Schema switchNamespace(Schema schema, Map<String, String> namespaceOverride); static Path serializeAsPath(GenericRecord record, boolean includeFieldNames, boolean replacePathSeparators); static String sanitizeSchemaString(String schemaString); static GenericRecord slowDeserializeGenericRecord(byte[] serializedRecord, Schema schema); static Schema decorateRecordSchema(Schema inputSchema, @Nonnull List<Field> fieldList); static GenericRecord decorateRecord(GenericRecord inputRecord, @Nonnull Map<String, Object> fieldMap,
      Schema outputSchema); static GenericRecord overrideNameAndNamespace(GenericRecord input, String nameOverride, Optional<Map<String, String>> namespaceOverride); static Schema overrideNameAndNamespace(Schema input, String nameOverride, Optional<Map<String, String>> namespaceOverride); static boolean isSchemaRecursive(Schema schema, Optional<Logger> logger); static Pair<Schema, List<SchemaEntry>> dropRecursiveFields(Schema schema); }
AvroUtils { public static boolean checkReaderWriterCompatibility(Schema readerSchema, Schema writerSchema, boolean ignoreNamespace) { if (ignoreNamespace) { List<Schema.Field> fields = deepCopySchemaFields(readerSchema); readerSchema = Schema.createRecord(writerSchema.getName(), writerSchema.getDoc(), writerSchema.getNamespace(), readerSchema.isError()); readerSchema.setFields(fields); } return SchemaCompatibility.checkReaderWriterCompatibility(readerSchema, writerSchema).getType().equals(SchemaCompatibility.SchemaCompatibilityType.COMPATIBLE); }  static boolean checkReaderWriterCompatibility(Schema readerSchema, Schema writerSchema, boolean ignoreNamespace); static Schema addSchemaCreationTime(Schema inputSchema, Schema outputSchema); static String getSchemaCreationTime(Schema inputSchema); static Schema setSchemaCreationTime(Schema inputSchema, String creationTime); static List<Field> deepCopySchemaFields(Schema readerSchema); static void convertFieldToSchemaWithProps(Map<String,JsonNode> fieldProps, Schema targetSchemaObj); static Optional<Schema> getFieldSchema(Schema schema, String fieldLocation); static Optional<Field> getField(Schema schema, String fieldLocation); static Optional<Object> getFieldValue(GenericRecord record, String fieldLocation); static Map<String, Object> getMultiFieldValue(GenericRecord record, String fieldLocation); @SuppressWarnings("unchecked") static Map<String, String> toStringMap(Object map); static GenericRecord convertRecordSchema(GenericRecord record, Schema newSchema); static byte[] recordToByteArray(GenericRecord record); static Schema getSchemaFromDataFile(Path dataFile, FileSystem fs); static Schema parseSchemaFromFile(Path filePath, FileSystem fs); static void writeSchemaToFile(Schema schema, Path filePath, FileSystem fs, boolean overwrite); static void writeSchemaToFile(Schema schema, Path filePath, Path tempFilePath, FileSystem fs, boolean overwrite); static void writeSchemaToFile(Schema schema, Path filePath, FileSystem fs, boolean overwrite, FsPermission perm); static void writeSchemaToFile(Schema schema, Path filePath, Path tempFilePath, FileSystem fs, boolean overwrite,
      FsPermission perm); static Schema getDirectorySchema(Path directory, FileSystem fs, boolean latest); static Schema getDirectorySchema(Path directory, Configuration conf, boolean latest); static Schema nullifyFieldsForSchemaMerge(Schema oldSchema, Schema newSchema); static Optional<Schema> removeUncomparableFields(Schema schema); static Schema switchName(Schema schema, String newName); static Schema switchNamespace(Schema schema, Map<String, String> namespaceOverride); static Path serializeAsPath(GenericRecord record, boolean includeFieldNames, boolean replacePathSeparators); static String sanitizeSchemaString(String schemaString); static GenericRecord slowDeserializeGenericRecord(byte[] serializedRecord, Schema schema); static Schema decorateRecordSchema(Schema inputSchema, @Nonnull List<Field> fieldList); static GenericRecord decorateRecord(GenericRecord inputRecord, @Nonnull Map<String, Object> fieldMap,
      Schema outputSchema); static GenericRecord overrideNameAndNamespace(GenericRecord input, String nameOverride, Optional<Map<String, String>> namespaceOverride); static Schema overrideNameAndNamespace(Schema input, String nameOverride, Optional<Map<String, String>> namespaceOverride); static boolean isSchemaRecursive(Schema schema, Optional<Logger> logger); static Pair<Schema, List<SchemaEntry>> dropRecursiveFields(Schema schema); static final String FIELD_LOCATION_DELIMITER; }