《Java 消息中间件·可靠消息与消费治理》10.1 顺序的粒度与代价

// RecordAccumulator.java:496-505
    public void reenqueue(ProducerBatch batch, long now) {
        batch.reenqueued(now);
        Deque<ProducerBatch> deque = getOrCreateDeque(batch.topicPartition);
        synchronized (deque) {
            if (transactionManager != null)
                insertInSequenceOrder(deque, batch);
            else
                deque.addFirst(batch);
        }
    }

// RecordAccumulator.java:511-539（节选）
        Deque<ProducerBatch> dq = bigBatch.split(targetSplitBatchSize);
        int numSplitBatches = dq.size();
        Deque<ProducerBatch> partitionDequeue = getOrCreateDeque(bigBatch.topicPartition);
        while (!dq.isEmpty()) {
            ProducerBatch batch = dq.pollLast();
            incomplete.add(batch);
            // We treat the newly split batches as if they are not even tried.
            synchronized (partitionDequeue) {
                if (transactionManager != null) {
                    // We should track the newly created batches since they already have assigned sequences.
                    transactionManager.addInFlightBatch(batch);
                    insertInSequenceOrder(partitionDequeue, batch);
                } else {
                    partitionDequeue.addFirst(batch);
                }
            }
        }

// ProducerConfig.java:291-293
            + "Allowing retries while setting <code>enable.idempotence</code> to <code>false</code> and <code>" + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "</code> 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.";

// RecordAccumulator.java:849-857
            // If the queued batch already has an assigned sequence, then it is being retried.
            // In this case, we wait until the next immediate batch is ready and drain that.
            // We only move on when the next in line batch is complete (either successfully or due to
            // a fatal broker error). This effectively reduces our in flight request count to 1.
            return firstInFlightSequence != RecordBatch.NO_SEQUENCE && first.hasSequence()
                    && first.baseSequence() != firstInFlightSequence;

RESULT inFlight=1 idem=false records=200 consumed=200 sendMs=133 failed=0 inversions=0 descents=0 duplicateValues=0 missingValues=0
RESULT inFlight=2 idem=false records=200 consumed=200 sendMs=109 failed=0 inversions=5044 descents=1 duplicateValues=0 missingValues=0
FIRST-INVERSION sendSeq=102 offset=50 before sendSeq=50 offset=147
RESULT inFlight=5 idem=false records=200 consumed=200 sendMs=94 failed=0 inversions=5243 descents=2 duplicateValues=0 missingValues=0
FIRST-INVERSION sendSeq=199 offset=0 before sendSeq=0 offset=1

0–49, 102–198, 50–101, 199

199, 0–49, 102–198, 50–101

[kafka-producer-network-thread | producer-1] WARN org.apache.kafka.clients.producer.internals.Sender - [Producer clientId=producer-1] Got error produce response with correlation id 9 on topic-partition v2c10.ord.dbg-0, retrying (2147483646 attempts left). Error: OUT_OF_ORDER_SEQUENCE_NUMBER
[kafka-producer-network-thread | producer-1] WARN org.apache.kafka.clients.producer.internals.Sender - [Producer clientId=producer-1] Got error produce response with correlation id 60 on topic-partition v2c10.ord.dbg-0, retrying (2147483595 attempts left). Error: OUT_OF_ORDER_SEQUENCE_NUMBER

RESULT inFlight=5 idem=true records=60 consumed=60 sendMs=101 failed=0 inversions=0 descents=0 duplicateValues=0 missingValues=0
```这是**拆批机制与幂等在途组合的边界**，不是幂等本身的语义问题：`MESSAGE_TOO_LARGE` 拆出来的批次继承了序列号，多批同时在途时序列号推进对不上，生产者宁可重试到投递超时也不跳号。本文没有继续挖根因，只把可复现的现象如实记下。

工程含义很明确：不要用「超过 broker 上限再靠拆批」当常规路径。消息大小应该在发送前就被约束住（11.1 节展开），让 `MESSAGE_TOO_LARGE` 只作为异常兜底。

这个边界必须说清楚，否则容易被误读：**触发条件是 `max.message.bytes` 被人为压到 1400 字节，导致批次被拆。** 正常配置下（1 MiB 上限、消息远小于上限），批次不会因为超限被拆开，`max.in.flight=5` + 幂等是 kafka-clients 的默认组合，也是推荐组合，不会出现上面的现象。别把这个实验读成「幂等不能配在途 5」——那等于关掉一个正确的默认值。

## 实验二：暂停 broker 四秒

拆批是人为构造的触发器。更接近真实故障的是 broker 短暂不可用：请求超时、客户端重连、重试。

### 设置

- 同样的单分区主题，`max.message.bytes` 用默认值，不触发拆批。
- 200 条同 key 消息，每 40 ms 发一条，总共约 8 秒。
- 发到第 3 秒时 `docker pause` 容器，4 秒后 `docker unpause`。
- `request.timeout.ms=3000`，因此在暂停期间的在途请求会超时重连。
- 三档在途 × 幂等开关各跑 3 次。

### 结果

本机实测（同上环境；每组 3 次，三次结果一致）：

| 在途 | 幂等 | 超时请求 | 消费条数 | 重复值 | 逆序对 | 相邻降序 |
|---|---|---|---|---|---|---|
| 1 | false | 1 | 201 | 1 | 0 | 0 |
| 2 | false | 2 | 202 | 2 | 2 | 1 |
| 5 | false | 5 | 205 | 5 | 20 | 4 |
| 1 | true | 1 | 200 | 0 | 0 | 0 |
| 2 | true | 2 | 200 | 0 | 0 | 0 |
| 5 | true | 5 | 200 | 0 | 0 | 0 |

原始输出（在途 5 的两组）：

```text
RESULT inFlight=5 idem=false records=200 consumed=205 sendMs=10457 failed=0 inversions=20 descents=4 duplicateValues=5 missingValues=0
FIRST-INVERSION sendSeq=54 offset=54 before sendSeq=53 offset=62
RESULT inFlight=5 idem=true records=200 consumed=200 sendMs=10034 failed=0 inversions=0 descents=0 duplicateValues=0 missingValues=0

idem=false ms=500 recPerSec=400000 | idem=true ms=478 recPerSec=418410
idem=false ms=494 recPerSec=404858 | idem=true ms=540 recPerSec=370370
idem=false ms=512 recPerSec=390625 | idem=true ms=426 recPerSec=469484

int workers = 4;
ExecutorService[] pool = new ExecutorService[workers];
for (int i = 0; i < workers; i++) {
    pool[i] = Executors.newSingleThreadExecutor();
}
// 同一个 key 永远落到同一个 worker
int idx = Math.floorMod(key.hashCode(), workers);
pool[idx].submit(() -> handle(record));