private <K> K serDeser(K conf) throws Exception {
  SerializationFactory factory = new SerializationFactory(CONF);
  Serializer<K> serializer =
    factory.getSerializer(GenericsUtil.getClass(conf));
  Deserializer<K> deserializer =
    factory.getDeserializer(GenericsUtil.getClass(conf));

  DataOutputBuffer out = new DataOutputBuffer();
  serializer.open(out);
  serializer.serialize(conf);
  serializer.close();

  DataInputBuffer in = new DataInputBuffer();
  in.reset(out.getData(), out.getLength());
  deserializer.open(in);
  K after = deserializer.deserialize(null);
  deserializer.close();
  return after;
}
 

private <E> E makeCopyForPassByValue(Serialization<E> serialization,
                                      E obj) throws IOException {
  Serializer<E> ser =
    serialization.getSerializer(GenericsUtil.getClass(obj));
  Deserializer<E> deser =
    serialization.getDeserializer(GenericsUtil.getClass(obj));

  DataOutputBuffer dof = threadLocalDataOutputBuffer.get();

  dof.reset();
  ser.open(dof);
  ser.serialize(obj);
  ser.close();
  obj = ReflectionUtils.newInstance(GenericsUtil.getClass(obj),
                                    getChainJobConf());
  ByteArrayInputStream bais =
    new ByteArrayInputStream(dof.getData(), 0, dof.getLength());
  deser.open(bais);
  deser.deserialize(obj);
  deser.close();
  return obj;
}
 

/**
 * Stores the array of items in the configuration with the given keyName.
 * 
 * @param <K> the class of the item
 * @param conf the configuration to use 
 * @param items the objects to be stored
 * @param keyName the name of the key to use
 * @throws IndexOutOfBoundsException if the items array is empty
 * @throws IOException : forwards Exceptions from the underlying 
 * {@link Serialization} classes.         
 */
public static <K> void storeArray(Configuration conf, K[] items,
    String keyName) throws IOException {

  DefaultStringifier<K> stringifier = new DefaultStringifier<K>(conf, 
      GenericsUtil.getClass(items[0]));
  try {
    StringBuilder builder = new StringBuilder();
    for (K item : items) {
      builder.append(stringifier.toString(item)).append(SEPARATOR);
    }
    conf.set(keyName, builder.toString());
  }
  finally {
    stringifier.close();
  }
}
 

/**
 * Restores the array of objects from the configuration.
 * 
 * @param <K> the class of the item
 * @param conf the configuration to use
 * @param keyName the name of the key to use
 * @param itemClass the class of the item
 * @return restored object
 * @throws IOException : forwards Exceptions from the underlying 
 * {@link Serialization} classes.
 */
public static <K> K[] loadArray(Configuration conf, String keyName,
    Class<K> itemClass) throws IOException {
  DefaultStringifier<K> stringifier = new DefaultStringifier<K>(conf,
      itemClass);
  try {
    String itemStr = conf.get(keyName);
    ArrayList<K> list = new ArrayList<K>();
    String[] parts = itemStr.split(SEPARATOR);

    for (String part : parts) {
      if (!part.isEmpty())
        list.add(stringifier.fromString(part));
    }

    return GenericsUtil.toArray(itemClass, list);
  }
  finally {
    stringifier.close();
  }
}
 

/**
 * A utility that tests serialization/deserialization. 
 * @param conf configuration to use, "io.serializations" is read to 
 * determine the serialization
 * @param <K> the class of the item
 * @param before item to (de)serialize
 * @return deserialized item
 */
public static <K> K testSerialization(Configuration conf, K before)
	throws Exception {

  SerializationFactory factory = new SerializationFactory(conf);
  Serializer<K> serializer 
    = factory.getSerializer(GenericsUtil.getClass(before));
  Deserializer<K> deserializer 
    = factory.getDeserializer(GenericsUtil.getClass(before));

  DataOutputBuffer out = new DataOutputBuffer();
  serializer.open(out);
  serializer.serialize(before);
  serializer.close();

  DataInputBuffer in = new DataInputBuffer();
  in.reset(out.getData(), out.getLength());
  deserializer.open(in);
  K after = deserializer.deserialize(null);
  deserializer.close();
  return after;
}
 

