Set<KIn> keys = mcc.getKeys();
int maxCSize = mcc.getMaxCollectorSize();
final Mapper<KIn, VIn, KOut, VOut> mapper = mcc.getMapper();
final boolean inputKeysSpecified = keys != null && !keys.isEmpty();
// hook map function into lifecycle and execute it
MapReduceTaskLifecycleService taskLifecycleService = MapReduceTaskLifecycleService.getInstance();
DataContainer dc = cache.getAdvancedCache().getDataContainer();
log.tracef("For m/r task %s invoking %s with input keys %s", mcc.getTaskId(), mcc, mcc.getKeys());
long start = log.isTraceEnabled() ? timeService.time() : 0;
final Set<KOut> intermediateKeys = new HashSet<KOut>();
try {
taskLifecycleService.onPreExecute(mapper, cache);
if (inputKeysSpecified) {
DefaultCollector<KIn, VIn, KOut, VOut> c = new DefaultCollector<KIn, VIn, KOut, VOut>(mcc, maxCSize);
for (KIn key : keys) {
VIn value = cache.get(key);
if (value != null) {
mapper.map(key, value, c);
}
}
combine(mcc, c);
Set<KOut> s = migrateIntermediateKeysAndValues(mcc, c.collectedValues());
intermediateKeys.addAll(s);
} else {
MapCombineTask<KIn, VIn, KOut, VOut> task = new MapCombineTask<KIn, VIn, KOut, VOut>(mcc, maxCSize);
dc.executeTask(new PrimaryOwnerFilter(cdl), task);
intermediateKeys.addAll(task.getMigratedIntermediateKeys());
//the last chunk of remaining keys/values to migrate
Map<KOut, List<VOut>> combinedValues = task.collectedValues();
Set<KOut> lastOne = migrateIntermediateKeysAndValues(mcc, combinedValues);
intermediateKeys.addAll(lastOne);
}
// in case we have stores, we have to process key/values from there as well
if (persistenceManager != null && !inputKeysSpecified) {
KeyFilter<?> keyFilter = new CompositeKeyFilter(new PrimaryOwnerFilter(cdl),
new CollectionKeyFilter(dc.keySet()));
MapCombineTask<KIn, VIn, KOut, VOut> task = new MapCombineTask<KIn, VIn, KOut, VOut>(mcc, maxCSize);
persistenceManager.processOnAllStores(keyFilter, task, true, false);
intermediateKeys.addAll(task.getMigratedIntermediateKeys());
//the last chunk of remaining keys/values to migrate
Map<KOut, List<VOut>> combinedValues = task.collectedValues();
Set<KOut> lastOne = migrateIntermediateKeysAndValues(mcc, combinedValues);
intermediateKeys.addAll(lastOne);
}
} finally {
if (log.isTraceEnabled()) {
log.tracef("Map phase for task %s took %s milliseconds", mcc.getTaskId(),
timeService.timeDuration(start, TimeUnit.MILLISECONDS));
}
taskLifecycleService.onPostExecute(mapper);
}
return intermediateKeys;
}