You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by di...@apache.org on 2023/11/07 08:53:43 UTC
(kafka) branch 3.5 updated: KAFKA-15771: fix concurrency bug in ProduceRequest#partitionSizes() (#14674)
This is an automated email from the ASF dual-hosted git repository.
divijv pushed a commit to branch 3.5
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.5 by this push:
new 5829fca0a7f KAFKA-15771: fix concurrency bug in ProduceRequest#partitionSizes() (#14674)
5829fca0a7f is described below
commit 5829fca0a7f65cd0b0c8f8afdb5a28124d4ae7b7
Author: Xiaobing Fang <bi...@qq.com>
AuthorDate: Tue Nov 7 16:44:07 2023 +0800
KAFKA-15771: fix concurrency bug in ProduceRequest#partitionSizes() (#14674)
A commit fixes a bug in ProduceRequest#partitionSizes() that may cause this method to incorrectly returning an empty or incomplete response for a thread when another thread is in the process of initialising it.
Reviewers: Divij Vaidya <di...@amazon.com>, hudeqi <12...@qq.com>, vamossagar12 <sa...@gmail.com>
--------------------------------
Co-authored-by: fangxiaobing <fa...@kuaishou.com>
---
.../main/java/org/apache/kafka/common/requests/ProduceRequest.java | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java
index 758631a1d87..78b59463026 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java
@@ -133,14 +133,15 @@ public class ProduceRequest extends AbstractRequest {
// this method may be called by different thread (see the comment on data)
synchronized (this) {
if (partitionSizes == null) {
- partitionSizes = new HashMap<>();
+ Map<TopicPartition, Integer> tmpPartitionSizes = new HashMap<>();
data.topicData().forEach(topicData ->
topicData.partitionData().forEach(partitionData ->
- partitionSizes.compute(new TopicPartition(topicData.name(), partitionData.index()),
+ tmpPartitionSizes.compute(new TopicPartition(topicData.name(), partitionData.index()),
(ignored, previousValue) ->
partitionData.records().sizeInBytes() + (previousValue == null ? 0 : previousValue))
)
);
+ partitionSizes = tmpPartitionSizes;
}
}
}