private <K> K serDeser(K conf) throws Exception {
  SerializationFactory factory = new SerializationFactory(CONF);
  Serializer<K> serializer =
    factory.getSerializer(GenericsUtil.getClass(conf));
  Deserializer<K> deserializer =
    factory.getDeserializer(GenericsUtil.getClass(conf));

  DataOutputBuffer out = new DataOutputBuffer();
  serializer.open(out);
  serializer.serialize(conf);
  serializer.close();

  DataInputBuffer in = new DataInputBuffer();
  in.reset(out.getData(), out.getLength());
  deserializer.open(in);
  K after = deserializer.deserialize(null);
  deserializer.close();
  return after;
}
 

private <E> E makeCopyForPassByValue(Serialization<E> serialization,
                                      E obj) throws IOException {
  Serializer<E> ser =
    serialization.getSerializer(GenericsUtil.getClass(obj));
  Deserializer<E> deser =
    serialization.getDeserializer(GenericsUtil.getClass(obj));

  DataOutputBuffer dof = threadLocalDataOutputBuffer.get();

  dof.reset();
  ser.open(dof);
  ser.serialize(obj);
  ser.close();
  obj = ReflectionUtils.newInstance(GenericsUtil.getClass(obj),
                                    getChainJobConf());
  ByteArrayInputStream bais =
    new ByteArrayInputStream(dof.getData(), 0, dof.getLength());
  deser.open(bais);
  deser.deserialize(obj);
  deser.close();
  return obj;
}
 

/**
 * Stores the array of items in the configuration with the given keyName.
 * 
 * @param <K> the class of the item
 * @param conf the configuration to use 
 * @param items the objects to be stored
 * @param keyName the name of the key to use
 * @throws IndexOutOfBoundsException if the items array is empty
 * @throws IOException : forwards Exceptions from the underlying 
 * {@link Serialization} classes.         
 */
public static <K> void storeArray(Configuration conf, K[] items,
    String keyName) throws IOException {

  DefaultStringifier<K> stringifier = new DefaultStringifier<K>(conf, 
      GenericsUtil.getClass(items[0]));
  try {
    StringBuilder builder = new StringBuilder();
    for (K item : items) {
      builder.append(stringifier.toString(item)).append(SEPARATOR);
    }
    conf.set(keyName, builder.toString());
  }
  finally {
    stringifier.close();
  }
}
 

/**
 * Restores the array of objects from the configuration.
 * 
 * @param <K> the class of the item
 * @param conf the configuration to use
 * @param keyName the name of the key to use
 * @param itemClass the class of the item
 * @return restored object
 * @throws IOException : forwards Exceptions from the underlying 
 * {@link Serialization} classes.
 */
public static <K> K[] loadArray(Configuration conf, String keyName,
    Class<K> itemClass) throws IOException {
  DefaultStringifier<K> stringifier = new DefaultStringifier<K>(conf,
      itemClass);
  try {
    String itemStr = conf.get(keyName);
    ArrayList<K> list = new ArrayList<K>();
    String[] parts = itemStr.split(SEPARATOR);

    for (String part : parts) {
      if (!part.isEmpty())
        list.add(stringifier.fromString(part));
    }

    return GenericsUtil.toArray(itemClass, list);
  }
  finally {
    stringifier.close();
  }
}
 

/**
 * A utility that tests serialization/deserialization. 
 * @param conf configuration to use, "io.serializations" is read to 
 * determine the serialization
 * @param <K> the class of the item
 * @param before item to (de)serialize
 * @return deserialized item
 */
