apoorvmittal10 commented on code in PR #17870:
URL: https://github.com/apache/kafka/pull/17870#discussion_r1908497961
##########
core/src/main/java/kafka/server/share/DelayedShareFetch.java:
##########
@@ -207,20 +212,21 @@ LinkedHashMap<TopicIdPartition,
FetchRequest.PartitionData> acquirablePartitions
LinkedHashMap<TopicIdPartition, FetchRequest.PartitionData>
topicPartitionData = new LinkedHashMap<>();
sharePartitions.forEach((topicIdPartition, sharePartition) -> {
- int partitionMaxBytes =
shareFetch.partitionMaxBytes().getOrDefault(topicIdPartition, 0);
// Add the share partition to the list of partitions to be fetched
only if we can
// acquire the fetch lock on it.
if (sharePartition.maybeAcquireFetchLock()) {
try {
// If the share partition is already at capacity, we
should not attempt to fetch.
if (sharePartition.canAcquireRecords()) {
+ // We do not know the total partitions that can be
acquired at this stage, hence we set maxBytes
+ // to 0 for now and will update it before doing the
replica manager fetch.
topicPartitionData.put(
topicIdPartition,
new FetchRequest.PartitionData(
topicIdPartition.topicId(),
sharePartition.nextFetchOffset(),
0,
- partitionMaxBytes,
+ 0,
Review Comment:
I was thinking some other default value like 1MB but if you delay the
creation of FetchRequest.PartitionData to the time when maxBytes is known then
anyways it ll not be required.
--
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
