'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
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)'

def append(self, tp, timestamp_ms, key, value, max_time_to_block_ms):

    assert isinstance(tp, TopicPartition), 'not    TopicPartition'
    assert (not self._closed), 'RecordAccumulator    is    closed'
    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]:
            dq = self._batches[tp]
            if dq:
                last = dq[(-1)]
                future = last.try_append(timestamp_ms, key, value)
                if (future is not None):
                    batch_is_full = ((len(dq) > 1) or last.records.is_full())
                    return (future, batch_is_full, False)
        message_size = (MessageSet.HEADER_SIZE + Message.HEADER_SIZE)
        if (key is not None):
            message_size += len(key)
        if (value is not None):
            message_size += len(value)
        assert (message_size <= self.config['buffer_memory']), 'message    too    big'
        size = max(self.config['batch_size'], message_size)
        log.debug('Allocating    a    new    %d    byte    message    buffer    for    %s', size, tp)
        buf = self._free.allocate(size, max_time_to_block_ms)
        with self._tp_locks[tp]:
            assert (not self._closed), 'RecordAccumulator    is    closed'
            if dq:
                last = dq[(-1)]
                future = last.try_append(timestamp_ms, key, value)
                if (future is not None):
                    self._free.deallocate(buf)
                    batch_is_full = ((len(dq) > 1) or last.records.is_full())
                    return (future, batch_is_full, False)
            records = MessageSetBuffer(buf, self.config['batch_size'], self.config['compression_type'], self.config['message_version'])
            batch = RecordBatch(tp, records, self.config['message_version'])
            future = batch.try_append(timestamp_ms, key, value)
            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()
