@Test public void testConvertsMetadataMultiOutput() throws DataConversionException { final int numRecordsToReturn = 2; DummyConverter dummyConverter = new DummyConverter(numRecordsToReturn); MetadataConverterWrapper<String, String, String, String> wrapper = new MetadataConverterWrapper<>(dummyConverter); Iterable<RecordWithMetadata<String>> records1 = wrapper.convertRecord("foo", new RecordWithMetadata<String>("bar", buildMetadata(1)), new WorkUnitState()); Iterable<RecordWithMetadata<String>> records2 = wrapper.convertRecord("foo", new RecordWithMetadata<String>("baz", buildMetadata(2)), new WorkUnitState()); Iterator<RecordWithMetadata<String>> record1It = records1.iterator(); Iterator<RecordWithMetadata<String>> record2It = records2.iterator(); for (int i = 0; i < numRecordsToReturn; i++) { RecordWithMetadata<String> record1 = record1It.next(); Assert.assertEquals(record1.getRecord(), "converted" + String.valueOf(i)); Assert.assertEquals(record1.getMetadata().getGlobalMetadata().getDatasetUrn(), "dataset-id:1"); RecordWithMetadata<String> record2 = record2It.next(); Assert.assertEquals(record2.getRecord(), "converted" + String.valueOf(i)); Assert.assertEquals(record2.getMetadata().getGlobalMetadata().getDatasetUrn(), "dataset-id:2"); } }
@Override public Iterable<RecordWithMetadata<DO>> convertRecord(SO outputSchema, Object inputRecord, WorkUnitState workUnit) throws DataConversionException { final Metadata metadata = getMetadataFromRecord(inputRecord); final DI innerRecord = getRecordFromRecord(inputRecord); Iterable<DO> outputRecords = innerConverter.convertRecord(outputSchema, innerRecord, workUnit); return Iterables.transform(outputRecords, new Function<DO, RecordWithMetadata<DO>>() { @Nullable @Override public RecordWithMetadata<DO> apply(@Nullable DO input) { return new RecordWithMetadata<>(input, metadata); } }); }
MetadataConverterWrapper extends Converter<SI, SO, Object, RecordWithMetadata<DO>> { @Override public Iterable<RecordWithMetadata<DO>> convertRecord(SO outputSchema, Object inputRecord, WorkUnitState workUnit) throws DataConversionException { final Metadata metadata = getMetadataFromRecord(inputRecord); final DI innerRecord = getRecordFromRecord(inputRecord); Iterable<DO> outputRecords = innerConverter.convertRecord(outputSchema, innerRecord, workUnit); return Iterables.transform(outputRecords, new Function<DO, RecordWithMetadata<DO>>() { @Nullable @Override public RecordWithMetadata<DO> apply(@Nullable DO input) { return new RecordWithMetadata<>(input, metadata); } }); } }
MetadataConverterWrapper extends Converter<SI, SO, Object, RecordWithMetadata<DO>> { @Override public Iterable<RecordWithMetadata<DO>> convertRecord(SO outputSchema, Object inputRecord, WorkUnitState workUnit) throws DataConversionException { final Metadata metadata = getMetadataFromRecord(inputRecord); final DI innerRecord = getRecordFromRecord(inputRecord); Iterable<DO> outputRecords = innerConverter.convertRecord(outputSchema, innerRecord, workUnit); return Iterables.transform(outputRecords, new Function<DO, RecordWithMetadata<DO>>() { @Nullable @Override public RecordWithMetadata<DO> apply(@Nullable DO input) { return new RecordWithMetadata<>(input, metadata); } }); } MetadataConverterWrapper(Converter<SI, SO, DI, DO> innerConverter); }
MetadataConverterWrapper extends Converter<SI, SO, Object, RecordWithMetadata<DO>> { @Override public Iterable<RecordWithMetadata<DO>> convertRecord(SO outputSchema, Object inputRecord, WorkUnitState workUnit) throws DataConversionException { final Metadata metadata = getMetadataFromRecord(inputRecord); final DI innerRecord = getRecordFromRecord(inputRecord); Iterable<DO> outputRecords = innerConverter.convertRecord(outputSchema, innerRecord, workUnit); return Iterables.transform(outputRecords, new Function<DO, RecordWithMetadata<DO>>() { @Nullable @Override public RecordWithMetadata<DO> apply(@Nullable DO input) { return new RecordWithMetadata<>(input, metadata); } }); } MetadataConverterWrapper(Converter<SI, SO, DI, DO> innerConverter); @Override Converter<SI, SO, Object, RecordWithMetadata<DO>> init(WorkUnitState workUnit); @Override void close(); @Override SO convertSchema(SI inputSchema, WorkUnitState workUnit); @Override Iterable<RecordWithMetadata<DO>> convertRecord(SO outputSchema, Object inputRecord, WorkUnitState workUnit); @Override State getFinalState(); }
MetadataConverterWrapper extends Converter<SI, SO, Object, RecordWithMetadata<DO>> { @Override public Iterable<RecordWithMetadata<DO>> convertRecord(SO outputSchema, Object inputRecord, WorkUnitState workUnit) throws DataConversionException { final Metadata metadata = getMetadataFromRecord(inputRecord); final DI innerRecord = getRecordFromRecord(inputRecord); Iterable<DO> outputRecords = innerConverter.convertRecord(outputSchema, innerRecord, workUnit); return Iterables.transform(outputRecords, new Function<DO, RecordWithMetadata<DO>>() { @Nullable @Override public RecordWithMetadata<DO> apply(@Nullable DO input) { return new RecordWithMetadata<>(input, metadata); } }); } MetadataConverterWrapper(Converter<SI, SO, DI, DO> innerConverter); @Override Converter<SI, SO, Object, RecordWithMetadata<DO>> init(WorkUnitState workUnit); @Override void close(); @Override SO convertSchema(SI inputSchema, WorkUnitState workUnit); @Override Iterable<RecordWithMetadata<DO>> convertRecord(SO outputSchema, Object inputRecord, WorkUnitState workUnit); @Override State getFinalState();  }