Copilot commented on code in PR #23663:
URL: https://github.com/apache/kafka/pull/23663#discussion_r4156346337
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TxnPartitionEntry.java:
##########
@@ -133,7 +171,17 @@ int maybeUpdateLastAckedSequence(int sequence) {
}
void removeInFlightBatch(ProducerBatch batch) {
- inflightBatchesBySequence.remove(batch);
+ if (!inflightBatchesBySequence.remove(batch)) {
+ return;
+ }
+ if (!inflightBatchesBySequence.isEmpty()
+ && batch.baseSequence() >
inflightBatchesBySequence.first().baseSequence()) {
Review Comment:
Sequence numbers wrap from `Integer.MAX_VALUE` to 0, so numeric ordering
does not indicate send order. If an older batch near `Integer.MAX_VALUE` is
still in flight and a newer wrapped batch at sequence 0 completes, this
comparison is false and the completed batch is not counted; additional sends
can then evict the older batch from the broker window. Conversely, completing
the high-sequence batch first can record an older batch as being ahead. Track
batch send order with wrap-aware state rather than comparing raw base sequences.
##########
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 new `continue` leaves the leader in `ReadyCheckResult.readyNodes` even
though no batch is drained. `Sender.sendProducerData` forces `pollTimeout` to 0
whenever that set is nonempty, so the expected failover case can busy-spin for
up to `request.timeout.ms` while the oldest request is unresolved. Make the
readiness calculation exclude deduplication-window-blocked partitions, or
otherwise communicate an empty drain back to `Sender` so it can block for
network I/O.
--
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]