public static <K> K testSerialization(Configuration conf, K before)
	throws Exception {

  SerializationFactory factory = new SerializationFactory(conf);
  Serializer<K> serializer 
    = factory.getSerializer(GenericsUtil.getClass(before));
  Deserializer<K> deserializer 
    = factory.getDeserializer(GenericsUtil.getClass(before));

  DataOutputBuffer out = new DataOutputBuffer();
  serializer.open(out);
  serializer.serialize(before);
  serializer.close();

  DataInputBuffer in = new DataInputBuffer();
  in.reset(out.getData(), out.getLength());
  deserializer.open(in);
  K after = deserializer.deserialize(null);
  deserializer.close();
  return after;
}
 

private <E> E makeCopyForPassByValue(Serialization<E> serialization,
                                      E obj) throws IOException {
  Serializer<E> ser =
    serialization.getSerializer(GenericsUtil.getClass(obj));
  Deserializer<E> deser =
    serialization.getDeserializer(GenericsUtil.getClass(obj));

  DataOutputBuffer dof = threadLocalDataOutputBuffer.get();

  dof.reset();
  ser.open(dof);
  ser.serialize(obj);
  ser.close();
  obj = ReflectionUtils.newInstance(GenericsUtil.getClass(obj),
                                    getChainJobConf());
  ByteArrayInputStream bais =
    new ByteArrayInputStream(dof.getData(), 0, dof.getLength());
  deser.open(bais);
  deser.deserialize(obj);
  deser.close();
  return obj;
}
 

private <K> K serDeser(K conf) throws Exception {
  SerializationFactory factory = new SerializationFactory(CONF);
  Serializer<K> serializer =
    factory.getSerializer(GenericsUtil.getClass(conf));
  Deserializer<K> deserializer =
    factory.getDeserializer(GenericsUtil.getClass(conf));

  DataOutputBuffer out = new DataOutputBuffer();
  serializer.open(out);
  serializer.serialize(conf);
  serializer.close();

  DataInputBuffer in = new DataInputBuffer();
  in.reset(out.getData(), out.getLength());
  deserializer.open(in);
  K after = deserializer.deserialize(null);
  deserializer.close();
  return after;
}
 

/**
 * A utility that tests serialization/deserialization. 
 * @param <K> the class of the item
 * @param conf configuration to use, "io.serializations" is read to 
 * determine the serialization
 * @param before item to (de)serialize
 * @return deserialized item
 */
public static<K> K testSerialization(Configuration conf, K before) 
  throws Exception {
  
  SerializationFactory factory = new SerializationFactory(conf);
  Serializer<K> serializer 
    = factory.getSerializer(GenericsUtil.getClass(before));
  Deserializer<K> deserializer 
    = factory.getDeserializer(GenericsUtil.getClass(before));
 
  DataOutputBuffer out = new DataOutputBuffer();
  serializer.open(out);
  serializer.serialize(before);
  serializer.close();
  
  DataInputBuffer in = new DataInputBuffer();
  in.reset(out.getData(), out.getLength());
  deserializer.open(in);
  K after = deserializer.deserialize(null);
  deserializer.close();
  
  assertEquals(before, after);
  return after;
}
 

/**
 * Stores the array of items in the configuration with the given keyName.
 * 
 * @param <K> the class of the item
 * @param conf the configuration to use 
 * @param items the objects to be stored
 * @param keyName the name of the key to use
 * @throws IndexOutOfBoundsException if the items array is empty
 * @throws IOException : forwards Exceptions from the underlying 
 * {@link Serialization} classes.         
 */
public static <K> void storeArray(Configuration conf, K[] items,
    String keyName) throws IOException {

  DefaultStringifier<K> stringifier = new DefaultStringifier<K>(conf, 
      GenericsUtil.getClass(items[0]));
  try {
    StringBuilder builder = new StringBuilder();
    for (K item : items) {
      builder.append(stringifier.toString(item)).append(SEPARATOR);
    }
    conf.set(keyName, builder.toString());
  }
  finally {
    stringifier.close();
  }
}
 

