package com.twitter.elephantbird.mapreduce.output;

import java.io.IOException;

import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import com.twitter.elephantbird.mapreduce.input.BinaryConverterProvider;
import com.twitter.elephantbird.mapreduce.io.BinaryBlockWriter;
import com.twitter.elephantbird.mapreduce.io.BinaryConverter;
import com.twitter.elephantbird.mapreduce.io.GenericWritable;
import com.twitter.elephantbird.util.HadoopCompat;
import com.twitter.elephantbird.util.HadoopUtils;

import org.apache.hadoop.conf.Configuration;

/**
 * Generic OutputFormat for records to be stored as lzo-compressed protobuf blocks.
 */
public class LzoGenericBlockOutputFormat<M> extends LzoOutputFormat<M, GenericWritable<M>> {

  private static String CLASS_CONF_KEY = "elephantbird.class.for.LzoGenericBlockOutputFormat";
  private static String GENERIC_ENCODER_KEY = "elephantbird.encoder.class.for.LzoGenericBlockOutputFormat";

  public static void setGenericConverterClassConf(Class<?> clazz, Configuration conf) {
    HadoopUtils.setClassConf(conf, GENERIC_ENCODER_KEY, clazz);
  }

  public static void setClassConf(Class<?> clazz, Configuration conf) {
    HadoopUtils.setClassConf(conf, CLASS_CONF_KEY, clazz);
  }

  @Override
  public RecordWriter<M, GenericWritable<M>> getRecordWriter(TaskAttemptContext job)
      throws IOException, InterruptedException {
    Configuration conf = HadoopCompat.getConfiguration(job);
    String encoderClassName = conf.get(GENERIC_ENCODER_KEY);
    Class<?> typeRef = null;
    Class<?> encoderClazz = null;
    BinaryConverterProvider<?> converterProvider = null;
    // get the converter provider from job conf
    // which then gives us the BinaryConverter for the type M in question
    try {
      String typeRefClass = conf.get(CLASS_CONF_KEY);
      typeRef = conf.getClassByName(typeRefClass);
      encoderClazz = conf.getClassByName(encoderClassName);
      converterProvider = (BinaryConverterProvider<?>)encoderClazz.newInstance();
    } catch (ClassNotFoundException e) {
      throw new RuntimeException(e);
    } catch (InstantiationException e) {
      throw new RuntimeException("failed to instantiate class '" + encoderClassName + "'", e);
    } catch (IllegalAccessException e) {
      throw new RuntimeException(e);
    }

    BinaryConverter<?> converter = converterProvider.getConverter(conf);

    return new LzoBinaryBlockRecordWriter<M, GenericWritable<M>>(new BinaryBlockWriter(
        getOutputStream(job), typeRef, converter));
  }
}

@@ -177,16 +177,63 @@
 Context;
+%0Aimport org.apache.hadoop.util.ReflectionUtils;
 %0A%0Aimport
@@ -1619,42 +1619,8 @@
 ll;%0A
-    Class%3C?%3E encoderClazz = null;%0A
     
@@ -1913,68 +1913,8 @@
 s);%0A
-      encoderClazz = conf.getClassByName(encoderClassName);%0A
     
@@ -1967,20 +1967,24 @@
 %3C?%3E)
-encoderClazz
+ ReflectionUtils
 .new
@@ -1996,86 +1996,60 @@
 nce(
-);
 %0A    
-%7D catch (ClassNotFoundException e) %7B%0A      throw new RuntimeException(e
+    conf.getClassByName(encoderClassName), conf
 );%0A 
@@ -2064,21 +2064,21 @@
 ch (
-Instantiation
+ClassNotFound
 Exce
@@ -2192,86 +2192,8 @@
    %7D
- catch (IllegalAccessException e) %7B%0A      throw new RuntimeException(e);%0A    %7D
 %0A%0A  

use ReflectionUtils