ListMultiMapWrittable in Hadoop

ListMultiMapWritable.java
package com.srini.writable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.hadoop.io.AbstractMapWritable;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.util.ReflectionUtils;

import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ListMultimap;
import com.google.common.collect.Multimap;
import com.google.common.collect.Multiset;

public class ListMultiMapWritable extends AbstractMapWritable implements
        ListMultimap<Writable, Writable> {
    private ListMultimap<Writable, Writable> map;

    public ListMultiMapWritable() {
        super();
        this.map = ArrayListMultimap.create();
    }

    public ListMultiMapWritable(ListMultimap<Writable, Writable> map) {
        super();
        this.map = map;
    }

    @Override
    public void clear() {
        this.map.clear();

    }

    @Override
    public boolean containsEntry(Object key, Object value) {
        return this.containsEntry(key, value);
    }

    @Override
    public boolean containsKey(Object key) {
        return this.map.containsKey(key);
    }

    @Override
    public boolean containsValue(Object value) {
        return this.map.containsKey(value);
    }

    @Override
    public Collection<Entry<Writable, Writable>> entries() {
        return this.map.entries();
    }

    @Override
    public boolean isEmpty() {
        return this.map.isEmpty();
    }

    @Override
    public Set<Writable> keySet() {
        return this.map.keySet();
    }

    @Override
    public Multiset<Writable> keys() {
        return this.map.keys();
    }

    @Override
    public boolean put(Writable key, Writable value) {
        return this.map.put(key, value);
    }

    @Override
    public boolean putAll(
            Multimap<? extends Writable, ? extends Writable> values) {
        return this.map.putAll(values);
    }

    @Override
    public boolean putAll(Writable key, Iterable<? extends Writable> value) {
        return this.map.putAll(key, value);
    }

    @Override
    public boolean remove(Object key, Object value) {
        return this.map.remove(key, value);
    }

    @Override
    public int size() {
        return this.map.size();
    }

    @Override
    public Collection<Writable> values() {
        return this.map.values();
    }

    @Override
    public Map<Writable, Collection<Writable>> asMap() {
        return this.map.asMap();
    }

    @Override
    public List<Writable> get(Writable key) {
        return this.map.get(key);
    }

    @Override
    public List<Writable> removeAll(Object key) {
        return this.map.removeAll(key);
    }

    @Override
    public List<Writable> replaceValues(Writable key,
            Iterable<? extends Writable> value) {
        return this.map.replaceValues(key, value);
    }

    @SuppressWarnings("unchecked")
    @Override
    public void readFields(DataInput in) throws IOException {
        super.readFields(in);
        this.map.clear();
        int entries = in.readInt();
        for (int i = 0; i < entries; i++) {
            Writable key = (Writable) ReflectionUtils.newInstance(
                    getClass(in.readByte()), getConf());
            key.readFields(in);
            Writable value = (Writable) ReflectionUtils.newInstance(
                    getClass(in.readByte()), getConf());
            value.readFields(in);
            this.map.put(key, value);
        }

    }

    @Override
    public void write(DataOutput out) throws IOException {
        super.write(out);
        out.writeInt(this.map.size());
        Collection<Entry<Writable, Writable>> entries = this.map.entries();
        for (Entry<Writable, Writable> entry : entries) {
            out.writeByte(getId(entry.getKey().getClass()));
            entry.getKey().write(out);
            out.writeByte(getId(entry.getValue().getClass()));
            entry.getValue().write(out);
        }

    }

}