// Path: src/main/java/com/urbanairship/datacube/backfill/CollectionWritable.java
// public class CollectionWritable implements Writable {
//     private Collection<? extends Writable> collection;
//     private Class<? extends Writable> elementClass;
//     
//     public CollectionWritable() { } // No-op constructor is required for deserializing
//     
//     public CollectionWritable(Class<? extends Writable> elementClass, 
//             Collection<? extends Writable> collection) {
//         this.elementClass = elementClass;
//         this.collection = collection;
//     }
// 
//     @SuppressWarnings("unchecked")
//     @Override
//     public void readFields(DataInput in) throws IOException {
//         try {
//             String valueClassName = in.readUTF();
//             elementClass = (Class<? extends Writable>)Class.forName(valueClassName);
//             if(!Writable.class.isAssignableFrom(elementClass)) {
//                 throw new RuntimeException("valueClass is not writable: " + valueClassName);
//             }
//             int numElements = in.readInt();
//             List<Writable> newList = new ArrayList<Writable>(numElements);
//             for(int i=0; i<numElements; i++) {
//                 Writable instance = elementClass.newInstance();
//                 instance.readFields(in);
//                 newList.add(instance);
//             }
//             collection = newList;
//         } catch (Exception e) {
//             throw new RuntimeException(e);
//         }
//     }
// 
//     @Override
//     public void write(DataOutput out) throws IOException {
//         out.writeUTF(elementClass.getName());
//         out.writeInt(collection.size());
//         for(Writable writable: collection) {
//             writable.write(out);
//         }
//     }
//     
//     public Collection<? extends Writable> getCollection() {
//         return collection;
//     }
// }


import com.urbanairship.datacube.backfill.CollectionWritable;
import org.apache.commons.codec.binary.Base64;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.DataInputBuffer;
import org.apache.hadoop.io.DataOutputBuffer;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
/*
Copyright 2012 Urban Airship and Contributors
*/

package com.urbanairship.datacube.collectioninputformat;



/**
 * A Hadoop InputFormat that lets you use an arbitrary Collection as input to a mapreduce job.
 * Each item in the input Collection will be given to a single map() call as its key.
 */
public class CollectionInputFormat extends InputFormat<Writable, NullWritable> {
    public static final String CONFKEY_COLLECTION = "collectioninputformat.collectionitems";
    public static final String CONFKEY_VALCLASS = "collectioninputformat.valueclass";

    /**
     * Stores the given collection as a configuration value. When getSplits() runs later,
     * it will be able to deserialize the collection and use it.
     */
    public static <T extends Writable> void setCollection(Job job, Class<T> valueClass,
                                                          Collection<T> collection) throws IOException {
        setCollection(job.getConfiguration(), valueClass, collection);
    }

    public static <T extends Writable> void setCollection(Configuration conf, Class<T> valueClass,
                                                          Collection<T> collection) throws IOException {
        conf.set(CONFKEY_COLLECTION, toBase64(valueClass, collection));
        conf.set(CONFKEY_VALCLASS, valueClass.getName());
    }

    public static <T extends Writable> String toBase64(Class<T> valueClass,
                                                       Collection<? extends Writable> collection) throws IOException {
        DataOutputBuffer out = new DataOutputBuffer();
// Path: src/main/java/com/urbanairship/datacube/backfill/CollectionWritable.java
// public class CollectionWritable implements Writable {
//     private Collection<? extends Writable> collection;
//     private Class<? extends Writable> elementClass;
//     
//     public CollectionWritable() { } // No-op constructor is required for deserializing
//     
//     public CollectionWritable(Class<? extends Writable> elementClass, 
//             Collection<? extends Writable> collection) {
//         this.elementClass = elementClass;
//         this.collection = collection;
//     }
// 
//     @SuppressWarnings("unchecked")
//     @Override
//     public void readFields(DataInput in) throws IOException {
//         try {
//             String valueClassName = in.readUTF();
//             elementClass = (Class<? extends Writable>)Class.forName(valueClassName);
//             if(!Writable.class.isAssignableFrom(elementClass)) {
//                 throw new RuntimeException("valueClass is not writable: " + valueClassName);
//             }
//             int numElements = in.readInt();
//             List<Writable> newList = new ArrayList<Writable>(numElements);
//             for(int i=0; i<numElements; i++) {
//                 Writable instance = elementClass.newInstance();
//                 instance.readFields(in);
//                 newList.add(instance);
//             }
//             collection = newList;
//         } catch (Exception e) {
//             throw new RuntimeException(e);
//         }
//     }
// 
//     @Override
//     public void write(DataOutput out) throws IOException {
//         out.writeUTF(elementClass.getName());
//         out.writeInt(collection.size());
//         for(Writable writable: collection) {
//             writable.write(out);
//         }
//     }
//     
//     public Collection<? extends Writable> getCollection() {
//         return collection;
//     }
// }


// Path: src/main/java/com/urbanairship/datacube/collectioninputformat/CollectionInputFormat.java
import com.urbanairship.datacube.backfill.CollectionWritable;
import org.apache.commons.codec.binary.Base64;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.DataInputBuffer;
import org.apache.hadoop.io.DataOutputBuffer;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;

/*
Copyright 2012 Urban Airship and Contributors
*/

package com.urbanairship.datacube.collectioninputformat;



/**
 * A Hadoop InputFormat that lets you use an arbitrary Collection as input to a mapreduce job.
 * Each item in the input Collection will be given to a single map() call as its key.
 */
public class CollectionInputFormat extends InputFormat<Writable, NullWritable> {
    public static final String CONFKEY_COLLECTION = "collectioninputformat.collectionitems";
    public static final String CONFKEY_VALCLASS = "collectioninputformat.valueclass";

    /**
     * Stores the given collection as a configuration value. When getSplits() runs later,
     * it will be able to deserialize the collection and use it.
     */
    public static <T extends Writable> void setCollection(Job job, Class<T> valueClass,
                                                          Collection<T> collection) throws IOException {
        setCollection(job.getConfiguration(), valueClass, collection);
    }

    public static <T extends Writable> void setCollection(Configuration conf, Class<T> valueClass,
                                                          Collection<T> collection) throws IOException {
        conf.set(CONFKEY_COLLECTION, toBase64(valueClass, collection));
        conf.set(CONFKEY_VALCLASS, valueClass.getName());
    }

    public static <T extends Writable> String toBase64(Class<T> valueClass,
                                                       Collection<? extends Writable> collection) throws IOException {
        DataOutputBuffer out = new DataOutputBuffer();
        new CollectionWritable(valueClass, collection).write(out);