'Allocate a buffer of the given size. This method blocks if there is not
enough memory and the buffer pool is configured with blocking mode.
Arguments:
size (int): The buffer size to allocate in bytes [ignored]
max_time_to_block_ms (int): The maximum time in milliseconds to
block for buffer memory to be available
Returns:
io.BytesIO'

def allocate(self, size, max_time_to_block_ms):

    with self._lock:
        if self._free:
            return self._free.popleft()
        elif (self._poolable_size == 0):
            return io.BytesIO()
        else:
            buf = None
            more_memory = threading.Condition(self._lock)
            self._waiters.append(more_memory)
            while (buf is None):
                start_wait = time.time()
                more_memory.wait((max_time_to_block_ms / 1000.0))
                end_wait = time.time()
                if self.wait_time:
                    self.wait_time.record((end_wait - start_wait))
                if self._free:
                    buf = self._free.popleft()
                else:
                    self._waiters.remove(more_memory)
                    raise Errors.KafkaTimeoutError('Failed    to    allocate    memory    within    the    configured    max    blocking    time')
            removed = self._waiters.popleft()
            assert (removed is more_memory), 'Wrong    condition'
            if (self._free and self._waiters):
                self._waiters[0].notify()
            return buf
