smjn commented on code in PR #22802:
URL: https://github.com/apache/kafka/pull/22802#discussion_r3569573568
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -379,29 +402,55 @@ public ProduceRequestData.TopicProduceData
topicProduceData() {
// Records have already been resolved (including any remote
storage reads) before this
// handler was added to the node map, so no blocking fetch happens
on the sender thread here.
Map<Long, Record> originalRecordData = resolvedRecordData;
+ int maxMessageBytes = dlqTopicMaxMessageBytes();
+ // In most cases the offset range is a single offset (see
DLQ_MAX_FETCH_BYTES). Track the
+ // running batch size incrementally via
DefaultRecord.sizeInBytes() - the same formula
+ // MemoryRecords itself uses - instead of re-serializing the whole
batch-so-far on every
+ // offset, which would make this loop quadratic in the number of
offsets.
List<SimpleRecord> simpleRecords = new ArrayList<>();
- for (long i = param.firstOffset(); i <= param.lastOffset(); i++) {
+ int batchSize = DefaultRecordBatch.RECORD_BATCH_OVERHEAD;
+ Long baseTimestamp = null;
+ for (long offset = nextOffsetToSend; offset <= param.lastOffset();
offset++) {
long timestamp = time.hiResClockMs();
ByteBuffer key = null;
ByteBuffer value = null;
- Record record = originalRecordData.get(i);
+ Record record = originalRecordData.get(offset);
if (record != null) {
key = record.hasKey() ? record.key() : null;
value = record.hasValue() ? record.value() : null;
}
- simpleRecords.add(new SimpleRecord(timestamp, key, value,
headers(i)));
+ Header[] recordHeaders = headers(offset);
+ if (baseTimestamp == null) {
+ baseTimestamp = timestamp;
+ }
+ int recordSize =
DefaultRecord.sizeInBytes(simpleRecords.size(), timestamp - baseTimestamp, key,
value, recordHeaders);
+
+ if (batchSize + recordSize > maxMessageBytes &&
!simpleRecords.isEmpty()) {
+ // Adding this record would exceed the limit and the batch
already has at least one
+ // record - stop here and send the rest in a follow-up
request.
+ break;
+ }
+ simpleRecords.add(new SimpleRecord(timestamp, key, value,
recordHeaders));
+ batchSize += recordSize;
+ if (batchSize > maxMessageBytes) {
+ // A single record (with its DLQ headers) already exceeds
the limit on its own;
+ // nothing to be gained by holding it back, so send it and
let the broker
+ // enforce/report the ultimate limit for this one, rather
than stalling forever.
+ break;
+ }
}
+ lastOffsetIncludedThisRound = nextOffsetToSend +
simpleRecords.size() - 1;
Review Comment:
Thanks @apoorvmittal10 for the insightful comment. The calculation above is
being performed on simpleRecords which contains the records destined to the DLQ
topic - which means it contains all records which have been dlq'ed (they may or
may not contain key/value)
So this code works regardless.
--
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]
