code: def allocate(self, size, max_time_to_block_ms):
          
          with self._lock:
              # check if we have a free buffer of the right size pooled
              if self._free:
                  return self._free.popleft()
  
              elif self._poolable_size == 0:
                  return io.BytesIO()
  
              else:
                  # we are out of buffers and will have to block
                  buf = None
                  more_memory = threading.Condition(self._lock)
                  self._waiters.append(more_memory)
                  # loop over and over until we have a buffer or have reserved
                  # enough memory to allocate one
                  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")
  
                  # remove the condition for this thread to let the next thread
                  # in line start getting memory
                  removed = self._waiters.popleft()
                  assert removed is more_memory, 'Wrong condition'
  
                  # signal any additional waiters if there is more memory left
                  # over for them
                  if self._free and self._waiters:
                      self._waiters[0].notify()
  
                  # unlock and return the buffer
                  return buf
docs: 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