ethanresnick opened a new pull request, #23853: URL: https://github.com/apache/kafka/pull/23853
JIRA: https://issues.apache.org/jira/browse/KAFKA-14359 ## Problem An idempotent producer (the default) can send a partition's second batch while the partition's first batch of the producer epoch (sequence 0) is still in flight. If the first batch then fails with a retriable error that leaves it out of the log (`NOT_LEADER_OR_FOLLOWER`, `NOT_ENOUGH_REPLICAS`, a request timeout, a lost connection), the broker has no state for the producer and accepts the second batch as the first one (KAFKA-19880). Every retry of sequence 0 then gets `OUT_OF_ORDER_SEQUENCE_NUMBER`. `TransactionManager.canRetry` retries it, because it is not the next sequence after the last acked one, until `delivery.timeout.ms` expires. The records of the first batch are lost, and the producer stalls on them for up to `delivery.timeout.ms`. The common trigger is producing right after `createTopics()`: the metadata already shows leaders, but the partitions are still loading. With a new 64-partition topic, 3,000 records of 10 KB, gzip, and 1,024 keys, the stock 4.3.1 client stalled in 19 of 20 runs against a fresh 4.1.0 broker and in 6 of 6 runs against a fresh 4.3.1 broker. The usual workaround is to call `Admin.listOffsets()` for every partition before the first send. A TRACE log of one stalled partition (4.3.1 client): ``` 06:29:27.530 WARN Sender - Got error produce response with correlation id 4 on topic-partition hang-jvmtr-1-8167-45, retrying (2147483646 attempts left). Error: NOT_LEADER_OR_FOLLOWER 06:29:27.544 TRACE TransactionManager - ProducerId: 15439; Set last ack'd sequence number for topic-partition hang-jvmtr-1-8167-45 to 1 06:29:27.778 WARN Sender - Got error produce response with correlation id 15 on topic-partition hang-jvmtr-1-8167-45, retrying (2147483645 attempts left). Error: OUT_OF_ORDER_SEQUENCE_NUMBER 06:29:27.783 TRACE TransactionManager - ProducerId: 15439; Set last ack'd sequence number for topic-partition hang-jvmtr-1-8167-45 to 2 06:29:27.961 WARN Sender - Got error produce response with correlation id 33 on topic-partition hang-jvmtr-1-8167-45, retrying (2147483644 attempts left). Error: OUT_OF_ORDER_SEQUENCE_NUMBER ``` Sequence 0 fails with `NOT_LEADER_OR_FOLLOWER`, sequence 1 is acknowledged 14 ms later, and every retry of sequence 0 then fails with `OUT_OF_ORDER_SEQUENCE_NUMBER` until the delivery timeout. ## Change While a partition's first batch of the current producer epoch is in flight and not acknowledged, the accumulator assigns no sequence to a new batch of that partition: - `TransactionManager.isAwaitingFirstAck(tp)`: the partition has in-flight batches, and `lastAckedSequence(tp)` is empty. - `RecordAccumulator.drainBatchesForOneNode` skips such a partition's new batch with `continue`. It does not stop the drain: `shouldStopDrainBatchesForPartition` makes the loop `break`, and the drain stores that partition as the next start index, so every other partition on the node would also wait for the first ack. - `RecordAccumulator.partitionReady` does not count such a partition as ready, as it does not count a muted one. Without this, the sender sees a ready node, drains nothing from it, and polls with a zero timeout until the ack arrives (a busy loop). Retries already have a sequence, so they are not affected; the existing `firstInFlightSequence` check still orders them. After the first ack, pipelining works as before, up to `max.in.flight.requests.per.connection`. ## Why it is safe - It only holds back a batch that has no sequence yet. No sequence, epoch, or retry logic changes. - The wait always ends: the in-flight first batch completes, fails, or expires. On expiry, the existing unresolved-sequence and epoch-bump logic runs as before. - It applies once per partition per producer epoch. For the idempotent producer, that is once per partition for the producer's life (and after each epoch bump). With transactions v2 (KIP-890), the epoch bumps at every commit, so it applies to each transaction's second batch per partition. The cost is up to one round trip before that batch. - Other partitions are not held back (the `continue`, and a test for it). ## Measurements (transactions v2) Broker 4.3.1 (one node, `transaction.version` 2, local Docker), Apple M4 Pro. 1 KB values, keys spread over the partitions. A transaction sends its records at once, then commits; latency is begin to commit. The idempotent runs send 100,000 records, then time 2,000 single-record sends. Medians of 5 rounds; the rule on and off alternate within each round. The JVM client is 4.3.1 with this change compiled into its `RecordAccumulator`, switched by a system property for the measurement only. | Workload | Partitions | Records/txn | records/s with | records/s without | p50 ms with | p50 ms without | p99 ms with | p99 ms without | | --- | --- | --- | --- | --- | --- | --- | --- | --- | | transactional | 1 | 10 | 8,303 | 8,132 | 1.10 | 1.12 | 2.18 | 2.44 | | transactional | 64 | 10 | 3,320 | 3,303 | 1.21 | 1.26 | 26.96 | 28.12 | | transactional | 1 | 2,000 | 111,233 | 113,109 | 16.68 | 16.41 | 38.70 | 38.84 | | transactional | 64 | 2,000 | 210,611 | 196,150 | 8.98 | 8.31 | 20.10 | 25.89 | | idempotent | 1 | - | 97,705 | 99,145 | 6.56 | 6.50 | 16.48 | 12.71 | | idempotent | 64 | - | 229,271 | 197,608 | 6.64 | 6.66 | 16.27 | 13.81 | Small transactions have one batch per partition, so nothing waits. Large transactions wait about one round trip per transaction; the medians differ by less than the run-to-run noise on this machine. The same rule in a WebAssembly build of the client (where a busy loop is much more expensive) showed no difference above the noise either: for 2,000-record transactions, 0.88-1.15 times the throughput without the change, over 15 rounds. ## Relation to other issues and PRs - **KAFKA-19880 / #22619** (broker): makes the broker reject a non-zero first sequence when it has no producer state. With it, the second batch in the scenario above fails instead of being appended, and the first batch's retry succeeds. This PR fixes the client side for brokers without #22619 (all released brokers today), and it avoids a wasted round of errors and retries with brokers that have it. The two changes are independent. - **KAFKA-15591** (fixed for 4.5.0): the broker rejects a non-zero first sequence on an empty log. That covers new topics on new brokers, but not a partition with records and no producer state (for example after the producer state expired). - **KAFKA-18905 / #23663** (open): adds a similar new-batch check (`wouldExceedBrokerDeduplicationWindow`) in the same loop, also as a `continue`. The two conditions are independent, and the patches touch different lines: #23663 applies cleanly on top of this PR, and this PR applies cleanly on top of #23663 (checked with `git apply --check` both ways). If both are merged, the two `TransactionManager` checks could become one method ("may this partition get a new sequence now?"). `testEpochBumpWhenOutOfOrderBatchesRetriedAndFirstBatchExpires` (the KAFKA-15591 test) now starts after one acknowledged batch; #23663 does not change that test. - **KAFKA-9199**: the other approach, failing or epoch-bumping a batch whose `OUT_OF_ORDER_SEQUENCE_NUMBER` cannot resolve. This PR prevents that state instead; the two can coexist. ## Tests New: - `SenderTest.testNoNewBatchSentUntilFirstBatchIsAcknowledged` (`NOT_LEADER_OR_FOLLOWER`, `NOT_ENOUGH_REPLICAS`, `REQUEST_TIMED_OUT`): the second batch is not sent while sequence 0 is in flight or retried, and both are delivered in order. - `SenderTest.testPartitionAwaitingFirstAckDoesNotBlockOtherPartitions`. - `RecordAccumulatorTest.testPartitionAwaitingFirstAckDoesNotStarveOtherPartitions`: the drain skips only the waiting partition. - `RecordAccumulatorTest.testPartitionAwaitingFirstAckIsNotReady`: a waiting partition does not make its leader ready. All four new tests (eight cases with their parameters) fail without the `RecordAccumulator` change. Changed: 11 existing `SenderTest` tests sent several batches before the first acknowledgement, which this change prevents. Ten of them now start with `sendAndAcknowledgeFirstBatch(tp0)` (a new helper), and their expected sequences move up by one until an epoch bump resets them. Their scenarios are unchanged otherwise: `testIdempotenceWithMultipleInflights`, `testIdempotenceWithMultipleInflightsRetriedInOrder`, `testIdempotenceWithMultipleInflightsWhereFirstFailsFatallyAndSequenceOfFutureBatchesIsAdjusted`, `testCorrectHandlingOfOutOfOrderResponses`, `testCorrectHandlingOfOutOfOrderResponsesWhenSecondSucceeds`, `testCorrectHandlingOfDuplicateSequenceError`, `testExpiryOfFirstBatchShouldNotCauseUnresolvedSequencesIfFutureBatchesSucceed`, `testExpiryOfFirstBatchShouldCauseEpochBumpIfFutureBatchesFail`, `testEpochBumpWhenOutOfOrderBatchesRetriedAndFirstBatchExpires`, `testUnresolvedSequencesAreNotFatal`. `testAwaitPendingRecordsBeforeCommittingTransaction` now acknowledges the first produce request before it expects the second one. Run locally on trunk 6556669 with JDK 25: `./gradlew :clients:test --tests 'org.apache.kafka.clients.producer.*'` (620 tests, all pass), `:clients:checkstyleMain :clients:checkstyleTest`, and `:clients:spotbugsMain` (all pass). Not run locally: the system tests. Outside Kafka's tree, the change was also tested against real brokers (a 64- and 256-partition new-topic race: the stock client stalled in every run, the patched one in none; and leader moves, lost responses, and a broker restart on a 3-broker cluster: no loss, duplicate, or reorder). Co-Authored-By: Claude Opus 5.5 <[email protected]> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