/**
 * Restores the array of objects from the configuration.
 * 
 * @param <K> the class of the item
 * @param conf the configuration to use
 * @param keyName the name of the key to use
 * @param itemClass the class of the item
 * @return restored object
 * @throws IOException : forwards Exceptions from the underlying 
 * {@link Serialization} classes.
 */
public static <K> K[] loadArray(Configuration conf, String keyName,
    Class<K> itemClass) throws IOException {
  DefaultStringifier<K> stringifier = new DefaultStringifier<K>(conf,
      itemClass);
  try {
    String itemStr = conf.get(keyName);
    ArrayList<K> list = new ArrayList<K>();
    String[] parts = itemStr.split(SEPARATOR);

    for (String part : parts) {
      if (!part.equals(""))
        list.add(stringifier.fromString(part));
    }

    return GenericsUtil.toArray(itemClass, list);
  }
  finally {
    stringifier.close();
  }
}
 

private <E> E makeCopyForPassByValue(Serialization<E> serialization,
                                      E obj) throws IOException {
  Serializer<E> ser =
    serialization.getSerializer(GenericsUtil.getClass(obj));
  Deserializer<E> deser =
    serialization.getDeserializer(GenericsUtil.getClass(obj));

  DataOutputBuffer dof = threadLocalDataOutputBuffer.get();

  dof.reset();
  ser.open(dof);
  ser.serialize(obj);
  ser.close();
  obj = ReflectionUtils.newInstance(GenericsUtil.getClass(obj),
                                    getChainJobConf());
  ByteArrayInputStream bais =
    new ByteArrayInputStream(dof.getData(), 0, dof.getLength());
  deser.open(bais);
  deser.deserialize(obj);
  deser.close();
  return obj;
}
 

private <K> K serDeser(K conf) throws Exception {
  SerializationFactory factory = new SerializationFactory(CONF);
  Serializer<K> serializer =
    factory.getSerializer(GenericsUtil.getClass(conf));
  Deserializer<K> deserializer =
    factory.getDeserializer(GenericsUtil.getClass(conf));

  DataOutputBuffer out = new DataOutputBuffer();
  serializer.open(out);
  serializer.serialize(conf);
  serializer.close();

  DataInputBuffer in = new DataInputBuffer();
  in.reset(out.getData(), out.getLength());
  deserializer.open(in);
  K after = deserializer.deserialize(null);
  deserializer.close();
  return after;
}
 

/**
 * A utility that tests serialization/deserialization. 
 * @param <K> the class of the item
 * @param conf configuration to use, "io.serializations" is read to 
 * determine the serialization
 * @param before item to (de)serialize
 * @return deserialized item
 */
public static<K> K testSerialization(Configuration conf, K before) 
  throws Exception {
  
  SerializationFactory factory = new SerializationFactory(conf);
  Serializer<K> serializer 
    = factory.getSerializer(GenericsUtil.getClass(before));
  Deserializer<K> deserializer 
    = factory.getDeserializer(GenericsUtil.getClass(before));
 
  DataOutputBuffer out = new DataOutputBuffer();
  serializer.open(out);
  serializer.serialize(before);
  serializer.close();
  
  DataInputBuffer in = new DataInputBuffer();
  in.reset(out.getData(), out.getLength());
  deserializer.open(in);
  K after = deserializer.deserialize(null);
  deserializer.close();
  
  assertEquals(before, after);
  return after;
}
 

/**
 * Stores the array of items in the configuration with the given keyName.
 * 
 * @param <K> the class of the item
 * @param conf the configuration to use 
 * @param items the objects to be stored
 * @param keyName the name of the key to use
 * @throws IndexOutOfBoundsException if the items array is empty
 * @throws IOException : forwards Exceptions from the underlying 
 * {@link Serialization} classes.         
 */
