package gobblin.compaction.mapreduce.avro;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.avro.Schema;
import org.apache.commons.codec.binary.Base64;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.lib.input.CombineFileSplit;


/**
 * A subclass of {@link org.apache.hadoop.mapreduce.lib.input.CombineFileSplit}. The purpose is to add the input file's
 * avro schema to a split.
 *
 *
 * @author mwol
 */
public class AvroCombineFileSplit extends CombineFileSplit {

  private Schema schema;

  public AvroCombineFileSplit() {
  }

  public AvroCombineFileSplit(Path[] paths, long[] startOffsets, long[] lengths, String[] locations) {
    super(paths, startOffsets, lengths, locations);
  }

  public AvroCombineFileSplit(CombineFileSplit old) throws IOException {
    super(old);
  }

  public AvroCombineFileSplit(Path[] paths, long[] startOffsets, long[] lengths, String[] locations, Schema schema) {
    super(paths, startOffsets, lengths, locations);
    this.schema = schema;
  }

  public AvroCombineFileSplit(CombineFileSplit old, Schema schema) throws IOException {
    this(old);
    this.schema = schema;
  }

  public Schema getSchema() {
    return schema;
  }

  @Override
  public void readFields(DataInput in) throws IOException {
    super.readFields(in);
    schema = new Schema.Parser().parse(fromBase64(Text.readString(in)));
  }

  @Override
  public void write(DataOutput out) throws IOException {
    super.write(out);
    Text.writeString(out, toBase64(schema.toString()));
  }

  @Override
  public String toString() {
    return super.toString() + " Schema: " + schema.toString();
  }

  private static String toBase64(String rawString) {
    Base64 base64decoder = new Base64();
    return new String(base64decoder.encode(rawString.getBytes()));
  }

  private static String fromBase64(String base64String) {
    Base64 base64decoder = new Base64();
    return new String(base64decoder.decode(base64String.getBytes()));
  }
}

@@ -329,16 +329,65 @@
 Split;%0A%0A
+import gobblin.configuration.ConfigurationKeys;%0A%0A
 %0A/**%0A * 
@@ -1915,26 +1915,120 @@
 ng.getBytes(
-))
+ConfigurationKeys.DEFAULT_CHARSET_ENCODING)),%0A        ConfigurationKeys.DEFAULT_CHARSET_ENCODING
 );%0A  %7D%0A%0A  pr
@@ -2187,18 +2187,112 @@
 etBytes(
-))
+ConfigurationKeys.DEFAULT_CHARSET_ENCODING)),%0A        ConfigurationKeys.DEFAULT_CHARSET_ENCODING
 );%0A  %7D%0A%7D

Fix findbug warnings on AvroCombineFileSplit