class AvroRecordConverter {

    public Record convert(Record input) throws AvroConversionException {
        Record result = new Record(outputSchema);
        for (Field outputField : outputSchema.getFields()) {
            // Default to matching by name
            String inputFieldName = outputField.name();
            if (fieldMapping.containsKey(outputField.name())) {
                inputFieldName = fieldMapping.get(outputField.name());
            }

            IndexedRecord currentRecord = input;
            Schema currentSchema = getNonNullSchema(inputSchema);
            while (inputFieldName.contains(".")) {
                // Recurse down the schema to find the right field.
                int dotIndex = inputFieldName.indexOf('.');
                String entityName = inputFieldName.substring(0, dotIndex);
                // Get the record object
                Object innerRecord = currentRecord.get(currentSchema.getField(
                        entityName).pos());
                if (innerRecord == null) {
                    // Probably hit a null record here. Just break out of the
                    // loop so that null object will be passed to convertData
                    // below.
                    currentRecord = null;
                    break;
                }
                if (innerRecord != null
                        && !(innerRecord instanceof IndexedRecord)) {
                    throw new AvroConversionException(inputFieldName
                            + " stepped through a non-record");
                }
                currentRecord = (IndexedRecord) innerRecord;

                // Get the schema. In case we had an optional record, choose
                // just the record.
                currentSchema = currentSchema.getField(entityName).schema();
                currentSchema = getNonNullSchema(currentSchema);
                inputFieldName = inputFieldName.substring(dotIndex + 1);
            }

            // Current should now be in the right place to read the record.
            Field f = currentSchema.getField(inputFieldName);
            if (currentRecord == null) {
                // We may have stepped into a null union type and gotten a null
                // result.
                Schema s = null;
                if (f != null) {
                    s = f.schema();
                }
                result.put(outputField.name(),
                        convertData(null, s, outputField.schema()));
            } else {
                result.put(
                        outputField.name(),
                        convertData(currentRecord.get(f.pos()), f.schema(),
                                outputField.schema()));
            }
        }
        return result;
    }

    public  AvroRecordConverter(Schema inputSchema, Schema outputSchema,
            Map<String, String> fieldMapping);
    public  AvroRecordConverter(Schema inputSchema, Schema outputSchema,
            Map<String, String> fieldMapping, Locale locale);

}