// Path: src/main/java/com/noctarius/castmapr/core/CollectorImpl.java
// public class CollectorImpl<Key, Value>
//     implements Collector<Key, Value>
// {
// 
//     public final Map<Key, List<Value>> emitted = new HashMap<Key, List<Value>>();
// 
//     @Override
//     public void emit( Key key, Value value )
//     {
//         List<Value> values = emitted.get( key );
//         if ( values == null )
//         {
//             values = new LinkedList<Value>();
//             emitted.put( key, values );
//         }
//         values.add( value );
//     }
// }
// 
// Path: src/main/java/com/noctarius/castmapr/spi/MapAware.java
// public interface MapAware<Key, Value>
// {
// 
//     /**
//      * This method is called after deserializing but before executing {@link Mapper#map(Object, Object, Collector)} or
//      * {@link Reducer#reduce(Object, java.util.Iterator)}.
//      * 
//      * @param map The map the implementing instance is executed against.
//      */
//     void setMap( IMap<Key, Value> map );
// 
// }
// 
// Path: src/main/java/com/noctarius/castmapr/spi/Mapper.java
// public abstract class Mapper<KeyIn, ValueIn, KeyOut, ValueOut>
//     implements Serializable
// {
// 
//     /**
//      * This method is called before the {@link #map(Object, Object, Collector)} method is executed for every value and
//      * can be used to initialize the internal state of the mapper or to emit a special value.
//      * 
//      * @param collector The {@link Collector} to be used for emitting values.
//      */
//     public void initialize( Collector<KeyOut, ValueOut> collector )
//     {
//     }
// 
//     /**
//      * The map method is called for every single key-value pair in the bound {@link IMap} instance on this cluster node
//      * and partition.<br>
//      * Due to it's nature of a DataGrid Hazelcast distributes values all over the cluster and so this method is executed
//      * on multiple servers at the same time.<br>
//      * If you want to know more about the implementation of MapReduce algorithms read the {@see <a
//      * href="http://research.google.com/archive/mapreduce-osdi04.pdf">Google Whitepaper on MapReduce</a>}.
//      * 
//      * @param key
//      * @param value
//      * @param collector
//      */
//     public abstract void map( KeyIn key, ValueIn value, Collector<KeyOut, ValueOut> collector );
// 
//     /**
//      * This method is called after the {@link #map(Object, Object, Collector)} method is executed for every value and
//      * can be used to finalize the internal state of the mapper or to emit a special value.
//      * 
//      * @param collector The {@link Collector} to be used for emitting values.
//      */
//     public void finalized( Collector<KeyOut, ValueOut> collector )
//     {
//     }
// 
// }
// 
// Path: src/main/java/com/noctarius/castmapr/spi/PartitionIdAware.java
// public interface PartitionIdAware
// {
// 
//     /**
//      * This method is called after deserializing but before executing {@link Mapper#map(Object, Object, Collector)} or
//      * {@link Reducer#reduce(Object, java.util.Iterator)}.
//      * 
//      * @param partitionId The current partitionId
//      */
//     void setPartitionId( int partitionId );
// 
// }
// 
// Path: src/main/java/com/noctarius/castmapr/spi/Reducer.java
// public interface Reducer<Key, Value>
//     extends Serializable
// {
// 
//     /**
//      * The reduce method is called either locally after of intermediate results are retrieved from mapping algorithms or
//      * (if the Reducer implementation is distributable - see {@link Distributable} or {@link DistributableReducer}) on
//      * the different cluster nodes.<br>
//      * <b>Caution: For distributable {@link Reducer}s you need to pay attention that the reducer is executed multiple
//      * times and only with intermediate results of one cluster node! This is totally ok for example for sum-algorithms
//      * but can be a problem for other kinds!</b><br>
//      * {@link Reducer} implementations are never distributed by default - they need to explicitly be marked using the
//      * {@link Distributable} annotation or by implementing the {@link DistributableReducer} interface.
//      * 
//      * @param key The reduced key
//      * @param values The values corresponding to the reduced key
//      * @return The reduced value
//      */
//     Value reduce( Key key, Iterator<Value> values );
// 
// }


