@@ -0,0 +1,2846 @@
+package com.google.cloud.dataflow.contrib.hadoop;%0A%0Aimport com.google.cloud.dataflow.sdk.coders.Coder;%0Aimport com.google.cloud.dataflow.sdk.io.BoundedSource;%0Aimport com.google.cloud.dataflow.sdk.options.PipelineOptions;%0Aimport com.google.cloud.dataflow.sdk.util.CoderUtils;%0Aimport com.google.cloud.dataflow.sdk.values.KV;%0Aimport org.apache.avro.mapred.AvroKey;%0Aimport org.apache.avro.mapreduce.AvroKeyInputFormat;%0Aimport org.apache.hadoop.io.NullWritable;%0Aimport org.apache.hadoop.mapreduce.InputSplit;%0Aimport org.apache.hadoop.mapreduce.lib.input.FileInputFormat;%0A%0Aimport java.io.IOException;%0A%0Apublic class AvroHadoopFileSource%3CT%3E extends HadoopFileSource%3CAvroKey%3CT%3E, NullWritable%3E %7B%0A%0A  private final Coder%3CT%3E avroCoder;%0A%0A  public AvroHadoopFileSource(String filepattern, Coder%3CT%3E avroCoder) %7B%0A    this(filepattern, avroCoder, null);%0A  %7D%0A%0A  public AvroHadoopFileSource(String filepattern, Coder%3CT%3E avroCoder, SerializableSplit serializableSplit) %7B%0A    super(filepattern,%0A        ClassUtil.%3CAvroKeyInputFormat%3CT%3E%3EcastClass(AvroKeyInputFormat.class),%0A        ClassUtil.%3CAvroKey%3CT%3E%3EcastClass(AvroKey.class),%0A        NullWritable.class, serializableSplit);%0A    this.avroCoder = avroCoder;%0A  %7D%0A%0A  @Override%0A  public BoundedReader%3CKV%3CAvroKey%3CT%3E, NullWritable%3E%3E createReader(PipelineOptions options) throws IOException %7B%0A    this.validate();%0A%0A    if (serializableSplit == null) %7B%0A      return new AvroHadoopFileReader%3C%3E(this, filepattern, formatClass);%0A    %7D else %7B%0A      return new AvroHadoopFileReader%3C%3E(this, filepattern, formatClass,%0A          serializableSplit.getSplit());%0A    %7D%0A  %7D%0A%0A  static class AvroHadoopFileReader%3CT%3E extends HadoopFileReader%3CAvroKey%3CT%3E, NullWritable%3E %7B%0A%0A    public AvroHadoopFileReader(BoundedSource%3CKV%3CAvroKey%3CT%3E, NullWritable%3E%3E source,%0A                                String filepattern, Class%3C? extends FileInputFormat%3C?, ?%3E%3E formatClass) %7B%0A      super(source, filepattern, formatClass);%0A    %7D%0A%0A    public AvroHadoopFileReader(BoundedSource%3CKV%3CAvroKey%3CT%3E, NullWritable%3E%3E source,%0A                                String filepattern, Class%3C? extends FileInputFormat%3C?, ?%3E%3E formatClass,%0A                                InputSplit split) %7B%0A      super(source, filepattern, formatClass, split);%0A    %7D%0A%0A%0A    @SuppressWarnings(%22unchecked%22)%0A    protected KV%3CAvroKey%3CT%3E, NullWritable%3E nextPair() throws IOException, InterruptedException %7B%0A      AvroKey%3CT%3E key = currentReader.getCurrentKey();%0A      NullWritable value = currentReader.getCurrentValue();%0A%0A      Coder%3CT%3E avroCoder = ((AvroHadoopFileSource%3CT%3E) this.getCurrentSource()).avroCoder;%0A      key = new AvroKey(CoderUtils.clone(avroCoder, key.datum()));%0A%0A      return KV.of(key, value);%0A    %7D%0A%0A  %7D%0A%0A  private static class ClassUtil %7B%0A    @SuppressWarnings(%22unchecked%22)%0A    private static %3CT%3E Class%3CT%3E castClass(Class%3C?%3E aClass) %7B%0A      return (Class%3CT%3E)aClass;%0A    %7D%0A  %7D%0A%0A%7D%0A

add AvroHadoopFileSource