《Java 消息中间件·Kafka 内核与客户端》7.2 RecordAccumulator 与批次缓存

// RecordAccumulator.java:84
    private final ConcurrentMap<String /*topic*/, TopicInfo> topicInfoMap = new CopyOnWriteMap<>();

// RecordAccumulator.java:1283-1290
    private static class TopicInfo {
        public final ConcurrentMap<Integer /*partition*/, Deque<ProducerBatch>> batches = new CopyOnWriteMap<>();
        public final BuiltInPartitioner builtInPartitioner;

        public TopicInfo(BuiltInPartitioner builtInPartitioner) {
            this.builtInPartitioner = builtInPartitioner;
        }
    }

// RecordAccumulator.java:87-88
    // The following variables are only accessed by the sender thread, so we don't need to protect them.
    private final Set<TopicPartition> muted;

// RecordAccumulator.java:312-338
                // 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;
                    }
                }

                if (buffer == null) {
                    int size = Math.max(this.batchSize, AbstractRecords.estimateSizeInBytesUpperBound(
                            RecordBatch.CURRENT_MAGIC_VALUE, 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();
                }

// MemoryRecordsBuilder.java:886-890
    public boolean isFull() {
        // note that the write limit is respected only after the first record is added which ensures we can always
        // create non-empty batches (this is used to disable batching when the producer's batch size is set to 0).
        return appendStream == CLOSED_STREAM || (this.numRecords > 0 && this.writeLimit <= estimatedBytesWritten());
    }

// RecordAccumulator.java:398-401
        dq.addLast(batch);
        incomplete.add(batch);

        return new RecordAppendResult(future, dq.size() > 1 || batch.isFull(), true, batch.estimatedSizeInBytes());

// RecordAccumulator.java:608-623
    private long batchReady(boolean exhausted, TopicPartition part, Node leader,
                            long waitedTimeMs, boolean backingOff, int backoffAttempts,
                            boolean full, long nextReadyCheckDelayMs, Set<Node> readyNodes) {
        if (!readyNodes.contains(leader) && !isMuted(part)) {
            long timeToWaitMs = backingOff ? retryBackoff.backoff(backoffAttempts > 0 ? backoffAttempts - 1 : 0) : lingerMs;
            boolean expired = waitedTimeMs >= timeToWaitMs;
            boolean transactionCompleting = transactionManager != null && transactionManager.isCompleting();
            boolean sendable = full
                    || expired
                    || exhausted
                    || closed
                    || flushInProgress()
                    || transactionCompleting;
            if (sendable && !backingOff) {
                readyNodes.add(leader);
            } else {

// RecordAccumulator.java:666
        boolean exhausted = this.free.queued() > 0;

// BufferPool.java:49-55
    private final long totalMemory;
    private final int poolableSize;
    private final ReentrantLock lock;
    private final Deque<ByteBuffer> free;
    private final Deque<Condition> waiters;
    /** Total available memory is the sum of nonPooledAvailableMemory and the number of byte buffers in free * poolableSize.  */
    private long nonPooledAvailableMemory;

// BufferPool.java:127-134
            // now check if the request is immediately satisfiable with the
            // memory on hand or if we need to block
            int freeListSize = freeSize() * this.poolableSize;
            if (this.nonPooledAvailableMemory + freeListSize >= size) {
                // we have enough unallocated or pooled memory to immediately
                // satisfy the request, but need to allocate the buffer
                freeUp(size);
                this.nonPooledAvailableMemory -= size;

// BufferPool.java:144-164
                    while (accumulated < size) {
                        long startWaitNs = time.nanoseconds();
                        long timeNs;
                        boolean waitingTimeElapsed;
                        try {
                            waitingTimeElapsed = !moreMemory.await(remainingTimeToBlockNs, TimeUnit.NANOSECONDS);
                        } finally {
                            long endWaitNs = time.nanoseconds();
                            timeNs = Math.max(0L, endWaitNs - startWaitNs);
                            recordWaitTime(timeNs);
                        }

                        if (this.closed)
                            throw new KafkaException("Producer closed while allocating memory");

                        if (waitingTimeElapsed) {
                            this.metrics.sensor("buffer-exhausted-records").record();
                            throw new BufferExhaustedException("Failed to allocate " + size + " bytes within the configured max blocking time "
                                + maxTimeToBlockMs + " ms. Total memory: " + totalMemory() + " bytes. Available memory: " + availableMemory()
                                + " bytes. Poolable size: " + poolableSize() + " bytes");
                        }

// BufferPool.java:260-268
    public void deallocate(ByteBuffer buffer, int size) {
        lock.lock();
        try {
            if (size == this.poolableSize && size == buffer.capacity()) {
                buffer.clear();
                this.free.add(buffer);
            } else {
                this.nonPooledAvailableMemory += size;
            }

SEND seq=0 partition=0 elapsedMs=17 futureDone=false
SEND seq=1 partition=1 elapsedMs=0 futureDone=false
SEND seq=2 partition=2 elapsedMs=5009 futureDone=true
SEND seq=3 partition=3 elapsedMs=5001 futureDone=true
SEND seq=4 partition=4 elapsedMs=8 futureDone=false
SEND seq=5 partition=5 elapsedMs=0 futureDone=false
SEND seq=6 partition=6 elapsedMs=5 futureDone=false
SEND seq=7 partition=7 elapsedMs=0 futureDone=false
SEND seq=8 partition=0 elapsedMs=5002 futureDone=true
SEND seq=9 partition=1 elapsedMs=5000 futureDone=true
SUMMARY lingerMs=10000 maxBlockMs=5000 bufferMemory=32768 okSends=24 maxSendMs=5011 totalSendMs=45087 thrown=none

FUTURE-FAILED seq=4 p=4 elapsedMs=3013 type=org.apache.kafka.clients.producer.BufferExhaustedException
FUTURE-FAILED-MSG Failed to allocate 16384 bytes within the configured max blocking time 3000 ms. Total memory: 32768 bytes. Available memory: 0 bytes. Poolable size: 16384 bytes
FUTURE-FAILED seq=5 p=5 elapsedMs=3011 type=org.apache.kafka.clients.producer.BufferExhaustedException
FUTURE-FAILED-MSG Failed to allocate 16384 bytes within the configured max blocking time 3000 ms. Total memory: 32768 bytes. Available memory: 0 bytes. Poolable size: 16384 bytes

SEND seq=7 p=7 elapsedMs=1006 future pending

同时能容纳的批次数 ≈ buffer.memory / batch.size
每个分区至少要的批次数 = 1（一个在发，一个在攒，实际按 2 估）
可用分区数 ≈ (buffer.memory / batch.size) / 2