DL1231 commented on code in PR #20837:
URL: https://github.com/apache/kafka/pull/20837#discussion_r2543997891
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -858,8 +871,16 @@ public ShareAcquiredRecords acquire(
// In record_limit mode, we need to ensure that we do not
acquire more than
// maxRecordsToAcquire. Hence, pass the remaining number
of records that can
// be acquired.
- int acquiredSubsetCount =
acquireSubsetBatchRecords(memberId, isRecordLimitMode, numRecordsRemaining,
firstBatch.baseOffset(), lastOffsetToAcquire, inFlightBatch, result);
- acquiredCount += acquiredSubsetCount;
+ BadRecordMarkerAndAcquiredCount
badRecordMarkerAndAcquiredCount = acquireSubsetBatchRecords(memberId,
isRecordLimitMode,
+ numRecordsRemaining, firstBatch.baseOffset(),
lastOffsetToAcquire, inFlightBatch, result);
+
+ acquiredCount +=
badRecordMarkerAndAcquiredCount.acquiredCount();
+ // If a bad record is present, return immediately and set
`maxRecordsToAcquire = -1`
+ // to prevent acquiring any new records afterwards.
+ if (badRecordMarkerAndAcquiredCount.badRecordMarker()) {
Review Comment:
Given a batch with the following state:
<pre>
offset 0 1 2 3 4
deliveryCount 3 1 1 1 1
state ACQUIRED AVAILABLE AVAILABLE AVAILABLE
AVAILABLE
</pre>
In this case, the records actually acquired would be offsets 1-4.
Following your suggestion, since the acquiredCount would be greater than 0,
this acquisition would be throttled. However, I'm hesitant because the record
causing the high delivery count (offset 0) wasn't part of this acquisition.
What are your thoughts on this case? Should we still throttle it?
--
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]
