public RecordAppendResult append(String topic,
                                     int partition,
                                     long timestamp,
                                     byte[] key,
                                     byte[] value,
                                     Header[] headers,
                                     AppendCallbacks callbacks,
                                     long maxTimeToBlock,
                                     boolean abortOnNewBatch,
                                     long nowMs,
                                     Cluster cluster) throws InterruptedException {
        TopicInfo topicInfo = topicInfoMap.computeIfAbsent(topic, k -> new TopicInfo(createBuiltInPartitioner(logContext, k, batchSize)));

        // We keep track of the number of appending thread to make sure we do not miss batches in
        // abortIncompleteBatches().
        appendsInProgress.incrementAndGet();
        ByteBuffer buffer = null;
        if (headers == null) headers = Record.EMPTY_HEADERS;
        try {
            // Loop to retry in case we encounter partitioner's race conditions.
            while (true) {
                // If the message doesn't have any partition affinity, so we pick a partition based on the broker
                // availability and performance.  Note, that here we peek current partition before we hold the
                // deque lock, so we'll need to make sure that it's not changed while we were waiting for the
                // deque lock.
                final BuiltInPartitioner.StickyPartitionInfo partitionInfo;
                final int effectivePartition;
                if (partition == RecordMetadata.UNKNOWN_PARTITION) {
                    partitionInfo = topicInfo.builtInPartitioner.peekCurrentPartitionInfo(cluster);
                    effectivePartition = partitionInfo.partition();
                } else {
                    partitionInfo = null;
                    effectivePartition = partition;
                }

                // Now that we know the effective partition, let the caller know.
                setPartition(callbacks, effectivePartition);

                // check if we have an in-progress batch
                Deque<ProducerBatch> dq = topicInfo.batches.computeIfAbsent(effectivePartition, k -> new ArrayDeque<>());
                synchronized (dq) {
                    // After taking the lock, validate that the partition hasn't changed and retry.
                    if (partitionChanged(topic, topicInfo, partitionInfo, dq, nowMs, cluster))
                        continue;

                    RecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callbacks, dq, nowMs);
                    if (appendResult != null) {
                        // If queue has incomplete batches we disable switch (see comments in updatePartitionInfo).
                        boolean enableSwitch = allBatchesFull(dq);
                        topicInfo.builtInPartitioner.updatePartitionInfo(partitionInfo, appendResult.appendedBytes, cluster, enableSwitch);
                        return appendResult;
                    }
                }

                // we don't have an in-progress record batch try to allocate a new batch
                if (abortOnNewBatch) {
                    // Return a result that will cause another call to append.
                    return new RecordAppendResult(null, false, false, true, 0);
                }

                if (buffer == null) {
                    byte maxUsableMagic = apiVersions.maxUsableProduceMagic();
                    int size = Math.max(this.batchSize, AbstractRecords.estimateSizeInBytesUpperBound(maxUsableMagic, compression.type(), key, value, headers));
                    log.trace("Allocating a new {} byte message buffer for topic {} partition {} with remaining timeout {}ms", size, topic, effectivePartition, maxTimeToBlock);
                    // This call may block if we exhausted buffer space.
                    buffer = free.allocate(size, maxTimeToBlock);
                    // Update the current time in case the buffer allocation blocked above.
                    // NOTE: getting time may be expensive, so calling it under a lock
                    // should be avoided.
                    nowMs = time.milliseconds();
                }

                synchronized (dq) {
                    // After taking the lock, validate that the partition hasn't changed and retry.
                    if (partitionChanged(topic, topicInfo, partitionInfo, dq, nowMs, cluster))
                        continue;

                    RecordAppendResult appendResult = appendNewBatch(topic, effectivePartition, dq, timestamp, key, value, headers, callbacks, buffer, nowMs);
                    // Set buffer to null, so that deallocate doesn't return it back to free pool, since it's used in the batch.
                    if (appendResult.newBatchCreated)
                        buffer = null;
                    // If queue has incomplete batches we disable switch (see comments in updatePartitionInfo).
                    boolean enableSwitch = allBatchesFull(dq);
                    topicInfo.builtInPartitioner.updatePartitionInfo(partitionInfo, appendResult.appendedBytes, cluster, enableSwitch);
                    return appendResult;
                }
            }
        } finally {
            free.deallocate(buffer);
            appendsInProgress.decrementAndGet();
        }
    }
@Test
    public void testUniformBuiltInPartitioner() throws Exception {

        mockRandom = new AtomicInteger();

        long totalSize = 1024 * 1024;
        int batchSize = 1024;  // note that this is also a "sticky" limit for the partitioner
        RecordAccumulator accum = createTestRecordAccumulator(batchSize, totalSize, Compression.NONE, 0);

        // Set up callbacks so that we know what partition is chosen.
        final AtomicInteger partition = new AtomicInteger(RecordMetadata.UNKNOWN_PARTITION);
        RecordAccumulator.AppendCallbacks callbacks = new RecordAccumulator.AppendCallbacks() {
            @Override
            public void setPartition(int p) {
                partition.set(p);
            }

            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {

            }
        };

        PartitionInfo part1 = MetadataResponse.toPartitionInfo(partMetadata1, nodes);
        PartitionInfo part2 = MetadataResponse.toPartitionInfo(partMetadata2, nodes);
        PartitionInfo part3 = MetadataResponse.toPartitionInfo(partMetadata3, nodes);
        Cluster cluster = new Cluster(null, asList(node1, node2), asList(part1, part2, part3),
                Collections.emptySet(), Collections.emptySet());

        // Produce small record, we should switch to first partition.
        accum.append(topic, RecordMetadata.UNKNOWN_PARTITION, 0L, null, value, Record.EMPTY_HEADERS,
                callbacks, maxBlockTimeMs, false, time.milliseconds(), cluster);
        assertEquals(partition1, partition.get());
        assertEquals(1, mockRandom.get());

        // Produce large record, we should exceed "sticky" limit, but produce to this partition
        // as we try to switch after the "sticky" limit is exceeded.  The switch is disabled
        // because of incomplete batch.
        byte[] largeValue = new byte[batchSize];
        accum.append(topic, RecordMetadata.UNKNOWN_PARTITION, 0L, null, largeValue, Record.EMPTY_HEADERS,
                callbacks, maxBlockTimeMs, false, time.milliseconds(), cluster);
        assertEquals(partition1, partition.get());
        assertEquals(1, mockRandom.get());

        // Produce large record, we should switch to next partition as we complete
        // previous batch and exceeded sticky limit.
        accum.append(topic, RecordMetadata.UNKNOWN_PARTITION, 0L, null, largeValue, Record.EMPTY_HEADERS,
                callbacks, maxBlockTimeMs, false, time.milliseconds(), cluster);
        assertEquals(partition2, partition.get());
        assertEquals(2, mockRandom.get());

        // Produce large record, we should switch to next partition as we complete
        // previous batch and exceeded sticky limit.
        accum.append(topic, RecordMetadata.UNKNOWN_PARTITION, 0L, null, largeValue, Record.EMPTY_HEADERS,
                callbacks, maxBlockTimeMs, false, time.milliseconds(), cluster);
        assertEquals(partition3, partition.get());
        assertEquals(3, mockRandom.get());

        // Produce large record, we should switch to next partition as we complete
        // previous batch and exceeded sticky limit.
        accum.append(topic, RecordMetadata.UNKNOWN_PARTITION, 0L, null, largeValue, Record.EMPTY_HEADERS,
                callbacks, maxBlockTimeMs, false, time.milliseconds(), cluster);
        assertEquals(partition1, partition.get());
        assertEquals(4, mockRandom.get());
    }