@Around("MapReduceInvokeAspect.partitionInvocationPointcut(pmr)")
	public Object mapreduceInvocation(ProceedingJoinPoint pjp, PartitionMapReduce pmr) throws Throwable {
		
		if (hazelcast != null) {
			
			Member local = hazelcast.getCluster().getLocalMember();
			
			// look for PartitionKey annotation in the method signature
			MethodSignature signature = (MethodSignature) pjp.getStaticPart().getSignature();
			PartitionKeyTrace trace   = getPartitionKeyTrace(pjp.getTarget(), signature);
			
			// split input collection into subsets, each subset resolving to
			// a grid partition it has affinity to
			PartitionReducer reducer = getPartitionReducer(pmr);
			PartitionMapper  mapper  = getPartitionMapper(pmr);
			
			if (mapper != null && reducer != null) {
				// map input to partitions
				Collection input = (Collection) pjp.getArgs()[trace.getPidx()];
				Map<Integer, Collection> splits = mapper.map(input);
				
				// process mapped input
				List<PartitionFuture> futures = new ArrayList<PartitionFuture>();
				for (Map.Entry<Integer, Collection> entry : splits.entrySet()) {
					if (!entry.getValue().isEmpty()) {
						
						// this is a dirty hack; seems I can't just use partition id :(
						Integer partitionKey = mapper.getPartitionKey(head(entry.getValue()));
						
						// replace original collection with current split
						Object[] args = pjp.getArgs();
						args[trace.getPidx()] = entry.getValue();
						
						// does this split belong to this member's partition ?
						Partition partition = hazelcast.getPartitionService().getPartition(partitionKey);
						if (partition.getOwner().equals(local)) {
							reducer.reduce(entry.getKey(), pjp.proceed(args));
						}
						else {
							// fire fringe event
							FringeEvent fevent = new FringeEvent(	partitionKey, 
																	signature.getDeclaringTypeName(), 
																	signature.getName(), 
																	signature.getParameterTypes(), 
																	args
							);
							try {
								Future<Object> fut = hazelcast.getExecutorService().submit(fevent);
								futures.add(
											new PartitionFuture(entry.getKey(), fut)
								);
							}
							catch (Exception ex) {
								throw new PartitionExecutionException("Unable to complete map-reduce execution :(", ex);
							}
						}
					}
				}
				// wait for collected futures to complete (best effort)
				//
				for (PartitionFuture fut : futures) {
					Object result = getFutureResult(fut);
					if (result != null) {
						reducer.reduce(fut.getPartitionId(), result);
					}
				}
				return reducer.collate();
			}
			else {
				log.log(Level.WARNING, "Either ParitionMapper or PartitionReducer is null, invocation will not be distributed !");
				return pjp.proceed();
			}
		}
		else {
			log.log(Level.WARNING, "Hazelcast instance has not been autowired, defaulting to non-distributed execution :?");
			return pjp.proceed();
		}
		
	}