You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/08/09 05:45:15 UTC
[GitHub] [kafka] sudeshwasnik commented on a diff in pull request #12462: [KAFKA-9965] Fix accumulator tryAppend, so that fresh new producerBatch is created
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org