import com.noctarius.castmapr.spi.PartitionIdAware;
import com.noctarius.castmapr.spi.Reducer;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import com.hazelcast.core.IMap;
import com.hazelcast.map.MapContainer;
import com.hazelcast.map.MapService;
import com.hazelcast.map.RecordStore;
import com.hazelcast.nio.serialization.Data;
import com.hazelcast.spi.BackupOperation;
import com.hazelcast.spi.ProxyService;
import com.hazelcast.spi.exception.RetryableHazelcastException;
import com.noctarius.castmapr.core.CollectorImpl;
import com.noctarius.castmapr.spi.MapAware;
import com.noctarius.castmapr.spi.Mapper;
        if ( keys == null || keys.size() == 0 )
        {
            for ( Entry<Data, Data> entry : recordStore.entrySetData() )
            {
                KeyIn key = (KeyIn) mapService.toObject( entry.getKey() );
                ValueIn value = (ValueIn) mapService.toObject( entry.getValue() );
                mapper.map( key, value, collector );
            }
        }
        else
        {
            // Iterate only defined keys
            for ( KeyIn key : keys )
            {
                Data dataKey = mapService.toData( key );
                ValueIn value = (ValueIn) mapService.toObject( recordStore.get( dataKey ) );
                mapper.map( key, value, collector );
            }
        }
        mapper.finalized( collector );

        if ( reducer != null )
        {
            if ( reducer instanceof PartitionIdAware )
            {
                ( (PartitionIdAware) reducer ).setPartitionId( partitionId );
            }
            if ( reducer instanceof MapAware )
            {
                IMap map = (IMap) proxyService.getDistributedObject( MapService.SERVICE_NAME, name );
// Path: src/main/java/com/noctarius/castmapr/core/CollectorImpl.java
// public class CollectorImpl<Key, Value>
//     implements Collector<Key, Value>
// {
// 
//     public final Map<Key, List<Value>> emitted = new HashMap<Key, List<Value>>();
// 
//     @Override
//     public void emit( Key key, Value value )
//     {
//         List<Value> values = emitted.get( key );
//         if ( values == null )
//         {
//             values = new LinkedList<Value>();
//             emitted.put( key, values );
//         }
//         values.add( value );
//     }
// }
// 
// Path: src/main/java/com/noctarius/castmapr/spi/MapAware.java
// public interface MapAware<Key, Value>
// {
// 
//     /**
//      * This method is called after deserializing but before executing {@link Mapper#map(Object, Object, Collector)} or
//      * {@link Reducer#reduce(Object, java.util.Iterator)}.
//      * 
//      * @param map The map the implementing instance is executed against.
//      */
//     void setMap( IMap<Key, Value> map );
// 
// }
// 
// Path: src/main/java/com/noctarius/castmapr/spi/Mapper.java
// public abstract class Mapper<KeyIn, ValueIn, KeyOut, ValueOut>
//     implements Serializable
// {
// 
//     /**
//      * This method is called before the {@link #map(Object, Object, Collector)} method is executed for every value and
//      * can be used to initialize the internal state of the mapper or to emit a special value.
//      * 
//      * @param collector The {@link Collector} to be used for emitting values.
//      */
//     public void initialize( Collector<KeyOut, ValueOut> collector )
//     {
//     }
// 
//     /**
//      * The map method is called for every single key-value pair in the bound {@link IMap} instance on this cluster node
//      * and partition.<br>
//      * Due to it's nature of a DataGrid Hazelcast distributes values all over the cluster and so this method is executed
//      * on multiple servers at the same time.<br>
//      * If you want to know more about the implementation of MapReduce algorithms read the {@see <a
//      * href="http://research.google.com/archive/mapreduce-osdi04.pdf">Google Whitepaper on MapReduce</a>}.
//      * 
//      * @param key
//      * @param value
//      * @param collector
//      */
//     public abstract void map( KeyIn key, ValueIn value, Collector<KeyOut, ValueOut> collector );
// 
//     /**
//      * This method is called after the {@link #map(Object, Object, Collector)} method is executed for every value and
//      * can be used to finalize the internal state of the mapper or to emit a special value.
//      * 
//      * @param collector The {@link Collector} to be used for emitting values.
//      */
//     public void finalized( Collector<KeyOut, ValueOut> collector )
//     {
//     }
// 
// }
// 
// Path: src/main/java/com/noctarius/castmapr/spi/PartitionIdAware.java
// public interface PartitionIdAware
// {
// 
//     /**
//      * This method is called after deserializing but before executing {@link Mapper#map(Object, Object, Collector)} or
//      * {@link Reducer#reduce(Object, java.util.Iterator)}.
//      * 
//      * @param partitionId The current partitionId
//      */
//     void setPartitionId( int partitionId );
// 
// }
// 
// Path: src/main/java/com/noctarius/castmapr/spi/Reducer.java
// public interface Reducer<Key, Value>
//     extends Serializable
// {
// 
//     /**
//      * The reduce method is called either locally after of intermediate results are retrieved from mapping algorithms or
//      * (if the Reducer implementation is distributable - see {@link Distributable} or {@link DistributableReducer}) on
//      * the different cluster nodes.<br>
//      * <b>Caution: For distributable {@link Reducer}s you need to pay attention that the reducer is executed multiple
//      * times and only with intermediate results of one cluster node! This is totally ok for example for sum-algorithms
//      * but can be a problem for other kinds!</b><br>
//      * {@link Reducer} implementations are never distributed by default - they need to explicitly be marked using the
//      * {@link Distributable} annotation or by implementing the {@link DistributableReducer} interface.
//      * 
//      * @param key The reduced key
//      * @param values The values corresponding to the reduced key
//      * @return The reduced value
//      */
//     Value reduce( Key key, Iterator<Value> values );
// 
// }


// Path: src/main/java/com/noctarius/castmapr/core/operation/IMapMapReduceOperation.java
import com.noctarius.castmapr.spi.PartitionIdAware;
import com.noctarius.castmapr.spi.Reducer;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import com.hazelcast.core.IMap;
import com.hazelcast.map.MapContainer;
import com.hazelcast.map.MapService;
import com.hazelcast.map.RecordStore;
import com.hazelcast.nio.serialization.Data;
import com.hazelcast.spi.BackupOperation;
import com.hazelcast.spi.ProxyService;
import com.hazelcast.spi.exception.RetryableHazelcastException;
import com.noctarius.castmapr.core.CollectorImpl;
import com.noctarius.castmapr.spi.MapAware;
import com.noctarius.castmapr.spi.Mapper;

        if ( keys == null || keys.size() == 0 )
        {
            for ( Entry<Data, Data> entry : recordStore.entrySetData() )
            {
                KeyIn key = (KeyIn) mapService.toObject( entry.getKey() );
                ValueIn value = (ValueIn) mapService.toObject( entry.getValue() );
                mapper.map( key, value, collector );
            }
        }
        else
        {
            // Iterate only defined keys
            for ( KeyIn key : keys )
            {
                Data dataKey = mapService.toData( key );
                ValueIn value = (ValueIn) mapService.toObject( recordStore.get( dataKey ) );
                mapper.map( key, value, collector );
            }
        }
        mapper.finalized( collector );

        if ( reducer != null )
        {
            if ( reducer instanceof PartitionIdAware )
            {
                ( (PartitionIdAware) reducer ).setPartitionId( partitionId );
            }
            if ( reducer instanceof MapAware )
            {
                IMap map = (IMap) proxyService.getDistributedObject( MapService.SERVICE_NAME, name );
                ( (MapAware) reducer ).setMap( map );