《Java 消息中间件·Kafka 内核与客户端》7.3 Sender 与 NetworkClient

// Sender.java:240-251
    @Override
    public void run() {
        log.debug("Starting Kafka producer I/O thread.");

        // main loop, runs until close is called
        while (running) {
            try {
                runOnce();
            } catch (Exception e) {
                log.error("Uncaught error in kafka producer I/O thread: ", e);
            }
        }

// Sender.java:258
        while (!forceClose && ((this.accumulator.hasUndrained() || this.client.inFlightRequestCount() > 0) || hasPendingTransactionalRequests())) {

// Sender.java:267
        while (!forceClose && transactionManager != null && transactionManager.hasOngoingTransaction()) {

// Sender.java:343-346
        long currentTimeMs = time.milliseconds();
        long pollTimeout = sendProducerData(currentTimeMs);
        client.poll(pollTimeout, currentTimeMs);
    }

// Sender.java:380-383
    private long sendProducerData(long now) {
        MetadataSnapshot metadataSnapshot = metadata.fetchMetadataSnapshot();
        // get the list of partitions with data ready to send
        RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(metadataSnapshot, now);

// Sender.java:417-426
        // create produce requests
        Map<Integer, List<ProducerBatch>> batches = this.accumulator.drain(metadataSnapshot, result.readyNodes, this.maxRequestSize, now);
        addToInflightBatches(batches);
        if (guaranteeMessageOrder) {
            // Mute all the partitions drained
            for (List<ProducerBatch> batchList : batches.values()) {
                for (ProducerBatch batch : batchList)
                    this.accumulator.mutePartition(batch.topicPartition);
            }
        }

// RecordAccumulator.java:889-893
                if (size + first.estimatedSizeInBytes() > maxSize && !ready.isEmpty()) {
                    // there is a rare case that a single batch size is larger than the request size due to
                    // compression; in this case we will still eventually send this batch in a single request
                    break;
                } else {

// Sender.java:896-905
    private void sendProduceRequest(long now, int destination, short acks, int timeout, List<ProducerBatch> batches) {
        if (batches.isEmpty())
            return;

        final Map<TopicPartition, ProducerBatch> recordsByPartition = new HashMap<>(batches.size());
        Map<String, Uuid> topicIds = topicIdsForBatches(batches);

        ProduceRequestData.TopicProduceDataCollection tpd = new ProduceRequestData.TopicProduceDataCollection();
        for (ProducerBatch batch : batches) {
            TopicPartition tp = batch.topicPartition;

// InFlightRequests.java:96-100
    public boolean canSendMore(String node) {
        Deque<NetworkClient.InFlightRequest> queue = requests.get(node);
        return queue == null || queue.isEmpty() ||
               (queue.peekFirst().send.completed() && queue.size() < this.maxInFlightRequestsPerConnection);
    }

Allowing retries while setting enable.idempotence to false and max.in.flight.requests.per.connection to greater than 1 will potentially change the
ordering of records because if two batches are sent to a single partition, and the first fails and is retried but the second
succeeds, then the records in the second batch may appear first.

// ConnectionState.java:29（枚举定义；上面 23-28 行是逐态注释）
    DISCONNECTED, CONNECTING, CHECKING_API_VERSIONS, READY, AUTHENTICATION_FAILED;

// NetworkClient.java:530-533
    private boolean canSendRequest(String node, long now) {
        return connectionStates.isReady(node, now) && selector.isChannelReady(node) &&
            inFlightRequests.canSendMore(node);
    }

// NetworkClient.java:642-673（节选；630-640 的签名与 abortedSends 分支本处省略）
        long metadataTimeout = metadataUpdater.maybeUpdate(now);
        long telemetryTimeout = telemetrySender != null ? telemetrySender.maybeUpdate(now) : Integer.MAX_VALUE;
        try {
            this.selector.poll(Utils.min(timeout, metadataTimeout, telemetryTimeout, defaultRequestTimeoutMs));
        } catch (IOException e) {
            log.error("Unexpected error during I/O", e);
        }

        // process completed actions
        long updatedNow = this.time.milliseconds();
        List<ClientResponse> responses = new ArrayList<>();
        handleCompletedSends(responses, updatedNow);
        handleCompletedReceives(responses, updatedNow);
        handleDisconnections(responses, updatedNow);
        handleConnections();
        handleInitiateApiVersionRequests(updatedNow);
        handleTimedOutConnections(responses, updatedNow);
        handleTimedOutRequests(responses, updatedNow);
        handleRebootstrap(responses, updatedNow);
        completeResponses(responses);

        return responses;
    }

    private void completeResponses(List<ClientResponse> responses) {
        for (ClientResponse response : responses) {
            try {
                response.onComplete();
            } catch (Exception e) {
                log.error("Uncaught error in request completion:", e);
            }
        }

docker exec mq-b2 /opt/kafka/bin/kafka-topics.sh --bootstrap-server 127.0.0.1:9092 \
  --create --topic b2.reorder --partitions 1 --replication-factor 1 \
  --config max.message.bytes=1024

// Sender.java:676-688
        if (error == Errors.MESSAGE_TOO_LARGE && batch.recordCount > 1 && !batch.isDone() &&
                (batch.magic() >= RecordBatch.MAGIC_VALUE_V2 || batch.isCompressed())) {
            // If the batch is too large, we split the batch and send the split batches again. We do not decrement
            // the retry attempts in this case.
            log.warn(
                "Got error produce response in correlation id {} on topic-partition {}, splitting and retrying ({} attempts left). Error: {}",
                correlationId,
                batch.topicPartition,
                this.retries - batch.attempts(),
                formatErrMsg(response));
            if (transactionManager != null)
                transactionManager.removeInFlightBatch(batch);
            this.accumulator.splitAndReenqueue(batch);

FIRST-INVERSION sendSeq=0 offset=61 laterSendSeq=9 offset=60
RESULT maxInFlight=5 idempotent=false batchSize=2048 records=60 failed=0 inversions=9
OFFSETS 61,62,63,64,65,66,67,68,69,60,70,71,72,73,...

Disconnecting from node 1 due to request timeout.
Cancelled in-flight PRODUCE request with correlation id 8 due to node 1 being disconnected (elapsed time since create: 2043ms, ...)
Cancelled in-flight PRODUCE request with correlation id 9 due to node 1 being disconnected (elapsed time since create: 2038ms, ...)
Got error produce response with correlation id 8 on topic-partition b2.pause-0, retrying (4 attempts left). Error: ... 
Got error produce response with correlation id 9 on topic-partition b2.pause-0, retrying (4 attempts left). Error: ...

docker exec mq-b2 /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 \
  --topic b2.pause --from-beginning --timeout-ms 5000 \
  --formatter-property print.offset=true

Offset:0	v0
Offset:1	v1
Offset:2	v2
Offset:3	v3
Offset:4	v4
Offset:5	v4
Offset:6	v3
Offset:7	v5
Offset:8	v6
Offset:9	v7
Offset:10	v8
Offset:11	v9
Offset:12	v10
Offset:13	v11

Offset:0	v0
Offset:1	v1
...
Offset:11	v11

// ProducerConfig.java:265-267
    // max.in.flight.requests.per.connection should be less than or equal to 5 when idempotence producer enabled to ensure message ordering
    // The value 5 is aligned with ProducerStateEntry#NUM_BATCHES_TO_RETAIN.
    private static final int MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE = 5;