This is an automated email from the ASF dual-hosted git repository.
AndrewJSchofield pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new ae9cf0f2692 KAFKA-20723: Resolve DLQ copy records per produce round.
(#22919)
ae9cf0f2692 is described below
commit ae9cf0f26925a4ee3f2aebedd71845f7d77a1d62
Author: Sushant Mahajan <[email protected]>
AuthorDate: Mon Jul 27 13:18:36 2026 +0530
KAFKA-20723: Resolve DLQ copy records per produce round. (#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.
Co-Authored-By: Claude Sonnet 5 <[email protected]>
Reviewers: Andrew Schofield <[email protected]>, Apoorv Mittal
<[email protected]>
---------
Co-authored-by: Claude Sonnet 5 <[email protected]>
---
.../share/dlq/ShareGroupDLQRecordFetcher.java | 240 +++++++++----
.../share/dlq/ShareGroupDLQStateManager.java | 157 +++++++--
.../share/dlq/ShareGroupDLQRecordFetcherTest.java | 244 ++++++++++++-
.../share/dlq/ShareGroupDLQStateManagerTest.java | 384 ++++++++++++++++++++-
4 files changed, 911 insertions(+), 114 deletions(-)
diff --git
a/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcher.java
b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcher.java
index 588077ade02..21acd854d10 100644
---
a/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcher.java
+++
b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcher.java
@@ -58,18 +58,29 @@ import java.util.concurrent.CompletableFuture;
* pending (a remote read in flight) the loop returns and is resumed from the
callback - so the calling
* thread is never blocked on remote storage IO and the synchronous path never
recurses.
*
- * <p>Best-effort: the returned future always completes normally with whatever
records could be read.
- * Offsets that cannot be read - locally or remotely - are simply absent from
the map, leaving the caller
- * to produce a DLQ record with headers only for them.
+ * <p>Best-effort: the returned future always completes normally with a {@link
FetchResult} holding whatever
+ * records could be read, plus the offset through which a definitive outcome
was reached this fetch. Offsets
+ * within that boundary that have no entry in {@link FetchResult#records()}
were either permanently excluded
+ * because they (or their containing batch) can never fit within {@code
maxDecompressedBytes}, or belong to
+ * a range given up on entirely after a genuine failure (a read error, or an
unexpected exception) - either
+ * way, the caller should produce a DLQ record with headers only for them.
Offsets beyond the boundary were
+ * never examined because this fetch ran out of budget before reaching them,
and are eligible for a fresh
+ * fetch with its own full budget.
*
* <p>{@code maxFetchBytes} only bounds the compressed, on-the-wire size of
each read; a compressed batch
* can still decompress into something far larger in memory. To keep a
pathologically compressible batch
- * from ballooning broker heap usage, decompression is bounded by {@code
maxDecompressedBytes} (shared across
- * the whole fetch): compressed batches are decompressed into a size-capped
flat buffer before any record is
- * parsed (see {@link #decompressBounded}), so even a single record with a
fabricated huge length can't force
- * a large allocation. A batch that would exceed the remaining budget is
simply discarded - its offsets are
- * skipped, same as any other partial-failure case - and the fetch continues
with whatever batches follow,
- * rather than one oversized batch poisoning the rest of the range.
+ * from ballooning broker heap usage, each batch is decompressed into a
size-capped flat buffer before any
+ * record is parsed (see {@link #decompressBounded}), bounded by {@code
maxDecompressedBytes} - so even a
+ * single record with a fabricated huge length can't force a large allocation.
A batch whose decompressed
+ * size exceeds {@code maxDecompressedBytes} on its own can never fit in any
fetch, so it is permanently
+ * excluded rather than retried.
+ *
+ * <p>{@code maxDecompressedBytes} also bounds the total amount of content
this fetch collects: once the
+ * records actually collected (compressed or not) reach that many bytes, or a
record is found that doesn't
+ * fit alongside what's already been collected, the fetch stops rather than
continuing to walk the rest of
+ * the requested range - reading further would just be discarded work, since
callers only ever need one
+ * message's worth of records at a time (see {@code
ShareGroupDLQStateManager}, which resolves one produce
+ * round at a time and passes its {@code dlqTopicMaxMessageBytes} as this
budget).
*
* <p>Instances are single-use: create one fetcher per {@link #fetch()} call.
*/
@@ -94,13 +105,27 @@ public class ShareGroupDLQRecordFetcher {
private final Map<Long, Record> recordMap;
private final long maxDecompressedBytes;
private final BufferSupplier bufferSupplier = BufferSupplier.create();
- private long decompressedBytes = 0;
+ // Cumulative sizeInBytes() of every record actually collected into
recordMap so far, regardless of
+ // whether it came from a compressed or uncompressed batch. Compared
against maxDecompressedBytes to
+ // decide whether the next record still fits alongside what's already been
collected this fetch.
+ private long collectedBytes = 0;
+ // The largest offset such that every offset from param.firstOffset()
through it has a definitive,
+ // in-scope outcome this fetch: collected into recordMap, or permanently
excluded because it (or its
+ // containing batch) can never fit within maxDecompressedBytes. Never
advances past a record or batch
+ // that was merely deferred for not fitting alongside collectedBytes -
those, and everything after them,
+ // are left untouched for a fresh fetch (with its own full budget) to
attempt. Starts one before the
+ // requested range so a fetch that resolves nothing still reports that
correctly.
+ private long lastResolvedOffset;
+ // Set once a record is found that doesn't fit alongside collectedBytes
but isn't itself permanently
+ // excluded - signals every loop in this class to stop immediately rather
than trying further batches or
+ // reads, since that record and everything after it must be left untouched
for a fresh fetch to attempt.
+ private boolean deferredRemainder = false;
// We are fetching data for one TopicIdPartition only. Hence, there is no
need to keep recreating
// the maxBytes map, and we can re-use a single copy. In similar vein, we
needn't clear the offsets
// map either and just update the value corresponding to the
TopicIdPartition key across iterations.
private final LinkedHashMap<TopicIdPartition, Long> offsets = new
LinkedHashMap<>();
private final LinkedHashMap<TopicIdPartition, Integer> maxBytesMap = new
LinkedHashMap<>();
- private final CompletableFuture<Map<Long, Record>> result = new
CompletableFuture<>();
+ private final CompletableFuture<FetchResult> result = new
CompletableFuture<>();
private final FetchParams fetchParams;
public ShareGroupDLQRecordFetcher(LogReader logReader, Time time,
ShareGroupDLQRecordParameter param,
@@ -115,6 +140,7 @@ public class ShareGroupDLQRecordFetcher {
this.recordMap = new HashMap<>(recordCount);
this.maxBytesMap.put(tp, maxFetchBytes);
this.maxDecompressedBytes = maxDecompressedBytes;
+ this.lastResolvedOffset = param.firstOffset() - 1;
this.fetchParams = new FetchParams(
FetchRequest.CONSUMER_REPLICA_ID, // -1, reading as a
consumer
-1, // replicaEpoch
@@ -129,19 +155,20 @@ public class ShareGroupDLQRecordFetcher {
/**
* Fetches the source records for the configured offset range.
*
- * @return A future that always completes normally with the records that
could be read, keyed by offset.
+ * @return A future that always completes normally with the records that
could be read, plus the
+ * boundary through which a definitive outcome was reached - see
{@link FetchResult}.
*/
- public CompletableFuture<Map<Long, Record>> fetch() {
+ public CompletableFuture<FetchResult> fetch() {
try {
runFrom(param.firstOffset());
} catch (Throwable e) {
// Never let an unexpected error - including an OutOfMemoryError
from a maliciously
// compressible record, or an
InvalidRecordException/KafkaException from a rejected
- // malformed one - escape. Uses complete() (not
result.complete(Map.of())) so that any
- // records already collected from earlier batches in this call are
still returned, rather
- // than discarding a partially-successful copy because of one bad
batch.
+ // malformed one - escape. Uses completeGivingUpOnRemainder() (not
result.complete(...))
+ // so that any records already collected from earlier batches in
this call are still
+ // returned, rather than discarding a partially-successful copy
because of one bad batch.
log.warn("Unexpected error fetching records for {}. Returning
records fetched so far.", param, e);
- complete();
+ completeGivingUpOnRemainder();
}
return result;
}
@@ -151,10 +178,14 @@ public class ShareGroupDLQRecordFetcher {
* 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 #shouldStop()} is true - there's no need to
keep reading once enough usable
+ * content has been collected for the caller's purposes, or a record has
been found that doesn't fit
+ * alongside it, even if {@link #endOffset} hasn't been reached yet.
*/
private void runFrom(long startFrom) {
long nextOffset = startFrom;
- while (nextOffset <= endOffset) {
+ while (nextOffset <= endOffset && !shouldStop()) {
offsets.put(tp, nextOffset);
CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
future =
@@ -175,7 +206,7 @@ public class ShareGroupDLQRecordFetcher {
// completes normally; any unexpected exceptional completion is
caught by fetch().
long advanced = collect(nextOffset,
logReadResult(future.getNow(null)));
if (advanced <= nextOffset) {
- complete(); // no progress - stop
+ completeAfterNoProgress(); // no progress - stop
return;
}
nextOffset = advanced;
@@ -183,6 +214,32 @@ public class ShareGroupDLQRecordFetcher {
complete();
}
+ /**
+ * True once enough content has been collected for the caller's purposes,
or a record has been found
+ * that doesn't fit alongside {@link #collectedBytes} - either way, every
loop in this class should stop
+ * immediately rather than attempting further batches or reads.
+ */
+ private boolean shouldStop() {
+ return deferredRemainder || collectedBytes >= maxDecompressedBytes;
+ }
+
+ /**
+ * Completes after a read makes no progress. This happens for two very
different reasons that need
+ * different handling: {@link #deferredRemainder} being set means a record
was found that doesn't fit
+ * alongside {@link #collectedBytes} - a budget-related stop worth
retrying fresh, so the boundary is
+ * preserved as-is via {@link #complete()}. Otherwise, nothing advanced
because the read itself found
+ * nothing new (an error, an empty result, or a batch containing only
already-seen offsets) - a genuine
+ * dead end that a retry with a fresh budget wouldn't fix, so {@link
#completeGivingUpOnRemainder()}
+ * gives up on the rest of the range instead of leaving it to be retried
forever.
+ */
+ private void completeAfterNoProgress() {
+ if (deferredRemainder) {
+ complete();
+ } else {
+ completeGivingUpOnRemainder();
+ }
+ }
+
/**
* Extracts the read result for the partition being fetched, or {@code
null} when none was produced
* (e.g. the read returned no entry for the partition).
@@ -199,21 +256,21 @@ public class ShareGroupDLQRecordFetcher {
try {
if (exception != null) {
log.warn("Unable to read records at offset {} for {}. Skipping
it.", readFrom, param, exception);
- complete();
+ completeGivingUpOnRemainder();
return;
}
long advanced = collect(readFrom, logReadResult);
if (advanced <= readFrom) {
- complete(); // no progress - stop
+ completeAfterNoProgress(); // no progress - stop
} else {
- runFrom(advanced); // resume the loop
+ runFrom(advanced); // resume the loop
}
} catch (Throwable e) {
// Never let an unexpected error - including an OutOfMemoryError
from a maliciously
// compressible record, or an
InvalidRecordException/KafkaException from a rejected
// malformed one - escape; return whatever was collected so far.
log.warn("Unexpected error processing records for {}. Returning
records fetched so far.", param, e);
- complete();
+ completeGivingUpOnRemainder();
}
}
@@ -236,13 +293,17 @@ public class ShareGroupDLQRecordFetcher {
/**
* Adds the records within the requested range to the map and returns the
offset to read from next
- * (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.
+ * (never moves backwards). Records below readFrom or above endOffset are
ignored.
+ *
+ * <p>Also stops before starting a new batch once {@link #shouldStop()} is
true - 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 keep being
+ * processed past the point where the fetch should have stopped.
*/
private long collectRecords(Records records, long readFrom) {
long nextOffset = readFrom;
for (RecordBatch batch : records.batches()) {
+ if (shouldStop()) return nextOffset;
nextOffset = collectFromBatch(batch, nextOffset);
}
return nextOffset;
@@ -271,8 +332,24 @@ public class ShareGroupDLQRecordFetcher {
// nextOffset backwards.
if (record.offset() < readFrom) continue;
if (record.offset() > endOffset) return nextOffset;
+ long recordSize = record.sizeInBytes();
+ if (recordSize > maxDecompressedBytes) {
+ // This one record can never fit within maxDecompressedBytes,
in this fetch or any other -
+ // exclude it permanently rather than leaving it for a retry
that could never succeed.
+ nextOffset = Math.max(nextOffset, record.offset() + 1);
+ lastResolvedOffset = Math.max(lastResolvedOffset,
record.offset());
+ continue;
+ }
+ if (collectedBytes + recordSize > maxDecompressedBytes) {
+ // Fits on its own, but not alongside what's already been
collected this fetch - stop here
+ // and leave this record, and everything after it, untouched
for a fresh fetch to attempt.
+ deferredRemainder = true;
+ return nextOffset;
+ }
recordMap.put(record.offset(), record);
+ collectedBytes += recordSize;
nextOffset = Math.max(nextOffset, record.offset() + 1); // never
moves backwards
+ lastResolvedOffset = Math.max(lastResolvedOffset, record.offset());
}
return nextOffset;
}
@@ -306,10 +383,11 @@ public class ShareGroupDLQRecordFetcher {
private long collectFromCompressedBatch(DefaultRecordBatch batch, long
readFrom) {
ByteBuffer decompressed = decompressBounded(batch);
if (decompressed == null) {
- // This batch alone would exceed the remaining budget: skip past
it (its offsets are left
- // unread) rather than aborting the whole fetch, so later batches
- which may well fit - are
- // still read.
- return Math.max(readFrom, batch.lastOffset() + 1);
+ // The batch's decompressed size exceeds maxDecompressedBytes on
its own, so it can never fit -
+ // in this fetch or any other. Exclude it permanently and move on
to whatever batches follow.
+ long lastOffsetInRange = Math.min(batch.lastOffset(), endOffset);
+ lastResolvedOffset = Math.max(lastResolvedOffset,
lastOffsetInRange);
+ return Math.max(readFrom, lastOffsetInRange + 1);
}
long baseOffset = batch.baseOffset();
@@ -325,26 +403,38 @@ public class ShareGroupDLQRecordFetcher {
// nextOffset backwards.
if (record.offset() < readFrom) continue;
if (record.offset() > endOffset) return nextOffset;
+ long recordSize = record.sizeInBytes();
+ if (collectedBytes + recordSize > maxDecompressedBytes) {
+ // The whole batch already fit within maxDecompressedBytes on
its own (decompression would
+ // have failed above otherwise), so this record only fails to
fit alongside what's already
+ // been collected - stop here and leave it, and everything
after it, untouched for a fresh
+ // fetch to attempt.
+ deferredRemainder = true;
+ return nextOffset;
+ }
recordMap.put(record.offset(), record);
+ collectedBytes += recordSize;
nextOffset = Math.max(nextOffset, record.offset() + 1); // never
moves backwards
+ lastResolvedOffset = Math.max(lastResolvedOffset, record.offset());
}
return nextOffset;
}
/**
* Decompresses the batch's record region into a flat buffer, reading in
{@link #DECOMPRESS_CHUNK_BYTES}
- * chunks and checking the cumulative total against the remaining
decompression budget before each chunk
- * is kept - before any record-level parsing happens. A single record with
a fabricated huge length can't
- * cause a large allocation this way: parsing only begins once the buffer
is already fully bounded, and the
- * buffer-based reader rejects a record whose declared length doesn't fit
what's left of it.
+ * chunks and checking the cumulative total against {@link
#maxDecompressedBytes} before each chunk is
+ * kept - before any record-level parsing happens. A single record with a
fabricated huge length can't
+ * cause a large allocation this way: parsing only begins once the buffer
is already fully bounded, and
+ * the buffer-based reader rejects a record whose declared length doesn't
fit what's left of it.
+ *
+ * <p>Bounded by the full {@code maxDecompressedBytes} regardless of
{@link #collectedBytes}: whether a
+ * batch fits is a property of the batch alone, not of how much of the
budget earlier batches in this
+ * fetch happened to use, so every batch gets the same, full-sized chance
to decompress.
*
- * @return the bounded, flipped buffer ready for reading, or {@code null}
if the remaining decompression
- * budget ({@link #maxDecompressedBytes}, shared across the whole
fetch) is already exhausted.
+ * @return the bounded, flipped buffer ready for reading, or {@code null}
if the batch's decompressed
+ * size exceeds {@link #maxDecompressedBytes} on its own.
*/
private ByteBuffer decompressBounded(DefaultRecordBatch batch) {
- long budget = maxDecompressedBytes - decompressedBytes;
- if (budget <= 0) return null;
-
int chunkBytes = Math.min(batch.sizeInBytes(), DECOMPRESS_CHUNK_BYTES);
try (InputStream in = batch.recordInputStream(bufferSupplier);
ByteBufferOutputStream out = new
ByteBufferOutputStream(chunkBytes)) {
@@ -353,14 +443,13 @@ public class ShareGroupDLQRecordFetcher {
long total = 0;
while ((nRead = in.read(chunk, 0, chunk.length)) != -1) {
total += nRead;
- if (total > budget) {
- log.warn("Decompressed batch data for {} exceeded the {}
byte budget. " +
- "Stopping record copy early to bound memory use.",
param, maxDecompressedBytes);
+ if (total > maxDecompressedBytes) {
+ log.warn("Decompressed batch data for {} exceeded the {}
byte budget on its own. " +
+ "The batch can never fit and will be permanently
skipped.", param, maxDecompressedBytes);
return null;
}
out.write(chunk, 0, nRead);
}
- decompressedBytes += total;
out.buffer().flip();
return out.buffer();
} catch (IOException e) {
@@ -371,14 +460,9 @@ public class ShareGroupDLQRecordFetcher {
/**
* Fallback used only for legacy magic v0/v1 compressed batches (no
bounded-buffer path available for
* that format - see {@link #asDefaultRecordBatch}): iterates records one
at a time via {@link
- * RecordBatch#streamingIterator}, tracking a cumulative
decompressed-bytes total across the whole fetch
- * and stopping once {@link #maxDecompressedBytes} is exceeded. Unlike
{@link
- * #collectFromCompressedBatch}, this does not prevent a single oversized
record's initial allocation -
- * it only bounds growth across multiple records/batches. Because that
allocation already happened (the
- * byte count reflects real, already-materialized memory rather than data
that was rejected before use),
- * the overshoot is not undone: {@link #decompressedBytes} stays over
budget, so later batches on this
- * path keep getting skipped too. The rest of the current batch is skipped
- not the whole fetch - so
- * later batches (e.g. uncompressed ones, which carry no such risk) are
still read.
+ * RecordBatch#streamingIterator}. Unlike {@link
#collectFromCompressedBatch}, this does not prevent a
+ * single oversized record's initial allocation before its size is known -
it only decides, once a
+ * record has already been decompressed, whether to keep it or discard it.
*/
private long collectWithCumulativeCap(RecordBatch batch, long readFrom) {
long nextOffset = readFrom;
@@ -388,24 +472,34 @@ public class ShareGroupDLQRecordFetcher {
if (record.offset() < readFrom) continue;
if (record.offset() > endOffset) return nextOffset;
- decompressedBytes += record.sizeInBytes();
- if (decompressedBytes > maxDecompressedBytes) {
- log.warn("Decompressed record data for {} exceeded {}
bytes at offset {}. " +
- "Skipping the rest of this batch to bound memory use.",
- param, maxDecompressedBytes, record.offset());
- return Math.max(nextOffset, batch.lastOffset() + 1);
+ long recordSize = record.sizeInBytes();
+ if (recordSize > maxDecompressedBytes) {
+ // This one record can never fit within
maxDecompressedBytes, in this fetch or any
+ // other - exclude it permanently rather than leaving it
for a retry that could never
+ // succeed.
+ nextOffset = Math.max(nextOffset, record.offset() + 1);
+ lastResolvedOffset = Math.max(lastResolvedOffset,
record.offset());
+ continue;
+ }
+ if (collectedBytes + recordSize > maxDecompressedBytes) {
+ // Fits on its own, but not alongside what's already been
collected this fetch - stop
+ // here and leave it, and everything after it, untouched
for a fresh fetch to attempt.
+ deferredRemainder = true;
+ return nextOffset;
}
recordMap.put(record.offset(), record);
+ collectedBytes += recordSize;
nextOffset = Math.max(nextOffset, record.offset() + 1);
+ lastResolvedOffset = Math.max(lastResolvedOffset,
record.offset());
}
}
return nextOffset;
}
/**
- * Completes the result future with an immutable snapshot of the records
collected so far. Offsets
- * that could not be read are absent from the map; the caller produces a
headers-only DLQ record for them.
+ * Completes the result future with an immutable snapshot of the records
collected so far, plus the
+ * boundary through which a definitive outcome was reached this fetch.
*/
private void complete() {
bufferSupplier.close();
@@ -414,6 +508,32 @@ public class ShareGroupDLQRecordFetcher {
if (recordCount != recordMap.size()) {
log.info("Total offsets requested: {}, Records found: {}",
recordCount, recordMap.size());
}
- result.complete(Map.copyOf(recordMap));
+ result.complete(new FetchResult(Map.copyOf(recordMap),
lastResolvedOffset));
+ }
+
+ /**
+ * Like {@link #complete()}, but first treats the entire remaining range -
from wherever this fetch got
+ * to, through {@link #endOffset} - as settled. Used when the reason for
stopping is a genuine failure
+ * (a read error, an unexpected exception, or a read that made no
progress) rather than running out of
+ * budget: unlike a budget-related stop, retrying a failure like this with
a fresh budget wouldn't help,
+ * so there's nothing to gain by leaving the remainder for a later fetch
to retry - the caller should
+ * treat it as permanently unavailable and produce headers only for it,
the same as any offset excluded
+ * for not fitting the decompression budget.
+ */
+ private void completeGivingUpOnRemainder() {
+ lastResolvedOffset = Math.max(lastResolvedOffset, endOffset);
+ complete();
+ }
+
+ /**
+ * @param records the source records that could be read, keyed by offset
+ * @param lastResolvedOffset the largest offset such that every offset
from the requested range's start
+ * through it has a definitive, in-scope outcome
- collected into
+ * {@code records}, or permanently excluded
because it (or its containing
+ * batch) can never fit within the configured
decompression budget. Offsets
+ * beyond it were never examined and should be
retried, with a fresh budget,
+ * by a later fetch.
+ */
+ public record FetchResult(Map<Long, Record> records, long
lastResolvedOffset) {
}
}
diff --git
a/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
index 51e674a514f..0661adca3e8 100644
---
a/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
+++
b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
@@ -198,14 +198,14 @@ public class ShareGroupDLQStateManager {
return future;
}
- // Resolve the source records once, here - on the calling thread for
local offsets and, for
+ // Resolve round 1's source records here - on the calling thread for
local offsets and, for
// tiered offsets, asynchronously on the remote-storage reader pool -
and enqueue only once
- // resolution finishes. This keeps both the local and remote reads off
the single sender
- // thread, and the memoized result is reused on every (re)send so
retries never re-fetch.
- // Records are only read when copy is enabled for the group and the
DLQ is correctly
+ // resolution finishes. Later rounds (if the range needs more than one
produce request) are
+ // resolved the same way from handleProduceResponse(), each with its
own fresh decompression
+ // budget. Records are only read when copy is enabled for the group
and the DLQ is correctly
// configured (validated above); otherwise we enqueue immediately.
if (cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
- requestHandler.resolveRecords().whenComplete((ignored,
ignoredError) -> enqueue(requestHandler));
+ requestHandler.resolveRound().whenComplete((ignored, ignoredError)
-> enqueue(requestHandler));
} else {
enqueue(requestHandler);
}
@@ -246,6 +246,26 @@ public class ShareGroupDLQStateManager {
sender.wakeup();
}
+ /**
+ * Invokes {@code handler.onComplete(response)}, isolating this one
handler's failure from
+ * whatever else is being processed in the same batch (see the
coalesced-response callback in
+ * {@link SendThread#generateRequests}, which invokes this once per
handler in a shared
+ * produce response). An uncaught exception here must not prevent the rest
of that batch's
+ * handlers from being notified - which would otherwise leave their {@link
#dlq} futures
+ * hanging forever - so it's caught, logged, and the offending handler's
own future is
+ * explicitly failed too, in case the exception happened before {@code
onComplete} reached its
+ * own completion call.
+ */
+ // Visibility for tests
+ static void completeHandlerSafely(ProduceRequestHandler handler,
ClientResponse response) {
+ try {
+ handler.onComplete(response);
+ } catch (Exception e) {
+ log.error("Uncaught error handling produce response for handler
{}.", handler, e);
+ handler.requestErrorResponse(e);
+ }
+ }
+
// Visibility for tests
class ProduceRequestHandler implements RequestCompletionHandler {
private final CompletableFuture<Void> result;
@@ -259,12 +279,22 @@ public class ShareGroupDLQStateManager {
private volatile Node dlqPartitionLeaderNode;
private volatile int dlqDestinationPartition;
private volatile ShareGroupDLQMetadataCacheHelper.TopicPartitionData
dlqTopicPartitionData;
- // The original source records, resolved once before this handler is
enqueued (see resolveRecords()).
- // Volatile because resolution runs off the sender thread - on the
calling thread for local offsets
- // and, for tiered offsets, on the remote-storage reader pool - while
this value is read on the
- // sender thread when the produce request is built. Memoized: set once
and reused for every (re)send,
- // so retries never re-fetch.
+ // The original source records for the CURRENT round only, resolved
before this round is added
+ // to the node map (see resolveRound()): once for round 1 (before this
handler is first
+ // enqueued), then again for each subsequent round after a successful
produce response advances
+ // nextOffsetToSend. Volatile because resolution runs off the sender
thread - on the calling
+ // thread for local offsets and, for tiered offsets, on the
remote-storage reader pool - while
+ // this value is read on the sender thread when the produce request is
built. Memoized per round:
+ // set once per round and reused for every retry of that round, so
retries never re-fetch.
private volatile Map<Long, Record> resolvedRecordData = Map.of();
+ // The largest offset such that everything from this round's start
through it has a definitive
+ // outcome (real content, or a permanently-excluded gap) per the fetch
behind resolvedRecordData -
+ // see ShareGroupDLQRecordFetcher.FetchResult. Caps
topicProduceData()'s walk so it never packs in
+ // offsets the fetch never got to attempt this round as headers-only;
those stay untouched for a
+ // fresh round (with a fresh decompression budget) to retry. Defaults
to param.lastOffset() so a
+ // handler that never calls resolveRound() (copy-record disabled)
still leaves the full range fair
+ // game for topicProduceData()'s own size-based packing.
+ private volatile long lastResolvedOffsetThisRound;
// The next offset that has not yet been included in a produce
request. Starts at
// param.firstOffset() and advances past whatever topicProduceData()
managed to fit within
// dlqTopicMaxMessageBytes() on each successful send, so a range that
doesn't fit in a single
@@ -295,6 +325,7 @@ public class ShareGroupDLQStateManager {
this.result = result;
this.nextOffsetToSend = param.firstOffset();
this.lastOffsetIncludedThisRound = param.firstOffset() - 1;
+ this.lastResolvedOffsetThisRound = param.lastOffset();
this.createTopicsBackoff = new ExponentialBackoffManager(
maxRPCRetryAttempts,
backoffMs,
@@ -411,7 +442,16 @@ public class ShareGroupDLQStateManager {
List<SimpleRecord> simpleRecords = new ArrayList<>();
int batchSize = DefaultRecordBatch.RECORD_BATCH_OVERHEAD;
Long baseTimestamp = null;
- for (long offset = nextOffsetToSend; offset <= param.lastOffset();
offset++) {
+ // Capped at lastResolvedOffsetThisRound, not just
param.lastOffset(): offsets beyond it were
+ // never attempted by this round's fetch and must stay untouched
for a fresh round to retry,
+ // rather than being packed in here as headers-only just because
they have no map entry yet.
+ // Floored at nextOffsetToSend itself (mirroring the single-record
floor below for the
+ // size-exceeds-limit case): a fetch that resolved nothing at all
for this round - e.g. a read
+ // that failed outright - would otherwise leave this loop with
zero iterations, producing an
+ // empty record batch, which the broker rejects outright. Sending
nextOffsetToSend alone,
+ // headers-only, guarantees forward progress even when nothing
could be resolved.
+ long roundEnd = Math.max(nextOffsetToSend,
Math.min(param.lastOffset(), lastResolvedOffsetThisRound));
+ for (long offset = nextOffsetToSend; offset <= roundEnd; offset++)
{
// Must be wall-clock (epoch) time: log retention decides
whether to delete this
// record's segment by comparing its timestamp against the
current wall-clock time.
long timestamp = time.milliseconds();
@@ -520,7 +560,11 @@ public class ShareGroupDLQStateManager {
} catch (ConfigException e) {
return false;
}
- // Source records were already resolved before enqueue; just
add to the node map.
+ // This path handles both round 1 (dlq() already resolved its
records before enqueue)
+ // and a retry of whichever round is currently in flight
(nextOffsetToSend is untouched
+ // by a failed produce, so that round's records - resolved
when it was first entered -
+ // are still valid); either way, this round's data is already
resolved, so just add to
+ // the node map without fetching again.
addRequestToNodeMap(dlqPartitionLeaderNode, this);
}
return isDlqTopicPresent;
@@ -615,7 +659,9 @@ public class ShareGroupDLQStateManager {
try {
populateDLQTopicData();
createTopicsBackoff.resetAttempts();
- // Source records were already resolved before
enqueue; just add to the node map.
+ // This path is only ever reached for round 1
(a brand-new handler still
+ // waiting on topic creation), whose records
dlq() already resolved before
+ // enqueue; just add to the node map.
addRequestToNodeMap(this.dlqPartitionLeaderNode, this);
} catch (ConfigException e) {
LOG.error("Error enqueueing after DLQ create
topic response {}.", this, e);
@@ -715,8 +761,11 @@ public class ShareGroupDLQStateManager {
if (lastOffsetIncludedThisRound <
param.lastOffset()) {
// Only part of the offset range fit within
dlqTopicMaxMessageBytes - continue
// sending the remainder as a follow-up
produce request instead of completing.
+ // Resolve the next round's source records (a
fresh decompression budget, scoped
+ // to what's left to send) before rejoining
the node map, mirroring the
+ // "resolved before eligible for coalescing"
invariant dlq() establishes for round 1.
nextOffsetToSend = lastOffsetIncludedThisRound
+ 1;
- addRequestToNodeMap(dlqPartitionLeaderNode(),
this);
+ resolveRound().whenComplete((ignored, err) ->
addRequestToNodeMap(dlqPartitionLeaderNode(), this));
} else {
this.result.complete(null);
}
@@ -765,34 +814,61 @@ public class ShareGroupDLQStateManager {
}
/**
- * Resolves the original source records for this handler once, before
it is enqueued - reading
- * from the local log on the calling thread and, for any offsets
tiered to remote storage,
- * asynchronously on the remote-storage reader pool. The result is
memoized in
- * {@link #resolvedRecordData} and reused for every (re)send, so the
single sender thread never
- * reads the log (neither local nor remote) and retries do not
re-fetch.
+ * Resolves the source records for the CURRENT round only - the window
starting at
+ * {@link #nextOffsetToSend} - reading from the local log on the
calling thread and, for any
+ * offsets tiered to remote storage, asynchronously on the
remote-storage reader pool. The result
+ * is memoized in {@link #resolvedRecordData} and reused for every
retry of this same round (this
+ * method is only ever called once per round - see {@link #dlq} for
round 1 and
+ * {@link #handleProduceResponse} for subsequent rounds - so a round's
data is never re-fetched by
+ * a retry of that round).
*
- * <p>A failed fetch is non-fatal: {@link #resolvedRecordData} stays
empty and the DLQ record is
- * produced with headers only (no key/value), mirroring how
individually unavailable offsets are skipped.
+ * <p>A failed fetch is non-fatal: {@link #resolvedRecordData} stays
empty, {@link
+ * #lastResolvedOffsetThisRound} is set to {@code param.lastOffset()}
so {@code topicProduceData()}
+ * treats the whole remaining range as settled, and the DLQ record is
produced with headers only
+ * (no key/value) for it - mirroring how individually unavailable
offsets are skipped. Unlike a
+ * fetch that partially succeeds (see {@link
ShareGroupDLQRecordFetcher.FetchResult}), a total
+ * failure here isn't a decompression-budget gap a fresh round could
resolve, so there's no reason
+ * to hold any of the range back for a retry. This applies equally to
an unexpected error thrown
+ * synchronously by {@link #maybeFetchRecordData} itself (e.g. a
cache-helper lookup) - not just
+ * one carried by the returned future's exceptional completion - since
this method is called
+ * directly from {@link #handleProduceResponse} (on the sender thread,
inside a
+ * {@code RequestCompletionHandler} callback) for round 2 onward:
letting an exception escape
+ * from here would propagate out of {@code onComplete()}, which -
depending on which internal
+ * path delivers the response - is not guaranteed to be caught before
reaching the broker's
+ * fatal-error handling.
*
- * @return A future that always completes normally, once resolution
has finished.
+ * @return A future that always completes normally, once this round's
resolution has finished.
*/
- CompletableFuture<Void> resolveRecords() {
+ CompletableFuture<Void> resolveRound() {
+ long roundStart = nextOffsetToSend;
CompletableFuture<Void> resolved = new CompletableFuture<>();
- maybeFetchRecordData().whenComplete((records, exception) -> {
- if (exception != null || records == null) {
- LOG.warn("Unable to fetch original record data for handler
{}. DLQ records will be produced with headers only.", this, exception);
- this.resolvedRecordData = Map.of();
- } else {
- this.resolvedRecordData = records;
- }
+ try {
+ maybeFetchRecordData(roundStart).whenComplete((fetchResult,
exception) -> {
+ if (exception != null || fetchResult == null) {
+ LOG.warn("Unable to fetch original record data for
handler {} for the round starting at offset {}. " +
+ "DLQ records will be produced with headers only
for this round.", this, roundStart, exception);
+ this.resolvedRecordData = Map.of();
+ this.lastResolvedOffsetThisRound = param.lastOffset();
+ } else {
+ this.resolvedRecordData = fetchResult.records();
+ this.lastResolvedOffsetThisRound =
fetchResult.lastResolvedOffset();
+ }
+ resolved.complete(null);
+ });
+ } catch (Throwable t) {
+ LOG.warn("Unexpected error resolving round starting at offset
{} for {}. " +
+ "DLQ records will be produced with headers only for this
round.", roundStart, this, t);
+ this.resolvedRecordData = Map.of();
+ this.lastResolvedOffsetThisRound = param.lastOffset();
resolved.complete(null);
- });
+ }
return resolved;
}
- private CompletableFuture<Map<Long, Record>> maybeFetchRecordData() {
+ private CompletableFuture<ShareGroupDLQRecordFetcher.FetchResult>
maybeFetchRecordData(long fromOffset) {
if
(!cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
- return CompletableFuture.completedFuture(Map.of());
+ return CompletableFuture.completedFuture(
+ new ShareGroupDLQRecordFetcher.FetchResult(Map.of(),
param.lastOffset()));
}
// Bounds decompression memory against a pathologically
compressible source record: there's
// no point retaining more decompressed data than the DLQ topic
could ever accept anyway, and
@@ -802,7 +878,18 @@ public class ShareGroupDLQStateManager {
int maxDecompressedBytes =
cacheHelper.shareGroupDlqTopic(param.groupId())
.map(cacheHelper::dlqTopicMaxMessageBytes)
.orElse(DLQ_MAX_FETCH_BYTES);
- return new ShareGroupDLQRecordFetcher(logReader, time, param,
DLQ_MAX_FETCH_BYTES, maxDecompressedBytes).fetch();
+ // param itself is never mutated - headers()/topicProduceData()
rely on its original,
+ // unwindowed firstOffset/lastOffset for the handler's whole
lifetime. Build a throwaway
+ // windowed copy only to scope this round's fetch (and its
decompression budget) to what's
+ // left to send.
+ ShareGroupDLQRecordParameter window;
+ if (fromOffset == param.firstOffset()) {
+ window = param;
+ } else {
+ window = new ShareGroupDLQRecordParameter(param.groupId(),
param.topicIdPartition(), fromOffset,
+ param.lastOffset(), param.deliveryCount(), param.cause());
+ }
+ return new ShareGroupDLQRecordFetcher(logReader, time, window,
DLQ_MAX_FETCH_BYTES, maxDecompressedBytes).fetch();
}
}
@@ -899,7 +986,7 @@ public class ShareGroupDLQStateManager {
// now the combined request has completed
// we need to create responses for individual
// requests which composed the combined request
- results.liveHandlers().forEach(handler ->
handler.onComplete(response));
+ results.liveHandlers().forEach(handler ->
completeHandlerSafely(handler, response));
wakeup();
}));
sending.add(destNode);
diff --git
a/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcherTest.java
b/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcherTest.java
index 5dac442912f..8867d271484 100644
---
a/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcherTest.java
+++
b/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcherTest.java
@@ -37,6 +37,9 @@ import org.apache.kafka.storage.internals.log.FetchDataInfo;
import org.apache.kafka.storage.internals.log.LogReadResult;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import java.io.ByteArrayOutputStream;
import java.io.InputStream;
@@ -44,11 +47,14 @@ import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.OptionalLong;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -89,7 +95,14 @@ class ShareGroupDLQRecordFetcherTest {
}
private Map<Long, Record> fetch(ShareGroupDLQRecordParameter param) throws
Exception {
- return fetcher(param).fetch().get(10, TimeUnit.SECONDS);
+ return fetcher(param).fetch().get(10, TimeUnit.SECONDS).records();
+ }
+
+ // The exact on-wire size of a single uncompressed record, computed via
the real API rather than
+ // hardcoded, so budget-boundary tests can align a budget precisely with a
record's own size.
+ private static int recordSizeInBytes(SimpleRecord simpleRecord) {
+ MemoryRecords batch = MemoryRecords.withRecords(Compression.NONE,
simpleRecord);
+ return
batch.batches().iterator().next().iterator().next().sizeInBytes();
}
// ---- helpers ----
@@ -255,12 +268,12 @@ class ShareGroupDLQRecordFetcherTest {
CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
pending = new CompletableFuture<>();
when(logReader.readAsync(any(), anySet(), any(), any(),
anyBoolean())).thenReturn(pending);
- CompletableFuture<Map<Long, Record>> resultFuture = fetcher(param(0L,
2L)).fetch();
+ CompletableFuture<ShareGroupDLQRecordFetcher.FetchResult> resultFuture
= fetcher(param(0L, 2L)).fetch();
assertFalse(resultFuture.isDone(), "Fetch should be waiting on the
pending read");
pending.complete(resultMap(success(record("k0", "v0"), record("k1",
"v1"), record("k2", "v2"))));
- Map<Long, Record> result = resultFuture.get(10, TimeUnit.SECONDS);
+ Map<Long, Record> result = resultFuture.get(10,
TimeUnit.SECONDS).records();
assertEquals(3, result.size());
assertRecord(result, 0L, "k0", "v0");
assertRecord(result, 1L, "k1", "v1");
@@ -272,13 +285,13 @@ class ShareGroupDLQRecordFetcherTest {
CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
pending = new CompletableFuture<>();
when(logReader.readAsync(any(), anySet(), any(), any(),
anyBoolean())).thenReturn(pending);
- CompletableFuture<Map<Long, Record>> resultFuture = fetcher(param(0L,
2L)).fetch();
+ CompletableFuture<ShareGroupDLQRecordFetcher.FetchResult> resultFuture
= fetcher(param(0L, 2L)).fetch();
assertFalse(resultFuture.isDone());
// Completing with no records makes no progress, so the resumed loop
stops and completes empty.
pending.complete(resultMap(success()));
- assertTrue(resultFuture.get(10, TimeUnit.SECONDS).isEmpty());
+ assertTrue(resultFuture.get(10, TimeUnit.SECONDS).records().isEmpty());
}
@Test
@@ -286,14 +299,14 @@ class ShareGroupDLQRecordFetcherTest {
CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
pending = new CompletableFuture<>();
when(logReader.readAsync(any(), anySet(), any(), any(),
anyBoolean())).thenReturn(pending);
- CompletableFuture<Map<Long, Record>> resultFuture = fetcher(param(0L,
2L)).fetch();
+ CompletableFuture<ShareGroupDLQRecordFetcher.FetchResult> resultFuture
= fetcher(param(0L, 2L)).fetch();
// An unexpected error while processing the resumed records must not
escape the callback.
Records throwing = mock(Records.class);
when(throwing.batches()).thenThrow(new RuntimeException("boom"));
pending.complete(resultMap(logReadResult(new FetchDataInfo(null,
throwing), Errors.NONE)));
- assertTrue(resultFuture.get(10, TimeUnit.SECONDS).isEmpty());
+ assertTrue(resultFuture.get(10, TimeUnit.SECONDS).records().isEmpty());
}
@Test
@@ -320,26 +333,226 @@ class ShareGroupDLQRecordFetcherTest {
assertRecord(result, 2L, "k2", "v2");
}
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("compressedBatchCases")
+ public void testCompressedBatchPossibilitiesReturnCorrectMapAndOffset(
+ String description, List<SimpleRecord> batchRecords, int
maxDecompressedBytes,
+ Set<Long> expectedPresentOffsets, long expectedLastResolvedOffset)
throws Exception {
+ MemoryRecords batch = MemoryRecords.withRecords(0L,
Compression.gzip().build(),
+ batchRecords.toArray(new SimpleRecord[0]));
+ whenReadAsync(done(logReadResult(new FetchDataInfo(null, batch),
Errors.NONE)));
+
+ long lastOffset = batchRecords.size() - 1;
+ ShareGroupDLQRecordFetcher.FetchResult result =
+ fetcher(param(0L, lastOffset),
maxDecompressedBytes).fetch().get(10, TimeUnit.SECONDS);
+
+ assertEquals(expectedPresentOffsets.size(), result.records().size(),
description);
+ for (long offset = 0; offset <= lastOffset; offset++) {
+ Record record = result.records().get(offset);
+ if (expectedPresentOffsets.contains(offset)) {
+ assertTrue(record != null, description + ": expected offset "
+ offset + " to be present");
+ SimpleRecord expected = batchRecords.get((int) offset);
+ assertArrayEquals(toArray(expected.key()),
toArray(record.key()), description);
+ assertArrayEquals(toArray(expected.value()),
toArray(record.value()), description);
+ } else {
+ assertNull(record, description + ": expected offset " + offset
+ " to be absent");
+ }
+ }
+ assertEquals(expectedLastResolvedOffset, result.lastResolvedOffset(),
description);
+ }
+
+ // Covers every shape of compressed-batch decompression outcome: a single
record that fits, a single
+ // record that can never fit on its own, several small records that
individually fit but collectively
+ // exceed the budget, and a mix of one oversized record alongside several
individually-fitting ones -
+ // in the last two cases, decompressBounded() rejects the whole batch (see
its class-level doc), so
+ // even the individually-fitting records within it are excluded, not just
the oversized one.
+ private static Stream<Arguments> compressedBatchCases() {
+ SimpleRecord small0 = record("k0", "v".repeat(20));
+ SimpleRecord small1 = record("k1", "v".repeat(20));
+ SimpleRecord small2 = record("k2", "v".repeat(20));
+ SimpleRecord small3 = record("k3", "v".repeat(20));
+ SimpleRecord small4 = record("k4", "v".repeat(20));
+ SimpleRecord big0 = record("k0", "v".repeat(2000));
+ int smallSize = recordSizeInBytes(small0);
+
+ return Stream.of(
+ Arguments.of("single small record fits within budget",
+ List.of(small0), smallSize, Set.of(0L), 0L),
+ Arguments.of("single large record can never fit and is permanently
excluded",
+ List.of(big0), smallSize, Set.of(), 0L),
+ Arguments.of("multiple small records all fit within budget",
+ List.of(small0, small1, small2), smallSize * 3, Set.of(0L, 1L,
2L), 2L),
+ Arguments.of("many small records whose combined size exceeds
budget - whole batch excluded",
+ List.of(small0, small1, small2, small3, small4), smallSize *
3, Set.of(), 4L),
+ Arguments.of("combination of one large and several small records -
whole batch excluded",
+ List.of(big0, small1, small2, small3, small4), smallSize * 3,
Set.of(), 4L)
+ );
+ }
+
+ @Test
+ public void
testSingleReadWithUncompressedMagicV2AndLegacyCompressedBatchesAllCollected()
throws Exception {
+ // One read returning three consecutive batches, each using a
different collection path in
+ // collectFromBatch()'s dispatch: uncompressed (collectUncompressed),
magic v2 compressed
+ // (collectFromCompressedBatch, the bounded-buffer path), and legacy
magic v0/v1 compressed
+ // (collectWithCumulativeCap, the streaming fallback). All comfortably
fit within budget, so this
+ // is purely about correct dispatch and offset/content tracking across
a heterogeneous sequence,
+ // not budget edge cases (already covered by the batch-specific tests).
+ MemoryRecords uncompressedBatch = MemoryRecords.withRecords(0L,
Compression.NONE, record("k0", "v0"));
+ MemoryRecords magicV2CompressedBatch = MemoryRecords.withRecords(1L,
Compression.gzip().build(), record("k1", "v1"));
+ MemoryRecords legacyCompressedBatch =
MemoryRecords.withRecords(RecordBatch.MAGIC_VALUE_V1, 2L,
+ Compression.gzip().build(), record("k2", "v2"));
+
+ whenReadAsync(done(logReadResult(new FetchDataInfo(null,
+ concatBatches(uncompressedBatch, magicV2CompressedBatch,
legacyCompressedBatch)), Errors.NONE)));
+
+ Map<Long, Record> result = fetcher(param(0L, 2L),
10_000).fetch().get(10, TimeUnit.SECONDS).records();
+
+ assertEquals(3, result.size());
+ assertRecord(result, 0L, "k0", "v0");
+ assertRecord(result, 1L, "k1", "v1");
+ assertRecord(result, 2L, "k2", "v2");
+ }
+
+ @Test
+ public void
testFetchStopsOnceEnoughContentCollectedWithoutScanningWholeRange() throws
Exception {
+ // A large range (0..100), but a budget sized to exactly the one
record's own size: once that
+ // record is collected the target has been reached, and the fetch
stops issuing further reads
+ // rather than scanning all the way to endOffset looking for more.
+ SimpleRecord simpleRecord = record("k0", "v".repeat(50));
+ int recordSize = recordSizeInBytes(simpleRecord);
+ whenReadAsync(done(success(simpleRecord)));
+
+ Map<Long, Record> result = fetcher(param(0L, 100L),
recordSize).fetch().get(10, TimeUnit.SECONDS).records();
+
+ assertEquals(1, result.size());
+ assertRecord(result, 0L, "k0", "v".repeat(50));
+ verify(logReader, times(1)).readAsync(any(), anySet(), any(), any(),
anyBoolean());
+ }
+
+ @Test
+ public void testMultipleBatchesInSingleReadStopAtBudgetBetweenBatches()
throws Exception {
+ // Two uncompressed batches concatenated into ONE read response - the
first batch alone already
+ // reaches the budget. collectRecords() must stop BEFORE starting the
second batch (not just wait
+ // for runFrom()'s between-reads check), or a single read containing
many batches could keep being
+ // processed past the point where the fetch should have stopped.
+ String value = "v".repeat(50);
+ MemoryRecords batch0 = MemoryRecords.withRecords(0L, Compression.NONE,
record("k0", value));
+ MemoryRecords batch1 = MemoryRecords.withRecords(1L, Compression.NONE,
record("k1", value));
+ int recordSize = recordSizeInBytes(record("k0", value));
+
+ whenReadAsync(done(logReadResult(new FetchDataInfo(null,
concatBatches(batch0, batch1)), Errors.NONE)));
+
+ Map<Long, Record> result = fetcher(param(0L, 1L),
recordSize).fetch().get(10, TimeUnit.SECONDS).records();
+
+ assertEquals(1, result.size());
+ assertRecord(result, 0L, "k0", value);
+ assertNull(result.get(1L), "Second batch must not be collected once
the first alone reached budget");
+ }
+
+ @Test
+ public void testUncompressedBatchStopsMidBatchOnceBudgetReached() throws
Exception {
+ // A single uncompressed batch with several records, budget sized to
exactly the first record's
+ // own size - collectUncompressed() must stop mid-batch, not just
between batches/reads, since a
+ // single uncompressed batch carries no decompression risk and so has
no other size cap of its own.
+ // The record that would push collectedBytes over budget is excluded
(not included), and the fetch
+ // stops there rather than examining records after it in the same
batch.
+ String value = "v".repeat(50);
+ int recordSize = recordSizeInBytes(record("k0", value));
+ whenReadAsync(done(success(record("k0", value), record("k1", value),
record("k2", value))));
+
+ Map<Long, Record> result = fetcher(param(0L, 2L),
recordSize).fetch().get(10, TimeUnit.SECONDS).records();
+
+ assertEquals(1, result.size());
+ assertRecord(result, 0L, "k0", value);
+ assertNull(result.get(1L), "Budget was reached after the first record;
one that doesn't fit alongside it must not be collected");
+ assertNull(result.get(2L), "The fetch stops at the first record that
doesn't fit; later records in the same batch must not be examined");
+ }
+
+ @Test
+ public void testCompressedBatchAfterUncompressedRespectsCombinedBudget()
throws Exception {
+ // The first batch is uncompressed and alone consumes most of the
budget. The second batch is
+ // compressed and, on its own, decompresses to well within the full
budget - decompression succeeds
+ // - but it still must not be collected, because it doesn't fit
alongside collectedBytes from the
+ // uncompressed batch already retained this fetch. Proves the
per-record fit check gates on
+ // collectedBytes accumulated across every batch in the fetch, not
just what the current batch
+ // itself decompresses to.
+ String uncompressedValue = "v".repeat(150);
+ String compressedValue = "x".repeat(80);
+ MemoryRecords uncompressedBatch = MemoryRecords.withRecords(0L,
Compression.NONE, record("k0", uncompressedValue));
+ MemoryRecords compressedBatch = MemoryRecords.withRecords(1L,
Compression.gzip().build(), record("k1", compressedValue));
+
+ whenReadAsync(done(logReadResult(
+ new FetchDataInfo(null, concatBatches(uncompressedBatch,
compressedBatch)), Errors.NONE)));
+
+ Map<Long, Record> result = fetcher(param(0L, 1L), 200).fetch().get(10,
TimeUnit.SECONDS).records();
+
+ assertRecord(result, 0L, "k0", uncompressedValue);
+ assertNull(result.get(1L),
+ "Compressed batch must not be collected - it doesn't fit alongside
what the uncompressed batch already retained");
+ }
+
+ @Test
+ public void
testOversizedUncompressedRecordExcludedButLaterRecordStillCollected() throws
Exception {
+ // The first record's own size exceeds the entire budget on its own -
it can never fit, in this
+ // fetch or any other - so it's permanently excluded rather than
blocking the smaller record after
+ // it, which fits comfortably since the excluded record never consumed
any of collectedBytes.
+ SimpleRecord small = record("k1", "v1");
+ int budget = recordSizeInBytes(small);
+ whenReadAsync(done(success(record("k0", "v".repeat(500)), small)));
+
+ Map<Long, Record> result = fetcher(param(0L, 1L),
budget).fetch().get(10, TimeUnit.SECONDS).records();
+
+ assertNull(result.get(0L), "A record whose own size exceeds the budget
must be permanently excluded");
+ assertRecord(result, 1L, "k1", "v1");
+ }
+
+ @Test
+ public void testHardStopPreventsFurtherReads() throws Exception {
+ // Once a record is found that doesn't fit alongside what's already
been collected, the fetch stops
+ // immediately - it must not attempt further reads for the rest of the
range, even though offsets
+ // remain unexamined. Everything from that point on is left untouched
for a fresh fetch to retry.
+ String value = "v".repeat(50);
+ int recordSize = recordSizeInBytes(record("k0", value));
+ whenReadAsync(
+ done(logReadResult(new FetchDataInfo(null,
+ MemoryRecords.withRecords(0L, Compression.NONE, record("k0",
value))), Errors.NONE)),
+ done(logReadResult(new FetchDataInfo(null,
+ MemoryRecords.withRecords(1L, Compression.NONE, record("k1",
value))), Errors.NONE)),
+ done(logReadResult(new FetchDataInfo(null,
+ MemoryRecords.withRecords(2L, Compression.NONE, record("k2",
value))), Errors.NONE)));
+
+ ShareGroupDLQRecordFetcher.FetchResult result =
+ fetcher(param(0L, 10L), recordSize + 1).fetch().get(10,
TimeUnit.SECONDS);
+
+ assertEquals(1, result.records().size());
+ assertRecord(result.records(), 0L, "k0", value);
+ assertEquals(0L, result.lastResolvedOffset(), "Only the first record
was resolved; the rest must be left for a fresh fetch");
+ verify(logReader, times(2)).readAsync(any(), anySet(), any(), any(),
anyBoolean());
+ }
+
@Test
public void testCompressedBatchExceedingDecompressedBudgetSkipsBatch()
throws Exception {
// Highly compressible values so the batch is small on the wire but
decompresses well past a
// tiny budget - simulates a decompression-bomb-shaped batch. The
whole batch is decompressed
- // into a bounded buffer before any record is parsed, so exceeding the
budget yields none of its
- // records rather than a partial subset.
+ // into a bounded buffer before any record is parsed, so a batch that
can never fit within the
+ // budget is permanently excluded, yielding none of its records rather
than a partial subset.
String bigValue = "x".repeat(10_000);
whenReadAsync(done(success(Compression.gzip().build(),
record("k0", bigValue), record("k1", bigValue), record("k2",
bigValue))));
- Map<Long, Record> result = fetcher(param(0L, 2L), 100).fetch().get(10,
TimeUnit.SECONDS);
+ ShareGroupDLQRecordFetcher.FetchResult result =
+ fetcher(param(0L, 2L), 100).fetch().get(10, TimeUnit.SECONDS);
- assertTrue(result.isEmpty(), "Expected the over-budget batch to be
skipped, yielding no records");
+ assertTrue(result.records().isEmpty(), "Expected the over-budget batch
to be skipped, yielding no records");
+ assertEquals(2L, result.lastResolvedOffset(), "The whole excluded
batch must count as resolved so it is never retried");
}
@Test
public void
testCompressedBatchExceedingDecompressedBudgetSkippedButLaterBatchStillCollected()
throws Exception {
// First batch is highly compressible so it alone decompresses well
past a tiny budget; the second
// batch is uncompressed and carries no decompression risk. Only the
first batch's offsets should be
- // skipped - the fetch must not abort the whole range, so the second
batch is still collected.
+ // permanently excluded - the fetch must not abort the whole range, so
the second batch is still
+ // collected.
String bigValue = "x".repeat(10_000);
MemoryRecords compressedBatch = MemoryRecords.withRecords(0L,
Compression.gzip().build(),
record("k0", bigValue), record("k1", bigValue), record("k2",
bigValue));
@@ -348,7 +561,7 @@ class ShareGroupDLQRecordFetcherTest {
whenReadAsync(done(logReadResult(
new FetchDataInfo(null, concatBatches(compressedBatch,
uncompressedBatch)), Errors.NONE)));
- Map<Long, Record> result = fetcher(param(0L, 3L), 100).fetch().get(10,
TimeUnit.SECONDS);
+ Map<Long, Record> result = fetcher(param(0L, 3L), 100).fetch().get(10,
TimeUnit.SECONDS).records();
assertNull(result.get(0L));
assertNull(result.get(1L));
@@ -359,7 +572,8 @@ class ShareGroupDLQRecordFetcherTest {
@Test
public void
testLegacyCompressedBatchExceedingCumulativeCapSkipsBatchButLaterBatchStillCollected()
throws Exception {
// Legacy magic v0/v1 batches use the cumulative-cap fallback (no
bounded-buffer path for that
- // format). Exceeding the cap should still only skip the rest of the
offending batch, not abort the
+ // format). Each record's own size is checked individually as it's
decompressed: a record that can
+ // never fit on its own is permanently excluded, but the scan
continues rather than aborting the
// whole fetch - a later, uncompressed batch remains unaffected.
String bigValue = "x".repeat(10_000);
MemoryRecords legacyCompressedBatch =
MemoryRecords.withRecords(RecordBatch.MAGIC_VALUE_V1, 0L,
@@ -369,7 +583,7 @@ class ShareGroupDLQRecordFetcherTest {
whenReadAsync(done(logReadResult(
new FetchDataInfo(null, concatBatches(legacyCompressedBatch,
uncompressedBatch)), Errors.NONE)));
- Map<Long, Record> result = fetcher(param(0L, 3L), 100).fetch().get(10,
TimeUnit.SECONDS);
+ Map<Long, Record> result = fetcher(param(0L, 3L), 100).fetch().get(10,
TimeUnit.SECONDS).records();
assertNull(result.get(0L));
assertNull(result.get(1L));
diff --git
a/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java
b/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java
index 307003022b4..5e88c3f9dbc 100644
---
a/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java
+++
b/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java
@@ -54,8 +54,10 @@ import org.apache.kafka.test.TestUtils;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentMatcher;
import org.mockito.Mockito;
+import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
@@ -65,6 +67,7 @@ import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.OptionalLong;
import java.util.Set;
@@ -97,9 +100,12 @@ import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anySet;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
@@ -1716,22 +1722,141 @@ class ShareGroupDLQStateManagerTest {
verify(mockMetrics, times(3)).recordDLQProduce(GROUP_ID);
}
+ @Test
+ public void
testCompleteHandlerSafelyIsolatesOneHandlersUncaughtExceptionFromOthers() {
+ // Simulates the coalesced-response callback in
SendThread.generateRequests(), which invokes
+ // completeHandlerSafely() once per handler sharing one produce
response. If one handler's
+ // onComplete() throws, that must not prevent other handlers in the
same batch from being
+ // notified (which would otherwise leave their dlq() futures hanging
forever), and the
+ // offending handler's own future must still be explicitly failed
rather than left hanging.
+ stateManager = builder().build();
+ ShareGroupDLQStateManager.ProduceRequestHandler throwing =
+ spy(newHandlerForCoalesceTest(stateManager, GROUP_ID, 0));
+ throwing.populateDLQTopicData();
+ doThrow(new RuntimeException("boom")).when(throwing).onComplete(any());
+
+ ShareGroupDLQStateManager.ProduceRequestHandler normal =
+ spy(newHandlerForCoalesceTest(stateManager, GROUP_ID, 1));
+ normal.populateDLQTopicData();
+
+ // A null response is a valid, already-handled input to onComplete()
(see its own null
+ // check) and lets this test avoid needing a real ClientResponse.
+ List.of(throwing, normal).forEach(handler ->
+ ShareGroupDLQStateManager.completeHandlerSafely(handler, null));
+
+ verify(throwing).onComplete(null);
+ verify(throwing).requestErrorResponse(any(RuntimeException.class));
+ verify(normal).onComplete(null);
+ verify(normal, never()).requestErrorResponse(any());
+ }
+
+ @Test
+ public void
testResolveRoundSurvivesSynchronousExceptionFromMaybeFetchRecordData() throws
Exception {
+ // maybeFetchRecordData() begins with a synchronous cache-helper call
+ // (isShareGroupDlqCopyRecordEnabled) before ever returning a future.
If that call throws -
+ // e.g. a cache-helper bug - resolveRound() must not let the exception
escape synchronously:
+ // for round 2 onward this method is called directly from
handleProduceResponse() on the
+ // sender thread, inside a RequestCompletionHandler callback, where an
uncaught exception can
+ // (depending on the internal response-delivery path) propagate all
the way to the broker's
+ // fatal-error handling rather than just failing this one handler.
+ ShareGroupDLQMetadataCacheHelper cacheHelper =
cacheHelper(DEFAULT_LEADER);
+
when(cacheHelper.isShareGroupDlqCopyRecordEnabled(GROUP_ID)).thenThrow(new
RuntimeException("cache helper boom"));
+ stateManager = builder().withCacheHelper(cacheHelper).build();
+
+ // A multi-offset range - not newHandlerForCoalesceTest()'s usual
single-offset (0,0) convention.
+ // That distinction matters here: with only one offset in range, "the
single-record floor forces
+ // exactly one offset through" and "the whole remaining range is
deliberately given up on" look
+ // identical, which would mask a regression back to the narrower,
floor-only behavior. A
+ // multi-offset range only comes out right if
lastResolvedOffsetThisRound is actually set to
+ // param.lastOffset() on a total failure, not just nextOffsetToSend's
floor value.
+ ShareGroupDLQRecordParameter param = new ShareGroupDLQRecordParameter(
+ GROUP_ID, new TopicIdPartition(SOURCE_TOPIC_ID, 0, "source-topic"),
+ 0L, 2L, Optional.of((short) 1), Optional.of(new
RuntimeException("simulated cause")));
+ ShareGroupDLQStateManager.ProduceRequestHandler handler =
stateManager.new ProduceRequestHandler(
+ param, new CompletableFuture<>(),
ShareGroupDLQStateManager.REQUEST_BACKOFF_MS,
+ ShareGroupDLQStateManager.REQUEST_BACKOFF_MAX_MS, 3);
+ handler.populateDLQTopicData();
+
+ CompletableFuture<Void> resolved = handler.resolveRound();
+
+ assertNull(resolved.get(5, TimeUnit.SECONDS),
+ "resolveRound() must complete normally even when
maybeFetchRecordData() throws synchronously");
+
+ // Falls back to headers-only for the WHOLE remaining range in this
one round, same as any other
+ // total-failure case - not just a single, floored offset.
+ ProduceRequestData.TopicProduceData topicData =
handler.topicProduceData();
+ List<Record> records = new ArrayList<>();
+ ((MemoryRecords)
topicData.partitionData().get(0).records()).records().forEach(records::add);
+ assertEquals(3, records.size(),
+ "The whole 3-offset range must be given up on in this one round,
not just a single floored offset");
+ for (Record record : records) {
+ assertFalse(record.hasKey());
+ assertFalse(record.hasValue());
+ }
+ }
+
// --- DLQ record with copy record enabled ---
private static FetchDataInfo recordsInfo(SimpleRecord... records) {
return new FetchDataInfo(null,
MemoryRecords.withRecords(Compression.NONE, records));
}
+ // The exact on-wire size of a single uncompressed record, computed via
the real API rather than
+ // hardcoded, so a test can size dlqTopicMaxMessageBytes precisely against
a record's own raw size.
+ private static int recordSizeInBytes(SimpleRecord simpleRecord) {
+ MemoryRecords batch = MemoryRecords.withRecords(Compression.NONE,
simpleRecord);
+ return
batch.batches().iterator().next().iterator().next().sizeInBytes();
+ }
+
// A read result carrying the given data and error. Other read metadata is
irrelevant to the fetcher.
private static LogReadResult logReadResult(FetchDataInfo info, Errors
error) {
return new LogReadResult(info, Optional.empty(), 0L, 0L, 0L, 0L, -1L,
OptionalLong.empty(), error);
}
- private static CompletableFuture<LinkedHashMap<TopicIdPartition,
LogReadResult>> asyncReadMap(
+ private static LinkedHashMap<TopicIdPartition, LogReadResult> resultMap(
TopicIdPartition topicIdPartition, LogReadResult result) {
LinkedHashMap<TopicIdPartition, LogReadResult> map = new
LinkedHashMap<>();
map.put(topicIdPartition, result);
- return CompletableFuture.completedFuture(map);
+ return map;
+ }
+
+ private static CompletableFuture<LinkedHashMap<TopicIdPartition,
LogReadResult>> asyncReadMap(
+ TopicIdPartition topicIdPartition, LogReadResult result) {
+ return CompletableFuture.completedFuture(resultMap(topicIdPartition,
result));
+ }
+
+ // Concatenates the given batches' on-wire bytes into a single
MemoryRecords, simulating one
+ // read returning multiple consecutive batches.
+ private static MemoryRecords concatBatches(List<MemoryRecords> batches) {
+ int size = batches.stream().mapToInt(b ->
b.buffer().remaining()).sum();
+ ByteBuffer combined = ByteBuffer.allocate(size);
+ batches.forEach(b -> combined.put(b.buffer().duplicate()));
+ combined.flip();
+ return MemoryRecords.readableRecords(combined);
+ }
+
+ // Stubs logReader.readAsync to behave like a real log keyed by starting
offset: a read
+ // requesting offset N returns only the batches at index >= N
(batchesByOffset.get(i) is the
+ // batch whose base offset is i), never earlier ones - unlike a stub that
always returns the
+ // same fixed blob regardless of the requested position, which would waste
each round's fresh
+ // decompression budget re-decompressing already-consumed batches.
+ private static void whenReadAsyncFromLog(LogReader logReader,
TopicIdPartition topicIdPartition,
+ List<MemoryRecords>
batchesByOffset) {
+ when(logReader.readAsync(any(), anySet(), any(), any(),
anyBoolean())).thenAnswer(invocation -> {
+ Map<TopicIdPartition, Long> offsets = invocation.getArgument(2);
+ int fromIndex = (int) Math.min(offsets.get(topicIdPartition),
batchesByOffset.size());
+ MemoryRecords records = fromIndex >= batchesByOffset.size()
+ ? MemoryRecords.EMPTY
+ : concatBatches(batchesByOffset.subList(fromIndex,
batchesByOffset.size()));
+ return asyncReadMap(topicIdPartition, logReadResult(new
FetchDataInfo(null, records), Errors.NONE));
+ });
+ }
+
+ // Matches the offsets map argument of readAsync() when it requests
starting from the given
+ // offset for the given partition - lets tests assert readAsync was (or
wasn't) called for a
+ // specific round's window, rather than relying on call order/count alone.
+ private static ArgumentMatcher<Map<TopicIdPartition, Long>>
offsetsRequesting(TopicIdPartition topicIdPartition, long offset) {
+ return offsets -> offsets != null &&
Objects.equals(offsets.get(topicIdPartition), offset);
}
// Stubs logReader.readAsync to return, in order, the given per-call
results for the partition,
@@ -1829,8 +1954,11 @@ class ShareGroupDLQStateManagerTest {
stateManager.start();
assertNull(stateManager.dlq(param, 1L, 5L, maxAttempts).get(5,
TimeUnit.SECONDS));
- // Records are resolved once, before enqueue, and the memoized
result is reused on every
- // (re)send. So despite three produce attempts the source log is
read exactly once.
+ // param() fits entirely in one produce round here (no chunking),
so all three attempts
+ // are retries of round 1: records are resolved once for that
round, before enqueue, and
+ // the memoized result is reused on every retry. So despite three
produce attempts the
+ // source log is read exactly once - a genuinely new round (not
exercised by this test)
+ // would trigger its own, separate readAsync call.
verify(logReader, times(1)).readAsync(any(), anySet(), any(),
any(), anyBoolean());
verify(mockMetrics, times(maxAttempts)).recordDLQProduce(GROUP_ID);
verify(mockMetrics).recordDLQRecordWrite(GROUP_ID, 3);
@@ -1840,6 +1968,254 @@ class ShareGroupDLQStateManagerTest {
}
}
+ @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.
+ int maxMessageBytes = 600;
+ ShareGroupDLQRecordParameter param = param();
+ TopicIdPartition tp = param.topicIdPartition();
+
+ List<MemoryRecords> batchesByOffset = new ArrayList<>();
+ List<byte[]> keys = new ArrayList<>();
+ List<byte[]> values = new ArrayList<>();
+ for (long offset = 0; offset <= 2; offset++) {
+ byte[] key = ("k" + offset).getBytes(StandardCharsets.UTF_8);
+ byte[] value = ("x".repeat(400) +
offset).getBytes(StandardCharsets.UTF_8);
+ keys.add(key);
+ values.add(value);
+ batchesByOffset.add(MemoryRecords.withRecords(offset,
Compression.gzip().build(),
+ new SimpleRecord(MOCK_TIME.milliseconds(), key, value)));
+ }
+
+ LogReader logReader = mock(LogReader.class);
+ whenReadAsyncFromLog(logReader, tp, batchesByOffset);
+
+ ShareGroupDLQMetadataCacheHelper cacheHelper =
cacheHelper(DEFAULT_LEADER);
+
when(cacheHelper.isShareGroupDlqCopyRecordEnabled(any())).thenReturn(true);
+
when(cacheHelper.dlqTopicMaxMessageBytes(anyString())).thenReturn(maxMessageBytes);
+
+ MockClient client = new MockClient(MOCK_TIME);
+ List<ProduceRequest> capturedProduces = new ArrayList<>();
+ MockClient.RequestMatcher captureProduce = body -> {
+ if (body instanceof ProduceRequest pr) {
+ capturedProduces.add(pr);
+ return true;
+ }
+ return false;
+ };
+ client.prepareResponseFrom(captureProduce,
successfulProduceResponse(0), DEFAULT_LEADER);
+ client.prepareResponseFrom(captureProduce,
successfulProduceResponse(0), DEFAULT_LEADER);
+ client.prepareResponseFrom(captureProduce,
successfulProduceResponse(0), DEFAULT_LEADER);
+
+ stateManager =
builder().withClient(client).withLogReader(logReader).withCacheHelper(cacheHelper).build();
+ stateManager.start();
+ assertNull(stateManager.dlq(param).get(10, TimeUnit.SECONDS));
+
+ assertEquals(3, capturedProduces.size(), "Expected one produce round
per offset");
+ Map<String, String> sharedHeaders = Map.of(
+ HEADER_DLQ_ERRORS_TOPIC, "source-topic",
+ HEADER_DLQ_ERRORS_PARTITION, "0",
+ HEADER_DLQ_ERRORS_GROUP, GROUP_ID,
+ HEADER_DLQ_ERRORS_DELIVERY_COUNT, "1",
+ HEADER_DLQ_ERRORS_MESSAGE, "simulated cause"
+ );
+ for (int round = 0; round < 3; round++) {
+ assertDlqProduceRecordHeaders(capturedProduces.get(round), Map.of(
+ 0, new ExpectedDlqPartition(round, round, sharedHeaders,
List.of(keys.get(round)), List.of(values.get(round)))
+ ));
+ }
+ verify(logReader, times(3)).readAsync(any(), anySet(), any(), any(),
anyBoolean());
+ }
+
+ @Test
+ public void
testFetcherBudgetSplitsRoundEvenWhenSizeFloorWouldHaveRoomForMore() throws
Exception {
+ // Isolates topicProduceData()'s lastResolvedOffsetThisRound cap from
the pre-existing
+ // single-record size floor (see
testDlqChunksOverMaxMessageBytesAcrossMultipleProduceRequests):
+ // dlqTopicMaxMessageBytes is generous enough that, size-wise alone,
round 1 could easily combine
+ // offset 0's real content with a cheap headers-only filler for offset
1 - a DLQ record is mostly
+ // fixed header overhead, so an unresolved offset costs very little to
pack in on top of real
+ // content that already fits. Without the cap, topicProduceData()
would happily sweep offset 1 in
+ // as headers-only and complete the whole range in one round,
permanently losing its real content
+ // (nextOffsetToSend would advance past it, and a fresh fetch would
never be attempted for it).
+ // With the cap, round 1 is limited to exactly what the fetcher
actually resolved - offset 0 alone,
+ // since its 2000-byte value combined with offset 1's would exceed the
fetcher's own budget even
+ // though each fits comfortably on its own - and offset 1 is correctly
deferred to a fresh round.
+ int maxMessageBytes = 3200;
+ ShareGroupDLQRecordParameter param = new ShareGroupDLQRecordParameter(
+ GROUP_ID, new TopicIdPartition(SOURCE_TOPIC_ID, 0, "source-topic"),
+ 0L, 1L, Optional.of((short) 1), Optional.of(new
RuntimeException("simulated cause")));
+ TopicIdPartition tp = param.topicIdPartition();
+
+ byte[] key0 = "k0".getBytes(StandardCharsets.UTF_8);
+ byte[] value0 = "x".repeat(2000).getBytes(StandardCharsets.UTF_8);
+ byte[] key1 = "k1".getBytes(StandardCharsets.UTF_8);
+ byte[] value1 = "y".repeat(2000).getBytes(StandardCharsets.UTF_8);
+ List<MemoryRecords> batchesByOffset = List.of(
+ MemoryRecords.withRecords(0L, Compression.gzip().build(),
+ new SimpleRecord(MOCK_TIME.milliseconds(), key0, value0)),
+ MemoryRecords.withRecords(1L, Compression.gzip().build(),
+ new SimpleRecord(MOCK_TIME.milliseconds(), key1, value1))
+ );
+ LogReader logReader = mock(LogReader.class);
+ whenReadAsyncFromLog(logReader, tp, batchesByOffset);
+
+ ShareGroupDLQMetadataCacheHelper cacheHelper =
cacheHelper(DEFAULT_LEADER);
+
when(cacheHelper.isShareGroupDlqCopyRecordEnabled(any())).thenReturn(true);
+
when(cacheHelper.dlqTopicMaxMessageBytes(anyString())).thenReturn(maxMessageBytes);
+
+ MockClient client = new MockClient(MOCK_TIME);
+ List<ProduceRequest> capturedProduces = new ArrayList<>();
+ MockClient.RequestMatcher captureProduce = body -> {
+ if (body instanceof ProduceRequest pr) {
+ capturedProduces.add(pr);
+ return true;
+ }
+ return false;
+ };
+ client.prepareResponseFrom(captureProduce,
successfulProduceResponse(0), DEFAULT_LEADER);
+ client.prepareResponseFrom(captureProduce,
successfulProduceResponse(0), DEFAULT_LEADER);
+
+ stateManager =
builder().withClient(client).withLogReader(logReader).withCacheHelper(cacheHelper).build();
+ stateManager.start();
+ assertNull(stateManager.dlq(param).get(10, TimeUnit.SECONDS));
+
+ assertEquals(2, capturedProduces.size(),
+ "Round 1 must stop at offset 0 (the fetcher's own budget), not
sweep offset 1 in as " +
+ "headers-only just because topicProduceData()'s size floor would
have had room for it");
+ Map<String, String> sharedHeaders = Map.of(
+ HEADER_DLQ_ERRORS_TOPIC, "source-topic",
+ HEADER_DLQ_ERRORS_PARTITION, "0",
+ HEADER_DLQ_ERRORS_GROUP, GROUP_ID,
+ HEADER_DLQ_ERRORS_DELIVERY_COUNT, "1",
+ HEADER_DLQ_ERRORS_MESSAGE, "simulated cause"
+ );
+ assertDlqProduceRecordHeaders(capturedProduces.get(0), Map.of(
+ 0, new ExpectedDlqPartition(0L, 0L, sharedHeaders, List.of(key0),
List.of(value0))
+ ));
+ assertDlqProduceRecordHeaders(capturedProduces.get(1), Map.of(
+ 0, new ExpectedDlqPartition(1L, 1L, sharedHeaders, List.of(key1),
List.of(value1))
+ ));
+ verify(logReader, times(2)).readAsync(any(), anySet(), any(), any(),
anyBoolean());
+ }
+
+ @Test
+ public void
testDlqRecordCopyFirstProduceRoundGoesOutBeforeLaterRoundResolves() throws
Exception {
+ // Forces exactly one offset per produce round via
topicProduceData()'s single-record floor
+ // (see
testDlqChunksOverMaxMessageBytesAcrossMultipleProduceRequests): the DLQ headers
alone
+ // (topic, partition, offset, group, delivery count, cause message)
make the wrapped record far
+ // bigger than dlqTopicMaxMessageBytes below, regardless of the source
record's own tiny size, so
+ // the floor kicks in on the very first offset every round. Round 1's
read is already resolved;
+ // round 2's read is held pending to simulate an in-flight (e.g.
remote-storage) fetch - proving
+ // round 1's produce goes out without waiting for round 2 to resolve.
+ ShareGroupDLQRecordParameter param = new ShareGroupDLQRecordParameter(
+ GROUP_ID, new TopicIdPartition(SOURCE_TOPIC_ID, 0, "source-topic"),
+ 0L, 1L, Optional.of((short) 1), Optional.of(new
RuntimeException("simulated cause")));
+ TopicIdPartition tp = param.topicIdPartition();
+
+ // Round 1's own fetch window covers the WHOLE param range (offsets 0
and 1, since round 1
+ // always starts at param.firstOffset()) - but dlqTopicMaxMessageBytes
below is sized to exactly
+ // offset 0's own record, so the fetch collects it, immediately
reaches its target, and stops on
+ // its own without needing to look for offset 1 at all. The genuinely
separate, held-pending read
+ // only comes later, from round 2's own resolveRound() call (triggered
by round 1's successful
+ // produce response).
+ SimpleRecord record0 = new SimpleRecord(MOCK_TIME.milliseconds(),
+ "k0".getBytes(StandardCharsets.UTF_8),
"v0".getBytes(StandardCharsets.UTF_8));
+ LogReader logReader = mock(LogReader.class);
+ CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
round1Read = asyncReadMap(tp,
+ logReadResult(recordsInfo(record0), Errors.NONE));
+ CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
round2Read = new CompletableFuture<>();
+ when(logReader.readAsync(any(), anySet(), any(), any(), anyBoolean()))
+ .thenReturn(round1Read)
+ .thenReturn(round2Read);
+
+ ShareGroupDLQMetadataCacheHelper cacheHelper =
cacheHelper(DEFAULT_LEADER);
+
when(cacheHelper.isShareGroupDlqCopyRecordEnabled(any())).thenReturn(true);
+
when(cacheHelper.dlqTopicMaxMessageBytes(anyString())).thenReturn(recordSizeInBytes(record0));
+
+ MockClient client = new MockClient(MOCK_TIME);
+ List<ProduceRequest> capturedProduces = new ArrayList<>();
+ MockClient.RequestMatcher captureProduce = body -> {
+ if (body instanceof ProduceRequest pr) {
+ capturedProduces.add(pr);
+ return true;
+ }
+ return false;
+ };
+ client.prepareResponseFrom(captureProduce,
successfulProduceResponse(0), DEFAULT_LEADER);
+ client.prepareResponseFrom(captureProduce,
successfulProduceResponse(0), DEFAULT_LEADER);
+
+ stateManager =
builder().withClient(client).withLogReader(logReader).withCacheHelper(cacheHelper).build();
+ stateManager.start();
+ CompletableFuture<Void> dlqFuture = stateManager.dlq(param);
+
+ TestUtils.waitForCondition(() -> capturedProduces.size() == 1,
TestUtils.DEFAULT_MAX_WAIT_MS,
+ "Expected round 1's produce request to go out while round 2's
fetch is still pending");
+ assertFalse(dlqFuture.isDone(), "Overall future must not complete
until round 2 resolves and is sent");
+
+ round2Read.complete(resultMap(tp, logReadResult(recordsInfo(new
SimpleRecord(MOCK_TIME.milliseconds(),
+ "k1".getBytes(StandardCharsets.UTF_8),
"v1".getBytes(StandardCharsets.UTF_8))), Errors.NONE)));
+
+ assertNull(dlqFuture.get(10, TimeUnit.SECONDS));
+ assertEquals(2, capturedProduces.size(), "Expected round 2's produce
request once its fetch resolved");
+ }
+
+ @Test
+ public void
testDlqRecordCopyRetryOfSameRoundDoesNotRefetchButNewRoundDoes() throws
Exception {
+ int maxAttempts = 3;
+ Timer realTimer = new SystemTimerReaper("shareGroupDLQTestTimer", new
SystemTimer("shareGroupDLQTestTimer"));
+ try {
+ // 2-offset range, 1-byte max message bytes forces exactly one
offset per round (same
+ // trick as
testDlqChunksOverMaxMessageBytesAcrossMultipleProduceRequests), giving
+ // round 1 = offset 0, round 2 = offset 1.
+ ShareGroupDLQRecordParameter param = new
ShareGroupDLQRecordParameter(
+ GROUP_ID, new TopicIdPartition(SOURCE_TOPIC_ID, 0,
"source-topic"),
+ 0L, 1L, Optional.of((short) 1), Optional.of(new
RuntimeException("simulated cause")));
+ TopicIdPartition tp = param.topicIdPartition();
+
+ List<MemoryRecords> batchesByOffset = List.of(
+ MemoryRecords.withRecords(0L, Compression.NONE,
+ new SimpleRecord(MOCK_TIME.milliseconds(),
"k0".getBytes(StandardCharsets.UTF_8), "v0".getBytes(StandardCharsets.UTF_8))),
+ MemoryRecords.withRecords(1L, Compression.NONE,
+ new SimpleRecord(MOCK_TIME.milliseconds(),
"k1".getBytes(StandardCharsets.UTF_8), "v1".getBytes(StandardCharsets.UTF_8)))
+ );
+ LogReader logReader = mock(LogReader.class);
+ whenReadAsyncFromLog(logReader, tp, batchesByOffset);
+
+ ShareGroupDLQMetadataCacheHelper cacheHelper =
cacheHelper(DEFAULT_LEADER);
+
when(cacheHelper.isShareGroupDlqCopyRecordEnabled(any())).thenReturn(true);
+
when(cacheHelper.dlqTopicMaxMessageBytes(anyString())).thenReturn(1);
+
+ MockClient client = new MockClient(MOCK_TIME);
+ // Round 1 attempt 1: disconnect (retriable). Round 1 attempt 2:
success. Round 2: success.
+ client.prepareResponseFrom(body -> body instanceof ProduceRequest,
null, DEFAULT_LEADER, true);
+ client.prepareResponseFrom(body -> body instanceof ProduceRequest,
successfulProduceResponse(0), DEFAULT_LEADER);
+ client.prepareResponseFrom(body -> body instanceof ProduceRequest,
successfulProduceResponse(0), DEFAULT_LEADER);
+
+ stateManager = builder()
+ .withClient(client)
+ .withLogReader(logReader)
+ .withCacheHelper(cacheHelper)
+ .withTimer(realTimer)
+ .build();
+ stateManager.start();
+ assertNull(stateManager.dlq(param, 1L, 5L, maxAttempts).get(10,
TimeUnit.SECONDS));
+
+ // Despite round 1 needing 2 produce attempts (1 retry), its
window (offset 0) is only
+ // fetched once - the retry reuses the already-resolved data (see
dlqTopicExists()).
+ // Round 2's window (offset 1) is a genuinely new round and gets
its own, separate fetch.
+ verify(logReader, times(1)).readAsync(any(), anySet(),
argThat(offsetsRequesting(tp, 0L)), any(), anyBoolean());
+ verify(logReader, times(1)).readAsync(any(), anySet(),
argThat(offsetsRequesting(tp, 1L)), any(), anyBoolean());
+ verify(logReader, times(2)).readAsync(any(), anySet(), any(),
any(), anyBoolean());
+ } finally {
+ Utils.closeQuietly(realTimer, "shareGroupDLQTestTimer");
+ }
+ }
+
@Test
public void testDLQRecordCopyEnabledButInvalidConfigSkipsFetch() throws
Exception {
LogReader logReader = mock(LogReader.class);