public static <K> void storeArray(Configuration conf, K[] items,
    String keyName) throws IOException {

  DefaultStringifier<K> stringifier = new DefaultStringifier<K>(conf, 
      GenericsUtil.getClass(items[0]));
  try {
    StringBuilder builder = new StringBuilder();
    for (K item : items) {
      builder.append(stringifier.toString(item)).append(SEPARATOR);
    }
    conf.set(keyName, builder.toString());
  }
  finally {
    stringifier.close();
  }
}
 

/**
 * Restores the array of objects from the configuration.
 * 
 * @param <K> the class of the item
 * @param conf the configuration to use
 * @param keyName the name of the key to use
 * @param itemClass the class of the item
 * @return restored object
 * @throws IOException : forwards Exceptions from the underlying 
 * {@link Serialization} classes.
 */
public static <K> K[] loadArray(Configuration conf, String keyName,
    Class<K> itemClass) throws IOException {
  DefaultStringifier<K> stringifier = new DefaultStringifier<K>(conf,
      itemClass);
  try {
    String itemStr = conf.get(keyName);
    ArrayList<K> list = new ArrayList<K>();
    String[] parts = itemStr.split(SEPARATOR);

    for (String part : parts) {
      if (!part.equals(""))
        list.add(stringifier.fromString(part));
    }

    return GenericsUtil.toArray(itemClass, list);
  }
  finally {
    stringifier.close();
  }
}
 

/**
 * Stores the item in the configuration with the given keyName.
 * 
 * @param <K>  the class of the item
 * @param conf the configuration to store
 * @param item the object to be stored
 * @param keyName the name of the key to use
 * @throws IOException : forwards Exceptions from the underlying 
 * {@link Serialization} classes. 
 */
public static <K> void store(Configuration conf, K item, String keyName)
throws IOException {

  DefaultStringifier<K> stringifier = new DefaultStringifier<K>(conf,
      GenericsUtil.getClass(item));
  conf.set(keyName, stringifier.toString(item));
  stringifier.close();
}
 

/**
 * Stores the item in the configuration with the given keyName.
 * 
 * @param <K>  the class of the item
 * @param conf the configuration to store
 * @param item the object to be stored
 * @param keyName the name of the key to use
 * @throws IOException : forwards Exceptions from the underlying 
 * {@link Serialization} classes. 
 */
public static <K> void store(Configuration conf, K item, String keyName)
throws IOException {

  DefaultStringifier<K> stringifier = new DefaultStringifier<K>(conf,
      GenericsUtil.getClass(item));
  conf.set(keyName, stringifier.toString(item));
  stringifier.close();
}
 

/**
 * Stores the item in the configuration with the given keyName.
 * 
 * @param <K>  the class of the item
 * @param conf the configuration to store
 * @param item the object to be stored
 * @param keyName the name of the key to use
 * @throws IOException : forwards Exceptions from the underlying 
 * {@link Serialization} classes. 
 */
public static <K> void store(Configuration conf, K item, String keyName)
throws IOException {

  DefaultStringifier<K> stringifier = new DefaultStringifier<K>(conf,
      GenericsUtil.getClass(item));
  conf.set(keyName, stringifier.toString(item));
  stringifier.close();
}
 

/**
 * Stores the item in the configuration with the given keyName.
 * 
 * @param <K>  the class of the item
 * @param conf the configuration to store
 * @param item the object to be stored
 * @param keyName the name of the key to use
 * @throws IOException : forwards Exceptions from the underlying 
 * {@link Serialization} classes. 
 */
public static <K> void store(Configuration conf, K item, String keyName)
throws IOException {

  DefaultStringifier<K> stringifier = new DefaultStringifier<K>(conf,
      GenericsUtil.getClass(item));
  conf.set(keyName, stringifier.toString(item));
  stringifier.close();
}