presto-main/src/main/java/io/prestosql/execution/MemoryTrackingRemoteTaskFactory.java/MemoryTrackingRemoteTaskFactory/createRemoteTask
class MemoryTrackingRemoteTaskFactory:    @Override
    public RemoteTask createRemoteTask(
            Session session,
            TaskId taskId,
            InternalNode node,
            PlanFragment fragment,
            Multimap<PlanNodeId, Split> initialSplits,
            OptionalInt totalPartitions,
            OutputBuffers outputBuffers,
            PartitionedSplitCountTracker partitionedSplitCountTracker,
            boolean summarizeTaskInfo)
    {
        RemoteTask task = remoteTaskFactory.createRemoteTask(session,
                taskId,
                node,
                fragment,
                initialSplits,
                totalPartitions,
                outputBuffers,
                partitionedSplitCountTracker,
                summarizeTaskInfo);

        task.addStateChangeListener(new UpdatePeakMemory(stateMachine));
        return task;
    }