frankvicky commented on code in PR #23663:
URL: https://github.com/apache/kafka/pull/23663#discussion_r4156423882
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java:
##########
@@ -1010,6 +1010,18 @@ private List<ProducerBatch>
drainBatchesForOneNode(MetadataSnapshot metadataSnap
if (shouldBackoff(first.hasLeaderChangedForTheOngoingRetry(),
first, first.waitedTimeMs(now)))
continue;
+ // Don't assign a sequence to a new batch while doing so could
push the partition's oldest in-flight
+ // batch out of the broker's deduplication window.
`max.in.flight.requests.per.connection` does not
+ // bound this on its own: across a leader change a batch can
still be in flight to the previous
+ // leader while later batches are drained to, and acknowledged
by, the current leader. Once enough
+ // of them have been appended, the broker forgets the oldest
batch and its retry fails with
+ // OUT_OF_ORDER_SEQUENCE. Retries are exempt because they
already have a sequence and are already
+ // counted, and they are serialised below by the
`firstInFlightSequence` check.
+ if (transactionManager != null && !first.hasSequence()
+ &&
transactionManager.wouldExceedBrokerDeduplicationWindow(tp)) {
+ continue;
Review Comment:
This follows the existing pattern: `hasUnresolvedSequence` and
`hasStaleProducerIdAndEpoch` stop the drain in the same way and are not
excluded from `ready()` either.
--
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]
