ijuma commented on code in PR #12096:
URL: https://github.com/apache/kafka/pull/12096#discussion_r860883998
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -175,11 +175,14 @@ private static class TopicPartitionEntry {
// responses which are due to the retention period elapsing, and those
which are due to actual lost data.
private long lastAckedOffset;
- private static final Comparator<ProducerBatch>
PRODUCER_BATCH_COMPARATOR = (b1, b2) -> {
- if (b1.baseSequence() < b2.baseSequence()) return -1;
- else if (b1.baseSequence() > b2.baseSequence()) return 1;
- else return Integer.compare(b1.hashCode(), b2.hashCode());
- };
+ // `inflightBatchesBySequence` should only have batches with the same
producer id and producer
+ // epoch, but there is an edge case where we may remove the wrong
batch if the comparator
+ // only takes `baseSequence` into account.
+ // See
https://github.com/apache/kafka/pull/12096#pullrequestreview-955554191 for
details.
+ private static final Comparator<ProducerBatch>
PRODUCER_BATCH_COMPARATOR =
+ Comparator.comparingLong(ProducerBatch::producerId)
+ .thenComparing(ProducerBatch::producerEpoch)
Review Comment:
I noticed after merging that this should have been `thenComparingInt` to
avoid unnecessary boxing. I fixed that in the cherry-picks, but it was too late
for the master change. I'll fix it in master via #12097.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
