public long acquireExecutionMemory(
            long required,
            MemoryMode mode,
            MemoryConsumer consumer) {
        assert (required >= 0);

        // If we are allocating Tungsten pages off-heap and receive a request to allocate on-heap
        // memory here, then it may not make sense to spill since that would only end up freeing
        // off-heap memory. This is subject to change, though, so it may be risky to make this
        // optimization now in case we forget to undo it late when making changes.
        synchronized (this) {
            long got = memoryManager.acquireMemory(required, taskAttemptId);

            // Try to release memory from other consumers first, then we can reduce the frequency of
            // spilling, avoid to have too many spilled files.
            if (got < required) {
                // Call spill() on other consumers to release memory
                for (MemoryConsumer c : consumers) {
                    if (c != consumer && c.getUsed() > 0) {
                        try {
                            long released = c.spill(required - got, consumer);
                            if (released > 0 && mode == tungstenMemoryMode) {
                                logger.debug("Task {} released {} from {} for {}", taskAttemptId,
                                        Utils.bytesToString(released), c, consumer);
                                got += memoryManager.acquireMemory(required - got, taskAttemptId);
                                if (got >= required) {
                                    break;
                                }
                            }
                        } catch (IOException e) {
                            logger.error("error while calling spill() on " + c, e);
                            throw new OutOfMemoryError("error while calling spill() on " + c + " : "
                                    + e.getMessage());
                        }
                    }
                }
            }

            // call spill() on itself
            if (got < required && consumer != null) {
                try {
                    long released = consumer.spill(required - got, consumer);
                    if (released > 0 && mode == tungstenMemoryMode) {
                        logger.debug("Task {} released {} from itself ({})", taskAttemptId,
                                Utils.bytesToString(released), consumer);
                        got += memoryManager.acquireMemory(required - got, taskAttemptId);
                    }
                } catch (IOException e) {
                    logger.error("error while calling spill() on " + consumer, e);
                    throw new OutOfMemoryError("error while calling spill() on " + consumer + " : "
                            + e.getMessage());
                }
            }

            if (consumer != null) {
                consumers.add(consumer);
            }
            logger.debug("Task {} acquire {} for {}", taskAttemptId, Utils.bytesToString(got), consumer);
            return got;
        }
    }
Acquire N bytes of memory for a consumer. If there is no enough memory, it will call
spill() of consumers to release more memory.

@return number of bytes successfully granted.
indexr-query-opt/src/main/java/org/apache/spark/memory/TaskMemoryManager.java