docs: def append(self, tp, timestamp_ms, key, value, headers, max_time_to_block_ms,
                 estimated_size=0):
          """Add a record to the accumulator, return the append result.
  
          The append result will contain the future metadata, and flag for
          whether the appended batch is full or a new batch is created
  
          Arguments:
              tp (TopicPartition): The topic/partition to which this record is
                  being sent
              timestamp_ms (int): The timestamp of the record (epoch ms)
              key (bytes): The key for the record
              value (bytes): The value for the record
              headers (List[Tuple[str, bytes]]): The header fields for the record
              max_time_to_block_ms (int): The maximum time in milliseconds to
                  block for buffer memory to be available
  
          Returns:
              tuple: (future, batch_is_full, new_batch_created)
          """
          assert isinstance(tp, TopicPartition), 'not TopicPartition'
          assert not self._closed, 'RecordAccumulator is closed'
          # We keep track of the number of appending thread to make sure we do
          # not miss batches in abortIncompleteBatches().
          self._appends_in_progress.increment()
          try:
              if tp not in self._tp_locks:
                  with self._tp_locks[None]:
                      if tp not in self._tp_locks:
                          self._tp_locks[tp] = threading.Lock()
  
              with self._tp_locks[tp]:
                  # check if we have an in-progress batch
                  dq = self._batches[tp]
                  if dq:
                      last = dq[-1]
                      future = last.try_append(timestamp_ms, key, value, headers)
                      if future is not None:
                          batch_is_full = len(dq) > 1 or last.records.is_full()
                          return future, batch_is_full, False
  
              size = max(self.config['batch_size'], estimated_size)
              log.debug("Allocating a new %d byte message buffer for %s", size, tp) # trace
              buf = self._free.allocate(size, max_time_to_block_ms)
              with self._tp_locks[tp]:
                  # Need to check if producer is closed again after grabbing the
                  # dequeue lock.
                  assert not self._closed, 'RecordAccumulator is closed'
  
                  if dq:
                      last = dq[-1]
                      future = last.try_append(timestamp_ms, key, value, headers)
                      if future is not None:
                          # Somebody else found us a batch, return the one we
                          # waited for! Hopefully this doesn't happen often...
                          self._free.deallocate(buf)
                          batch_is_full = len(dq) > 1 or last.records.is_full()
                          return future, batch_is_full, False
  
                  records = MemoryRecordsBuilder(
                      self.config['message_version'],
                      self.config['compression_attrs'],
                      self.config['batch_size']
                  )
  
                  batch = ProducerBatch(tp, records, buf)
                  future = batch.try_append(timestamp_ms, key, value, headers)
                  if not future:
                      raise Exception()
  
                  dq.append(batch)
                  self._incomplete.add(batch)
                  batch_is_full = len(dq) > 1 or batch.records.is_full()
                  return future, batch_is_full, True
          finally:
              self._appends_in_progress.decrement()