@Test public void testConversion() throws IOException { File inputFile = generateRecord(); WorkUnitState workUnitState = new WorkUnitState(); Schema inputSchema = new Schema.Parser().parse(getClass().getResourceAsStream("/converter/recursive.avsc")); GenericDatumReader<GenericRecord> datumReader = new GenericDatumReader<GenericRecord>(inputSchema); DataFileReader<GenericRecord> dataFileReader = new DataFileReader<GenericRecord>(inputFile, datumReader); GenericRecord inputRecord = dataFileReader.next(); AvroRecursionEliminatingConverter converter = new AvroRecursionEliminatingConverter(); Schema outputSchema = null; String recursiveFieldPath = "address.previous_address"; Assert.assertTrue(AvroUtils.getFieldSchema(inputSchema, recursiveFieldPath).isPresent()); try { outputSchema = converter.convertSchema(inputSchema, workUnitState); Assert.assertTrue(!AvroUtils.getFieldSchema(outputSchema, recursiveFieldPath).isPresent(), "Inner recursive field " + recursiveFieldPath + " should not be in output schema"); } catch (SchemaConversionException e) { Assert.fail(e.getMessage()); } GenericRecord outputRecord = null; try { outputRecord = converter.convertRecord(outputSchema, inputRecord, workUnitState).iterator().next(); } catch (DataConversionException e) { Assert.fail(e.getMessage()); } checkEquality("address.street_number", inputRecord, 1234, "Different value in input"); checkEquality("address.street_number", outputRecord, 1234, "Different value in output"); checkEquality("name", inputRecord, new Utf8("John"), "Different value in input"); checkEquality("name", outputRecord, new Utf8("John"), "Different value in output"); checkEquality("address.previous_address.city", inputRecord, new Utf8("San Francisco"), "Different value in input"); checkEquality("address.previous_address", outputRecord, null, "Failed to remove recursive field"); }
@Override public Schema convertSchema(Schema inputSchema, WorkUnitState workUnit) throws SchemaConversionException { Pair<Schema, List<AvroUtils.SchemaEntry>> results = AvroUtils.dropRecursiveFields(inputSchema); List<AvroUtils.SchemaEntry> recursiveFields = results.getSecond(); if (!recursiveFields.isEmpty()) { log.warn("Schema {} is recursive. Will drop fields [{}]", inputSchema.getFullName(), recursiveFields.stream().map(entry -> entry.getFieldName()).collect(Collectors.joining(","))); log.debug("Projected Schema = {}", results.getFirst()); } return results.getFirst(); }
AvroRecursionEliminatingConverter extends AvroToAvroConverterBase { @Override public Schema convertSchema(Schema inputSchema, WorkUnitState workUnit) throws SchemaConversionException { Pair<Schema, List<AvroUtils.SchemaEntry>> results = AvroUtils.dropRecursiveFields(inputSchema); List<AvroUtils.SchemaEntry> recursiveFields = results.getSecond(); if (!recursiveFields.isEmpty()) { log.warn("Schema {} is recursive. Will drop fields [{}]", inputSchema.getFullName(), recursiveFields.stream().map(entry -> entry.getFieldName()).collect(Collectors.joining(","))); log.debug("Projected Schema = {}", results.getFirst()); } return results.getFirst(); } }
AvroRecursionEliminatingConverter extends AvroToAvroConverterBase { @Override public Schema convertSchema(Schema inputSchema, WorkUnitState workUnit) throws SchemaConversionException { Pair<Schema, List<AvroUtils.SchemaEntry>> results = AvroUtils.dropRecursiveFields(inputSchema); List<AvroUtils.SchemaEntry> recursiveFields = results.getSecond(); if (!recursiveFields.isEmpty()) { log.warn("Schema {} is recursive. Will drop fields [{}]", inputSchema.getFullName(), recursiveFields.stream().map(entry -> entry.getFieldName()).collect(Collectors.joining(","))); log.debug("Projected Schema = {}", results.getFirst()); } return results.getFirst(); }  }
AvroRecursionEliminatingConverter extends AvroToAvroConverterBase { @Override public Schema convertSchema(Schema inputSchema, WorkUnitState workUnit) throws SchemaConversionException { Pair<Schema, List<AvroUtils.SchemaEntry>> results = AvroUtils.dropRecursiveFields(inputSchema); List<AvroUtils.SchemaEntry> recursiveFields = results.getSecond(); if (!recursiveFields.isEmpty()) { log.warn("Schema {} is recursive. Will drop fields [{}]", inputSchema.getFullName(), recursiveFields.stream().map(entry -> entry.getFieldName()).collect(Collectors.joining(","))); log.debug("Projected Schema = {}", results.getFirst()); } return results.getFirst(); }  @Override Schema convertSchema(Schema inputSchema, WorkUnitState workUnit); @Override Iterable<GenericRecord> convertRecordImpl(Schema outputSchema, GenericRecord inputRecord,
      WorkUnitState workUnit); }
AvroRecursionEliminatingConverter extends AvroToAvroConverterBase { @Override public Schema convertSchema(Schema inputSchema, WorkUnitState workUnit) throws SchemaConversionException { Pair<Schema, List<AvroUtils.SchemaEntry>> results = AvroUtils.dropRecursiveFields(inputSchema); List<AvroUtils.SchemaEntry> recursiveFields = results.getSecond(); if (!recursiveFields.isEmpty()) { log.warn("Schema {} is recursive. Will drop fields [{}]", inputSchema.getFullName(), recursiveFields.stream().map(entry -> entry.getFieldName()).collect(Collectors.joining(","))); log.debug("Projected Schema = {}", results.getFirst()); } return results.getFirst(); }  @Override Schema convertSchema(Schema inputSchema, WorkUnitState workUnit); @Override Iterable<GenericRecord> convertRecordImpl(Schema outputSchema, GenericRecord inputRecord,
      WorkUnitState workUnit);  }