code: public MemoryBlock allocatePage(long size, MemoryConsumer consumer) {
      assert(consumer != null);
      assert(consumer.getMode() == tungstenMemoryMode);
      if (size > MAXIMUM_PAGE_SIZE_BYTES) {
        throw new TooLargePageException(size);
      }
  
      long acquired = acquireExecutionMemory(size, consumer);
      if (acquired <= 0) {
        return null;
      }
  
      final int pageNumber;
      synchronized (this) {
        pageNumber = allocatedPages.nextClearBit(0);
        if (pageNumber >= PAGE_TABLE_SIZE) {
          releaseExecutionMemory(acquired, consumer);
          throw new IllegalStateException(
            "Have already allocated a maximum of " + PAGE_TABLE_SIZE + " pages");
        }
        allocatedPages.set(pageNumber);
      }
      MemoryBlock page = null;
      try {
        page = memoryManager.tungstenMemoryAllocator().allocate(acquired);
      } catch (OutOfMemoryError e) {
        logger.warn("Failed to allocate a page ({} bytes), try again.", acquired);
        // there is no enough memory actually, it means the actual free memory is smaller than
        // MemoryManager thought, we should keep the acquired memory.
        synchronized (this) {
          acquiredButNotUsed += acquired;
          allocatedPages.clear(pageNumber);
        }
        // this could trigger spilling to free some pages.
        return allocatePage(size, consumer);
      }
      page.pageNumber = pageNumber;
      pageTable[pageNumber] = page;
      if (logger.isTraceEnabled()) {
        logger.trace("Allocate page number {} ({} bytes)", pageNumber, acquired);
      }
      return page;
    }
docs: Allocate a block of memory that will be tracked in the MemoryManager's page table; this is
  intended for allocating large blocks of Tungsten memory that will be shared between operators.
  
  Returns `null` if there was not enough memory to allocate the page. May return a page that
  contains fewer bytes than requested, so callers should verify the size of returned pages.
  
  @throws TooLargePageException