sudeshwasnik commented on code in PR #12462:
URL: https://github.com/apache/kafka/pull/12462#discussion_r940920055
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java:
##########
@@ -273,26 +273,29 @@ public RecordAppendResult append(String topic,
// check if we have an in-progress batch
Deque<ProducerBatch> dq =
topicInfo.batches.computeIfAbsent(effectivePartition, k -> new ArrayDeque<>());
+ RecordAppendResult appendResult;
synchronized (dq) {
// After taking the lock, validate that the partition
hasn't changed and retry.
if
(topicInfo.builtInPartitioner.isPartitionChanged(partitionInfo)) {
log.trace("Partition {} for topic {} switched by a
concurrent append, retrying",
partitionInfo.partition(), topic);
continue;
}
- RecordAppendResult appendResult = tryAppend(timestamp,
key, value, headers, callbacks, dq, nowMs);
- if (appendResult != null) {
+ appendResult = tryAppend(timestamp, key, value, headers,
callbacks, dq, nowMs);
+ if (appendResult != null && !appendResult.newBatchCreated)
{
Review Comment:
@artemlivshits QQ -
[updatePartitionInfo](https://github.com/apache/kafka/blob/36bfc3a254c71498830188b207abfdc420525397/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java#L285)
was previously getting updated only when this [non-null RecordAppendResult was
returned](https://github.com/apache/kafka/blob/36bfc3a254c71498830188b207abfdc420525397/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java#L400)
ie when
1. Non-null ProducerBatch existed in deque.
2. There was room to add this record in that ProducerBatch.
With this change -
[topicInfo.builtInPartitioner.updatePartitionInfo](https://github.com/apache/kafka/blob/36bfc3a254c71498830188b207abfdc420525397/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java#L400)
will be called only for the above cases. appendResult.newBatchCreated is
marked true only when ProducerBatch is found `null`.
So invocation of `topicInfo.builtInPartitioner.updatePartitionInfo` will
experience no change.
before -> `null` ProducerBatch would result in abortOnNewBatch and finally
update
[builtInPartitioner](https://github.com/apache/kafka/blob/36bfc3a254c71498830188b207abfdc420525397/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java#L319),
with this change, builtInPartitioner will also get updated the same place.
(with appropriate appendResult.appendedBytes)
--
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]
