%evaluateMapTask
% Entry point for a map task on Hadoop.

%   Copyright 2014-2018 The MathWorks, Inc.

function out = evaluateMapTask(keyValueSinkEndpointString, keyValueSinkEndpointFolder, serializedTaskAttemptID, serializedInputSplit, finalOutputFolder, ~)
out = [];

import matlab.io.datastore.internal.ContextClassLoaderGuard;
import matlab.io.datastore.internal.HadoopConfiguration;
import parallel.internal.mapreduce.MapTask;
import parallel.internal.hadoop.HadoopShuffleSortStrategy;
import parallel.internal.hadoop.HadoopWorker;
import com.mathworks.toolbox.parallel.mapreduce.HadoopTaskAttemptID;

guard = ContextClassLoaderGuard(); %#ok<NASGU>

% Deserialize all information from Hadoop
dctSchedulerMessage(6, 'Beginning Map Task %s', serializedTaskAttemptID);
taskAttemptID = HadoopTaskAttemptID.forName(serializedTaskAttemptID);
configuration = HadoopConfiguration.getGlobalConfiguration();
try
    inputSplit = iDeserializeInputSplit(serializedInputSplit);
    
    taskID = taskAttemptID.getTaskID();
    jobInfo = HadoopWorker.Instance.Job.Info;
    
    ds = jobInfo.InputDatastore;
    
    % Delegate to MapTask to do the actual work.
    shuffleSortStrategy = HadoopShuffleSortStrategy(keyValueSinkEndpointString, keyValueSinkEndpointFolder);
    task = MapTask(taskID, inputSplit, ds, jobInfo.MapFunction, shuffleSortStrategy);
    task.run();
    
catch err
    try
        iSaveErrorToOutputFolder(err, taskAttemptID, finalOutputFolder, configuration);
    catch saveErr
        dctSchedulerMessage(2, 'Unable to save error to output folder. The error was:\n%s\n', saveErr.getReport());
    end
    error(err.getReport());
end

% Helper function that deserializes an input split based on the given type.
function inputSplit = iDeserializeInputSplit(serializedInputSplit)
inputSplit = matlab.bigdata.internal.hadoop.worker.deserializeSplit(...
    serializedInputSplit, @distcompdeserialize);

% Helper function to convert a Hadoop path into an IRI.
function iri = iConvertToIri(hadoopPath)
hdfsLoader = com.mathworks.storage.hdfsloader.GlobalHdfsLoader.get();
hadoopPath = hdfsLoader.newPath(hadoopPath);
iri = char(hadoopPath.toUri());

% Helper function that saves error to the output folder in case of failure.
function iSaveErrorToOutputFolder(err, taskAttemptID, outputFolder, configuration)
import parallel.internal.hadoop.FileSystemTools;
fileSystemTools = FileSystemTools(configuration);
% We have to convert the output folder, which is in Hadoop path format,
% into an IRI. This is because FileSystemTools expects IRIs, where special
% characters are percent encoded.
fileSystemTools.saveErrorToOutputFolder(err, iConvertToIri(outputFolder), taskAttemptID);
