You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "smjn (via GitHub)" <gi...@apache.org> on 2026/07/23 19:47:52 UTC
[PR] KAFKA-20723: Resolve DLQ copy records per produce round. [kafka]
smjn opened a new pull request, #22919:
URL: https://github.com/apache/kafka/pull/22919
* ShareGroupDLQStateManager previously resolved a DLQ record parameter's entire archived offset range once, eagerly, before the handler was even enqueued to send - capping total copyable content at roughly one message's worth (regardless of how many produce requests the range needed) and delaying the first produce request until the whole range, including any remote-tiered reads, had been read.
* Move to resolving one produce round at a time instead: resolveRound() now fetches only the window starting at nextOffsetToSend, with a fresh decompression budget per round, and handleProduceResponse() resolves the next round before rejoining the coalescing node map. This lets a large range recover real content across every round it needs, and lets the first produce request go out as soon as its own round resolves.
* Also harden two related failure paths surfaced while working through this: an uncaught exception from one handler's onComplete() no longer prevents other handlers sharing the same coalesced produce response from being notified (each is now wrapped via completeHandlerSafely(), with its own future explicitly failed if it throws), and resolveRound() no longer lets a synchronous exception from maybeFetchRecordData() escape into a RequestCompletionHandler callback, where depending on the response-delivery path it wasn't guaranteed to be caught before reaching the broker's fatal-error handling.
* Separately, in ShareGroupDLQRecordFetcher, tighten the decompression budget accounting the round-based design depends on: the fetch now stops once enough content has been collected without waiting to finish scanning a whole batch or a whole read (previously bounded only by maxFetchBytes, not the target budget), and the compressed-batch budget checks now account for bytes already collected from uncompressed batches too, not just prior compressed decompression - closing a gap where mixing both in one round could let compressed data claim close to a second full budget on top of what uncompressed data had already used.
--
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
Re: [PR] KAFKA-20723: Resolve DLQ copy records per produce round. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22919:
URL: https://github.com/apache/kafka/pull/22919#discussion_r3646168688
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcher.java:
##########
@@ -272,7 +294,13 @@ private long collectUncompressed(RecordBatch batch, long readFrom) {
if (record.offset() < readFrom) continue;
if (record.offset() > endOffset) return nextOffset;
recordMap.put(record.offset(), record);
+ collectedBytes += record.sizeInBytes();
Review Comment:
Yes, this was deliberate as the memory extension is bound to `max.message.bytes` and the produce side can handle this extra record.
--
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
Re: [PR] KAFKA-20723: Resolve DLQ copy records per produce round. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #22919:
URL: https://github.com/apache/kafka/pull/22919
--
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
Re: [PR] KAFKA-20723: Resolve DLQ copy records per produce round. [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #22919:
URL: https://github.com/apache/kafka/pull/22919#discussion_r3644282370
##########
server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java:
##########
@@ -1840,6 +1944,186 @@ public void testDLQRecordCopyEnabledFetchesOnceAcrossProduceRetries() throws Exc
}
}
+ @Test
+ public void testDlqRecordCopyRoundBudgetResetAcrossMultipleProduceRounds() throws Exception {
+ // Each source record's value is highly compressible (gzip) but decompresses to 400 bytes -
+ // comfortably within one round's decompression budget (maxDecompressedBytes, derived from
+ // dlqTopicMaxMessageBytes), but not within two rounds' worth combined. The resulting DLQ
+ // record (400-byte value plus DLQ headers) also exceeds maxMessageBytes on its own, which -
+ // via topicProduceData()'s single-record floor - forces exactly one offset per produce
+ // round, giving 3 rounds for param()'s 3-offset range.
+ //
+ // Contrast with the pre-per-round design: a single decompression budget shared across the
Review Comment:
nit: While this paragraph is interesting narrative for reviewing the PR, it's not helpful to leave it in the code for the future I feel.
--
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
Re: [PR] KAFKA-20723: Resolve DLQ copy records per produce round. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22919:
URL: https://github.com/apache/kafka/pull/22919#discussion_r3645020938
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcher.java:
##########
@@ -272,7 +294,13 @@ private long collectUncompressed(RecordBatch batch, long readFrom) {
if (record.offset() < readFrom) continue;
if (record.offset() > endOffset) return nextOffset;
recordMap.put(record.offset(), record);
+ collectedBytes += record.sizeInBytes();
Review Comment:
Yes, we are allowing that as the last record will still be capped by the max.delivery.limit. The producer side is smart enough to break it into 2 RPCs if needed.
--
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
Re: [PR] KAFKA-20723: Resolve DLQ copy records per produce round. [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #22919:
URL: https://github.com/apache/kafka/pull/22919#discussion_r3645091533
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcher.java:
##########
@@ -272,7 +294,13 @@ private long collectUncompressed(RecordBatch batch, long readFrom) {
if (record.offset() < readFrom) continue;
if (record.offset() > endOffset) return nextOffset;
recordMap.put(record.offset(), record);
+ collectedBytes += record.sizeInBytes();
Review Comment:
I didn't understand. Can the method return more data than the `maxDecompressedBytes`? If yes then shouldn't that be fixed irrespective of producer side smart logic?
--
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
Re: [PR] KAFKA-20723: Resolve DLQ copy records per produce round. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #22919:
URL: https://github.com/apache/kafka/pull/22919#issuecomment-5071412428
@apoorvmittal10 @AndrewJSchofield
Re-wrote the algorithm partially to simplify logic. Earlier we had to keep track of fresh budget, partial budget etc.
There was also the issue. Consider `max.message.bytes` = M. Consider 2 batches b1, b2 of decompressed size M-10, M then the previous code was discarding b2 completely skipping it which isn't correct.
In the new algorithm we track how many offsets have been scanned and return that as well as `lastResolvedOffset`.
for compressed case:
consider fetch for range [s, e]
case 0: FIRST BATCH, WITHIN LIMIT:
```
collectedBytes = 0
decompressed(b1) <= M =>
PICK entire batch;
collectedBytes = decompressed(b1) ;
lastResolvedOffset = b1.lastOffset,
nextOffset = b1.lastOffset + 1
NEXT ITER
```
case 1: FIRST BATCH, OUTSIDE LIMIT:
```
collectedBytes = 0
decompressed(b1) > M =>
SKIP entire batch;
collectedBytes = 0;
lastResolvedOffset = b1.lastOffset,
nextOffset max(b1.lastOffset + 1, e)
NEXT ITER
```
case 2: SOME DATA COLLECTED, NEW BATCH WITH VALID SINGLE LARGE RECORDS:
```
collectedBytes = C < M
decompressed(bi) <= M AND C + decompressed(bi) > M =>
PICK bi IN NEXT CALL TO FETCH;
deferred = true
FETCH ENDS
lastResolvedOffset returned to state manager
```
case 3: SOME DATA COLLECTED, NEW BATCH WITH VALID MULTIPLE RECORDS:
```
collectedBytes = C < M
decompressed(bi) <= M AND C + decompressed(bi) > M =>
for(record: bi) {
if(record.size + collectedBytes > M) {
deferred = true
return nextOffset
}
lastResolvedOffset=record.offset()
nextOffset=record.offset() + 1
collectedBytes += record.size
}
FETCH ENDS
lastResolvedOffset returned to state manager
```
--
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
Re: [PR] KAFKA-20723: Resolve DLQ copy records per produce round. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22919:
URL: https://github.com/apache/kafka/pull/22919#discussion_r3645020938
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcher.java:
##########
@@ -272,7 +294,13 @@ private long collectUncompressed(RecordBatch batch, long readFrom) {
if (record.offset() < readFrom) continue;
if (record.offset() > endOffset) return nextOffset;
recordMap.put(record.offset(), record);
+ collectedBytes += record.sizeInBytes();
Review Comment:
Yes, we are allowing that as the last record will still be capped by the max.delivery.limit.
--
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
Re: [PR] KAFKA-20723: Resolve DLQ copy records per produce round. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22919:
URL: https://github.com/apache/kafka/pull/22919#discussion_r3645020938
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcher.java:
##########
@@ -272,7 +294,13 @@ private long collectUncompressed(RecordBatch batch, long readFrom) {
if (record.offset() < readFrom) continue;
if (record.offset() > endOffset) return nextOffset;
recordMap.put(record.offset(), record);
+ collectedBytes += record.sizeInBytes();
Review Comment:
Yes, we are allowing that as the last record will still be capped by the max.delivery.limit. The producer side is smart enough to break it into 2 RPCs if needed.
--
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
Re: [PR] KAFKA-20723: Resolve DLQ copy records per produce round. [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #22919:
URL: https://github.com/apache/kafka/pull/22919#discussion_r3644893920
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcher.java:
##########
@@ -151,10 +163,14 @@ public CompletableFuture<Map<Long, Record>> fetch() {
* complete (local data, or remote data already resolved) the loop continues in place; when it is still
* pending (remote read in flight) the loop returns and is resumed from the callback - so the synchronous
* path never recurses and the async path resumes on a fresh stack (the remote storage reader thread).
+ *
+ * <p>Also stops once {@link #collectedBytes} reaches {@link #maxDecompressedBytes} - there's no need to
+ * keep reading once enough usable content has been collected for the caller's purposes (see the
+ * class-level doc), even if {@link #endOffset} hasn't been reached yet.
*/
private void runFrom(long startFrom) {
long nextOffset = startFrom;
- while (nextOffset <= endOffset) {
+ while (nextOffset <= endOffset && collectedBytes < maxDecompressedBytes) {
Review Comment:
Query: Should it be <=? Depends on whether maxDecompressedBytes limit is inclusive or exclusive? I think on topic it should be inclusive.
```suggestion
while (nextOffset <= endOffset && collectedBytes <= maxDecompressedBytes) {
```
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcher.java:
##########
@@ -272,7 +294,13 @@ private long collectUncompressed(RecordBatch batch, long readFrom) {
if (record.offset() < readFrom) continue;
if (record.offset() > endOffset) return nextOffset;
recordMap.put(record.offset(), record);
+ collectedBytes += record.sizeInBytes();
Review Comment:
Shouldn't we first check the `collectedBytes += record.sizeInBytes(); if (collectedBytes >= maxDecompressedBytes) return nextOffset;` prior considering the record to be put on map? What if the last record exceeds the limit?
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcher.java:
##########
@@ -239,10 +255,16 @@ private long collect(long readFrom, LogReadResult logReadResult) {
* (never moves backwards). Records below readFrom or above endOffset are ignored. A batch that would
* exceed the decompression budget ({@link #maxDecompressedBytes}) is skipped - its offsets are left
* unread - but the loop continues on to whatever batches follow.
+ *
+ * <p>Also stops before starting a new batch once {@link #collectedBytes} already reaches {@link
+ * #maxDecompressedBytes} - a single read can return many batches (up to {@code maxFetchBytes}
+ * worth), so this check can't wait for {@link #runFrom}'s between-reads check alone, or a read
+ * containing several batches could collect well past budget before that check ever runs again.
*/
private long collectRecords(Records records, long readFrom) {
long nextOffset = readFrom;
for (RecordBatch batch : records.batches()) {
+ if (collectedBytes >= maxDecompressedBytes) return nextOffset;
Review Comment:
Change depends on inclusive/exclusive comment above.
--
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
Re: [PR] KAFKA-20723: Resolve DLQ copy records per produce round. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22919:
URL: https://github.com/apache/kafka/pull/22919#discussion_r3645062361
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcher.java:
##########
@@ -151,10 +163,14 @@ public CompletableFuture<Map<Long, Record>> fetch() {
* complete (local data, or remote data already resolved) the loop continues in place; when it is still
* pending (remote read in flight) the loop returns and is resumed from the callback - so the synchronous
* path never recurses and the async path resumes on a fresh stack (the remote storage reader thread).
+ *
+ * <p>Also stops once {@link #collectedBytes} reaches {@link #maxDecompressedBytes} - there's no need to
+ * keep reading once enough usable content has been collected for the caller's purposes (see the
+ * class-level doc), even if {@link #endOffset} hasn't been reached yet.
*/
private void runFrom(long startFrom) {
long nextOffset = startFrom;
- while (nextOffset <= endOffset) {
+ while (nextOffset <= endOffset && collectedBytes < maxDecompressedBytes) {
Review Comment:
yes, this is deliberate since we want too loop until collectedbytes are < maxAllowed - if we do <=, then we will and equality happens, we will waste one round of LogReader.readAsync data. However, with <, we will probably be able to use some of the data which we read in readAsync.
--
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
Re: [PR] KAFKA-20723: Resolve DLQ copy records per produce round. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22919:
URL: https://github.com/apache/kafka/pull/22919#discussion_r3645062361
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcher.java:
##########
@@ -151,10 +163,14 @@ public CompletableFuture<Map<Long, Record>> fetch() {
* complete (local data, or remote data already resolved) the loop continues in place; when it is still
* pending (remote read in flight) the loop returns and is resumed from the callback - so the synchronous
* path never recurses and the async path resumes on a fresh stack (the remote storage reader thread).
+ *
+ * <p>Also stops once {@link #collectedBytes} reaches {@link #maxDecompressedBytes} - there's no need to
+ * keep reading once enough usable content has been collected for the caller's purposes (see the
+ * class-level doc), even if {@link #endOffset} hasn't been reached yet.
*/
private void runFrom(long startFrom) {
long nextOffset = startFrom;
- while (nextOffset <= endOffset) {
+ while (nextOffset <= endOffset && collectedBytes < maxDecompressedBytes) {
Review Comment:
yes, this is deliberate since we want too loop until collectedbytes are < maxAllowed - if we do <=, and equality happens, we will waste one round of LogReader.readAsync data. However, with <, we will probably be able to use some of the data which we read in readAsync.
--
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
Re: [PR] KAFKA-20723: Resolve DLQ copy records per produce round. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22919:
URL: https://github.com/apache/kafka/pull/22919#discussion_r3645069041
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcher.java:
##########
@@ -239,10 +255,16 @@ private long collect(long readFrom, LogReadResult logReadResult) {
* (never moves backwards). Records below readFrom or above endOffset are ignored. A batch that would
* exceed the decompression budget ({@link #maxDecompressedBytes}) is skipped - its offsets are left
* unread - but the loop continues on to whatever batches follow.
+ *
+ * <p>Also stops before starting a new batch once {@link #collectedBytes} already reaches {@link
+ * #maxDecompressedBytes} - a single read can return many batches (up to {@code maxFetchBytes}
+ * worth), so this check can't wait for {@link #runFrom}'s between-reads check alone, or a read
+ * containing several batches could collect well past budget before that check ever runs again.
*/
private long collectRecords(Records records, long readFrom) {
long nextOffset = readFrom;
for (RecordBatch batch : records.batches()) {
+ if (collectedBytes >= maxDecompressedBytes) return nextOffset;
Review Comment:
same, here as dependant on above
--
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