package com.linkedin.camus.sweeper.utils;

import java.io.IOException;

import org.apache.avro.Schema;
import org.apache.avro.mapred.AvroKey;
import org.apache.avro.mapreduce.AvroKeyOutputFormat;
import org.apache.avro.mapreduce.AvroKeyRecordWriter;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

/**
 * The AvroKeyOutputFormat class to relax the name validation
 * 
 * @author hcai
 *
 * @param <T> The Java type of the Avro data to serialize.
 */
public class RelaxedAvroKeyOutputFormat<T> extends AvroKeyOutputFormat<T> 
{
  private static final Log LOG =
    LogFactory.getLog(RelaxedAvroKeyOutputFormat.class.getName());

  private static final String CONF_OUTPUT_KEY_SCHEMA = "avro.schema.output.key";

  @Override
  public RecordWriter<AvroKey<T>, NullWritable> getRecordWriter(TaskAttemptContext context)
      throws IOException {
    LOG.info("getRecordWriter for" + context);
    // Get the writer schema.
    String schemaString = context.getConfiguration().get(CONF_OUTPUT_KEY_SCHEMA);
    Schema writerSchema  = schemaString != null ? RelaxedSchemaUtils.parseSchema(schemaString, context.getConfiguration()) : null;

    if (null == writerSchema) {
      throw new IOException(
          "AvroKeyOutputFormat requires an output schema. Use AvroJob.setOutputKeySchema().");
    }

    return new AvroKeyRecordWriter<T>(writerSchema, getCompressionCodec(context), getAvroFileOutputStream(context));

  }

}

@@ -96,16 +96,60 @@
 Schema;%0A
+import org.apache.avro.generic.GenericData;%0A
 import o
@@ -1567,16 +1567,35 @@
 rSchema,
+ GenericData.get(),
  getComp

Update the RelaxedAvroKeyOutputFormat based on Avro 1.7.7 method signature