This is an automated email from the ASF dual-hosted git repository.
apoorvmittal10 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 e0d2d650ed1 KAFKA-20678: Support share group DLQ remote record fetch.
(#22601)
e0d2d650ed1 is described below
commit e0d2d650ed1ae2b68624fd8de9d9db7fa266267c
Author: Sushant Mahajan <[email protected]>
AuthorDate: Tue Jun 30 03:20:00 2026 +0530
KAFKA-20678: Support share group DLQ remote record fetch. (#22601)
* In this PR, we have extracted the record fetch logic from
`ShareGroupDLQStateManager` and encapsulated it in
`ShareGroupDLQRecordFetcher`.
* The fetcher also gets the ability to fetch the records from the remote
tier and then returns the combined record map via its public `fetch`
method.
* Since the remote record fetch is inherently async, there is a change
in the way the records are populated:
* Earlier records population was lazy in that it was invoked when
handlers were coalesced. Now, we will issue the record fetch call when
the handler is enqueued. Otherwise, record population will block the
`Sender` thread.
* Furthermore, we also require storing additional state in the fetcher
so
that local and remote records could be held until combined.
This is also divergent from previous code.
* Lastly, we have added tests to `ShareGroupDLQRecordFetcherTest` and
`ReplicaManagerLogReaderTest`.
Reviewers: Apoorv Mittal <[email protected]>
---
.../server/share/ReplicaManagerLogReader.java | 128 +++++++
.../server/share/ReplicaManagerLogReaderTest.java | 403 +++++++++++++++++++++
.../org/apache/kafka/server/share/LogReader.java | 32 ++
.../share/dlq/ShareGroupDLQRecordFetcher.java | 226 ++++++++++++
.../share/dlq/ShareGroupDLQStateManager.java | 182 ++++------
.../share/dlq/ShareGroupDLQRecordFetcherTest.java | 291 +++++++++++++++
.../share/dlq/ShareGroupDLQStateManagerTest.java | 340 +++++++++++++----
7 files changed, 1425 insertions(+), 177 deletions(-)
diff --git a/core/src/main/java/kafka/server/share/ReplicaManagerLogReader.java
b/core/src/main/java/kafka/server/share/ReplicaManagerLogReader.java
index 71ce7c74ca9..acebdb6a735 100644
--- a/core/src/main/java/kafka/server/share/ReplicaManagerLogReader.java
+++ b/core/src/main/java/kafka/server/share/ReplicaManagerLogReader.java
@@ -20,10 +20,14 @@ import kafka.server.QuotaFactory;
import kafka.server.ReplicaManager;
import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.FetchRequest;
+import org.apache.kafka.server.log.remote.storage.RemoteLogManager;
import org.apache.kafka.server.share.LogReader;
import org.apache.kafka.server.storage.log.FetchParams;
+import org.apache.kafka.storage.internals.log.FetchDataInfo;
import org.apache.kafka.storage.internals.log.LogReadResult;
+import org.apache.kafka.storage.internals.log.RemoteStorageFetchInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -31,11 +35,14 @@ import org.slf4j.LoggerFactory;
import java.util.LinkedHashMap;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.stream.Collectors;
import scala.Tuple2;
import scala.collection.Seq;
import scala.jdk.javaapi.CollectionConverters;
+import scala.jdk.javaapi.OptionConverters;
import scala.runtime.BoxedUnit;
/**
@@ -92,4 +99,125 @@ public class ReplicaManagerLogReader implements LogReader {
log.trace("Data successfully retrieved by replica manager: {}",
responseData);
return responseData;
}
+
+ @Override
+ public CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
readAsync(
+ FetchParams fetchParams,
+ Set<TopicIdPartition> partitionsToFetch,
+ LinkedHashMap<TopicIdPartition, Long> topicPartitionFetchOffsets,
+ LinkedHashMap<TopicIdPartition, Integer> partitionMaxBytes,
+ boolean readRemote) {
+
+ if (partitionsToFetch.isEmpty()) {
+ return CompletableFuture.completedFuture(new LinkedHashMap<>());
+ }
+
+ // Perform the local read for all partitions once; remote follow-ups
(if any) are issued per partition.
+ LinkedHashMap<TopicIdPartition, LogReadResult> localReadResults =
+ read(fetchParams, partitionsToFetch, topicPartitionFetchOffsets,
partitionMaxBytes);
+
+ // One future per partition; combined into a single future once every
partition has resolved.
+ LinkedHashMap<TopicIdPartition, CompletableFuture<LogReadResult>>
futures = new LinkedHashMap<>();
+ for (TopicIdPartition topicIdPartition : partitionsToFetch) {
+ LogReadResult logReadResult =
localReadResults.get(topicIdPartition);
+ if (logReadResult == null) {
+ futures.put(topicIdPartition,
CompletableFuture.completedFuture(
+ new LogReadResult(Errors.UNKNOWN_SERVER_ERROR)));
+ continue;
+ }
+
+ FetchDataInfo localFetchDataInfo = logReadResult.info();
+ Errors error = logReadResult.error();
+ Optional<RemoteStorageFetchInfo> remoteStorageFetchInfo =
localFetchDataInfo.delayedRemoteStorageFetch;
+
+ // Return the local read directly when it carries data, when it
failed, or when the data is tiered
+ // but the caller does not want remote reads (those offsets are
simply skipped).
+ if (error != Errors.NONE || remoteStorageFetchInfo.isEmpty() ||
!readRemote) {
+ futures.put(topicIdPartition,
CompletableFuture.completedFuture(logReadResult));
+ continue;
+ }
+
+ // Tiered data - follow it to the remote tier asynchronously,
wrapping the remote read into a
+ // LogReadResult that carries the metadata from the local read.
+ futures.put(topicIdPartition,
readRemote(remoteStorageFetchInfo.get()).handle((remoteFetchDataInfo,
exception) -> {
+ if (exception != null) {
+ Throwable cause = exception instanceof CompletionException
&& exception.getCause() != null
+ ? exception.getCause() : exception;
+ log.warn("Unable to read partition {} from remote
storage.", topicIdPartition, cause);
+ return withInfoAndError(logReadResult, localFetchDataInfo,
Errors.forException(cause));
+ }
+ if (remoteFetchDataInfo == null) {
+ return withInfoAndError(logReadResult, localFetchDataInfo,
Errors.UNKNOWN_SERVER_ERROR);
+ }
+ return withInfoAndError(logReadResult, remoteFetchDataInfo,
Errors.NONE);
+ }));
+ }
+
+ return CompletableFuture.allOf(futures.values().toArray(new
CompletableFuture<?>[0]))
+ .thenApply(ignored -> {
+ LinkedHashMap<TopicIdPartition, LogReadResult> results = new
LinkedHashMap<>();
+ futures.forEach((topicIdPartition, future) ->
results.put(topicIdPartition, future.getNow(null)));
+ return results;
+ });
+ }
+
+ /**
+ * Returns a copy of {@code base} with its read data ({@code info}) and
{@code error} replaced, preserving
+ * all other read metadata (high watermark, log offsets, etc.). Used to
wrap a remote read into a
+ * LogReadResult that carries the metadata from the originating local read.
+ */
+ private static LogReadResult withInfoAndError(LogReadResult base,
FetchDataInfo info, Errors error) {
+ return new LogReadResult(
+ info,
+ base.divergingEpoch(),
+ base.highWatermark(),
+ base.leaderLogStartOffset(),
+ base.leaderLogEndOffset(),
+ base.followerLogStartOffset(),
+ base.fetchTimeMs(),
+ base.lastStableOffset(),
+ base.preferredReadReplica(),
+ error);
+ }
+
+ /**
+ * Reads asynchronously from the remote tier for an offset tiered off the
local log. The
+ * RemoteStorageFetchInfo is the descriptor surfaced by a prior local read
as
+ * FetchDataInfo#delayedRemoteStorageFetch. The read runs on the remote
storage reader pool so the
+ * caller's thread is not blocked; the future completes exceptionally when
remote storage is not
+ * configured or the read could not be completed. Used internally by
readAsync (package-private so
+ * it remains unit-testable).
+ */
+ // Visibility for testing
+ CompletableFuture<FetchDataInfo> readRemote(RemoteStorageFetchInfo
remoteStorageFetchInfo) {
+ CompletableFuture<FetchDataInfo> future = new CompletableFuture<>();
+
+ Optional<RemoteLogManager> remoteLogManager =
OptionConverters.toJava(replicaManager.remoteLogManager());
+ if (remoteLogManager.isEmpty()) {
+ future.completeExceptionally(new IllegalStateException(
+ "Cannot read " + remoteStorageFetchInfo + " from remote
storage as remote log manager is not configured."));
+ return future;
+ }
+
+ try {
+ // The read runs on the remote storage reader thread pool; the
callback completes the
+ // future on that pool's thread, so the caller's thread is never
blocked on remote IO.
+ remoteLogManager.get().asyncRead(remoteStorageFetchInfo, result ->
{
+ if (result.error().isPresent()) {
+ future.completeExceptionally(result.error().get());
+ } else if (result.fetchDataInfo().isPresent()) {
+ future.complete(result.fetchDataInfo().get());
+ } else {
+ future.completeExceptionally(new IllegalStateException(
+ "Remote read for " + remoteStorageFetchInfo + "
returned neither data nor error."));
+ }
+ });
+ } catch (Exception e) {
+ // e.g. RejectedExecutionException if the reader pool is shutting
down.
+ log.warn("Unable to schedule remote read for {}.",
remoteStorageFetchInfo, e);
+ future.completeExceptionally(e);
+ }
+
+ return future;
+ }
}
diff --git
a/core/src/test/java/kafka/server/share/ReplicaManagerLogReaderTest.java
b/core/src/test/java/kafka/server/share/ReplicaManagerLogReaderTest.java
new file mode 100644
index 00000000000..f0f75b329f2
--- /dev/null
+++ b/core/src/test/java/kafka/server/share/ReplicaManagerLogReaderTest.java
@@ -0,0 +1,403 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.server.share;
+
+import kafka.server.ReplicaManager;
+
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.record.internal.MemoryRecords;
+import org.apache.kafka.common.requests.FetchRequest;
+import org.apache.kafka.server.log.remote.storage.RemoteLogManager;
+import org.apache.kafka.server.storage.log.FetchIsolation;
+import org.apache.kafka.server.storage.log.FetchParams;
+import org.apache.kafka.storage.internals.log.FetchDataInfo;
+import org.apache.kafka.storage.internals.log.LogOffsetMetadata;
+import org.apache.kafka.storage.internals.log.LogReadResult;
+import org.apache.kafka.storage.internals.log.RemoteLogReadResult;
+import org.apache.kafka.storage.internals.log.RemoteStorageFetchInfo;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+
+import scala.Option;
+import scala.Tuple2;
+import scala.collection.immutable.Seq;
+import scala.jdk.javaapi.CollectionConverters;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class ReplicaManagerLogReaderTest {
+
+ private static final TopicIdPartition TOPIC_ID_PARTITION =
+ new TopicIdPartition(Uuid.randomUuid(), 0, "topic");
+ private static final TopicIdPartition TOPIC_ID_PARTITION_2 =
+ new TopicIdPartition(Uuid.randomUuid(), 1, "topic");
+
+ private static RemoteStorageFetchInfo remoteStorageFetchInfo() {
+ return new RemoteStorageFetchInfo(
+ 1024,
+ true,
+ TOPIC_ID_PARTITION,
+ new FetchRequest.PartitionData(TOPIC_ID_PARTITION.topicId(), 0L,
0L, 1024, Optional.empty()),
+ FetchIsolation.HIGH_WATERMARK);
+ }
+
+ private static FetchParams fetchParams() {
+ return new FetchParams(
+ FetchRequest.CONSUMER_REPLICA_ID, -1, 0L, 1, 1024,
FetchIsolation.HIGH_WATERMARK, Optional.empty());
+ }
+
+ private static void stubReadFromLog(ReplicaManager replicaManager,
List<Tuple2<TopicIdPartition, LogReadResult>> results) {
+ when(replicaManager.readFromLog(any(), any(), any(), anyBoolean()))
+ .thenReturn(CollectionConverters.asScala(results).toSeq());
+ }
+
+ private static LogReadResult localReadResult(FetchDataInfo info, Errors
error) {
+ LogReadResult logReadResult = mock(LogReadResult.class);
+ when(logReadResult.info()).thenReturn(info);
+ when(logReadResult.error()).thenReturn(error);
+ return logReadResult;
+ }
+
+ private static FetchDataInfo localData() {
+ return new FetchDataInfo(new LogOffsetMetadata(0L),
MemoryRecords.EMPTY);
+ }
+
+ private static FetchDataInfo tieredData(RemoteStorageFetchInfo
remoteStorageFetchInfo) {
+ return new FetchDataInfo(new LogOffsetMetadata(0L),
MemoryRecords.EMPTY, false,
+ Optional.empty(), Optional.of(remoteStorageFetchInfo));
+ }
+
+ private static LinkedHashMap<TopicIdPartition, Long>
offsets(TopicIdPartition... partitions) {
+ LinkedHashMap<TopicIdPartition, Long> offsets = new LinkedHashMap<>();
+ for (TopicIdPartition partition : partitions) {
+ offsets.put(partition, 0L);
+ }
+ return offsets;
+ }
+
+ private static LinkedHashMap<TopicIdPartition, Integer>
maxBytes(TopicIdPartition... partitions) {
+ LinkedHashMap<TopicIdPartition, Integer> maxBytes = new
LinkedHashMap<>();
+ for (TopicIdPartition partition : partitions) {
+ maxBytes.put(partition, 1024);
+ }
+ return maxBytes;
+ }
+
+ @Test
+ public void testReadReturnsEmptyWhenNoPartitionsToFetch() {
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+
+ ReplicaManagerLogReader logReader = new
ReplicaManagerLogReader(replicaManager);
+ LinkedHashMap<TopicIdPartition, LogReadResult> result =
+ logReader.read(fetchParams(), Set.of(), new LinkedHashMap<>(), new
LinkedHashMap<>());
+
+ assertTrue(result.isEmpty());
+ verify(replicaManager, never()).readFromLog(any(), any(), any(),
anyBoolean());
+ }
+
+ @Test
+ public void testReadReturnsResultsFromReplicaManager() {
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ LogReadResult logReadResult = mock(LogReadResult.class);
+ Seq<Tuple2<TopicIdPartition, LogReadResult>> readFromLogResult =
+ CollectionConverters.asScala(List.of(new
Tuple2<>(TOPIC_ID_PARTITION, logReadResult))).toSeq();
+ when(replicaManager.readFromLog(any(), any(), any(),
anyBoolean())).thenReturn(readFromLogResult);
+
+ LinkedHashMap<TopicIdPartition, Long> offsets = new LinkedHashMap<>();
+ offsets.put(TOPIC_ID_PARTITION, 5L);
+ LinkedHashMap<TopicIdPartition, Integer> maxBytes = new
LinkedHashMap<>();
+ maxBytes.put(TOPIC_ID_PARTITION, 1024);
+
+ ReplicaManagerLogReader logReader = new
ReplicaManagerLogReader(replicaManager);
+ LinkedHashMap<TopicIdPartition, LogReadResult> result =
+ logReader.read(fetchParams(), Set.of(TOPIC_ID_PARTITION), offsets,
maxBytes);
+
+ assertEquals(1, result.size());
+ assertSame(logReadResult, result.get(TOPIC_ID_PARTITION));
+ }
+
+ @Test
+ public void
testReadRemoteCompletesExceptionallyWhenRemoteLogManagerNotConfigured() {
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ when(replicaManager.remoteLogManager()).thenReturn(Option.empty());
+
+ ReplicaManagerLogReader logReader = new
ReplicaManagerLogReader(replicaManager);
+ CompletableFuture<FetchDataInfo> future =
logReader.readRemote(remoteStorageFetchInfo());
+
+ assertTrue(future.isCompletedExceptionally());
+ ExecutionException exception = assertThrows(ExecutionException.class,
() -> future.get(10, TimeUnit.SECONDS));
+ assertInstanceOf(IllegalStateException.class, exception.getCause());
+ }
+
+ @Test
+ public void testReadRemoteCompletesWithFetchedData() throws Exception {
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ RemoteLogManager remoteLogManager = mock(RemoteLogManager.class);
+
when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));
+
+ FetchDataInfo fetchDataInfo = new FetchDataInfo(new
LogOffsetMetadata(0L), MemoryRecords.EMPTY);
+ RemoteStorageFetchInfo fetchInfo = remoteStorageFetchInfo();
+ doAnswer(invocation -> {
+ Consumer<RemoteLogReadResult> callback = invocation.getArgument(1);
+ callback.accept(new
RemoteLogReadResult(Optional.of(fetchDataInfo), Optional.empty()));
+ return null;
+ }).when(remoteLogManager).asyncRead(eq(fetchInfo), any());
+
+ ReplicaManagerLogReader logReader = new
ReplicaManagerLogReader(replicaManager);
+ CompletableFuture<FetchDataInfo> future =
logReader.readRemote(fetchInfo);
+
+ assertSame(fetchDataInfo, future.get(10, TimeUnit.SECONDS));
+ }
+
+ @Test
+ public void testReadRemoteCompletesExceptionallyWhenReadResultHasError() {
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ RemoteLogManager remoteLogManager = mock(RemoteLogManager.class);
+
when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));
+
+ RuntimeException readError = new RuntimeException("remote read
failed");
+ doAnswer(invocation -> {
+ Consumer<RemoteLogReadResult> callback = invocation.getArgument(1);
+ callback.accept(new RemoteLogReadResult(Optional.empty(),
Optional.of(readError)));
+ return null;
+ }).when(remoteLogManager).asyncRead(any(), any());
+
+ ReplicaManagerLogReader logReader = new
ReplicaManagerLogReader(replicaManager);
+ CompletableFuture<FetchDataInfo> future =
logReader.readRemote(remoteStorageFetchInfo());
+
+ assertTrue(future.isCompletedExceptionally());
+ ExecutionException exception = assertThrows(ExecutionException.class,
() -> future.get(10, TimeUnit.SECONDS));
+ assertSame(readError, exception.getCause());
+ }
+
+ @Test
+ public void testReadRemoteCompletesExceptionallyWhenReadResultIsEmpty() {
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ RemoteLogManager remoteLogManager = mock(RemoteLogManager.class);
+
when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));
+
+ doAnswer(invocation -> {
+ Consumer<RemoteLogReadResult> callback = invocation.getArgument(1);
+ callback.accept(new RemoteLogReadResult(Optional.empty(),
Optional.empty()));
+ return null;
+ }).when(remoteLogManager).asyncRead(any(), any());
+
+ ReplicaManagerLogReader logReader = new
ReplicaManagerLogReader(replicaManager);
+ CompletableFuture<FetchDataInfo> future =
logReader.readRemote(remoteStorageFetchInfo());
+
+ assertTrue(future.isCompletedExceptionally());
+ ExecutionException exception = assertThrows(ExecutionException.class,
() -> future.get(10, TimeUnit.SECONDS));
+ assertInstanceOf(IllegalStateException.class, exception.getCause());
+ }
+
+ @Test
+ public void testReadRemoteCompletesExceptionallyWhenSchedulingRejected() {
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ RemoteLogManager remoteLogManager = mock(RemoteLogManager.class);
+
when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));
+
+ RejectedExecutionException rejected = new
RejectedExecutionException("reader pool shutting down");
+ doThrow(rejected).when(remoteLogManager).asyncRead(any(), any());
+
+ ReplicaManagerLogReader logReader = new
ReplicaManagerLogReader(replicaManager);
+ CompletableFuture<FetchDataInfo> future =
logReader.readRemote(remoteStorageFetchInfo());
+
+ assertTrue(future.isCompletedExceptionally());
+ ExecutionException exception = assertThrows(ExecutionException.class,
() -> future.get(10, TimeUnit.SECONDS));
+ assertSame(rejected, exception.getCause());
+ }
+
+ @Test
+ public void testReadAsyncReturnsEmptyWhenNoPartitionsToFetch() {
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+
+ ReplicaManagerLogReader logReader = new
ReplicaManagerLogReader(replicaManager);
+ LinkedHashMap<TopicIdPartition, LogReadResult> result =
+ logReader.readAsync(fetchParams(), Set.of(), new
LinkedHashMap<>(), new LinkedHashMap<>(), true).join();
+
+ assertTrue(result.isEmpty());
+ verify(replicaManager, never()).readFromLog(any(), any(), any(),
anyBoolean());
+ }
+
+ @Test
+ public void testReadAsyncReturnsLocalDataWhenNotTiered() throws Exception {
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ FetchDataInfo localData = localData();
+ stubReadFromLog(replicaManager,
+ List.of(new Tuple2<>(TOPIC_ID_PARTITION,
localReadResult(localData, Errors.NONE))));
+
+ ReplicaManagerLogReader logReader = new
ReplicaManagerLogReader(replicaManager);
+ LinkedHashMap<TopicIdPartition, LogReadResult> result =
+ logReader.readAsync(fetchParams(), Set.of(TOPIC_ID_PARTITION),
+ offsets(TOPIC_ID_PARTITION), maxBytes(TOPIC_ID_PARTITION),
true).get(10, TimeUnit.SECONDS);
+
+ assertEquals(Set.of(TOPIC_ID_PARTITION), result.keySet());
+ LogReadResult logReadResult = result.get(TOPIC_ID_PARTITION);
+ assertSame(localData, logReadResult.info());
+ assertEquals(Errors.NONE, logReadResult.error());
+ // Local data, so no remote read should have been attempted.
+ verify(replicaManager, never()).remoteLogManager();
+ }
+
+ @Test
+ public void testReadAsyncFollowsRemoteWhenTieredAndReadRemoteTrue() throws
Exception {
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ RemoteLogManager remoteLogManager = mock(RemoteLogManager.class);
+
when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));
+
+ RemoteStorageFetchInfo remoteStorageFetchInfo =
remoteStorageFetchInfo();
+ stubReadFromLog(replicaManager,
+ List.of(new Tuple2<>(TOPIC_ID_PARTITION,
localReadResult(tieredData(remoteStorageFetchInfo), Errors.NONE))));
+
+ FetchDataInfo remoteData = new FetchDataInfo(new
LogOffsetMetadata(0L), MemoryRecords.EMPTY);
+ doAnswer(invocation -> {
+ Consumer<RemoteLogReadResult> callback = invocation.getArgument(1);
+ callback.accept(new RemoteLogReadResult(Optional.of(remoteData),
Optional.empty()));
+ return null;
+ }).when(remoteLogManager).asyncRead(eq(remoteStorageFetchInfo), any());
+
+ ReplicaManagerLogReader logReader = new
ReplicaManagerLogReader(replicaManager);
+ LinkedHashMap<TopicIdPartition, LogReadResult> result =
+ logReader.readAsync(fetchParams(), Set.of(TOPIC_ID_PARTITION),
+ offsets(TOPIC_ID_PARTITION), maxBytes(TOPIC_ID_PARTITION),
true).get(10, TimeUnit.SECONDS);
+
+ LogReadResult logReadResult = result.get(TOPIC_ID_PARTITION);
+ assertSame(remoteData, logReadResult.info());
+ assertEquals(Errors.NONE, logReadResult.error());
+ verify(remoteLogManager).asyncRead(eq(remoteStorageFetchInfo), any());
+ }
+
+ @Test
+ public void testReadAsyncSkipsRemoteWhenReadRemoteFalse() throws Exception
{
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ FetchDataInfo tieredData = tieredData(remoteStorageFetchInfo());
+ stubReadFromLog(replicaManager,
+ List.of(new Tuple2<>(TOPIC_ID_PARTITION,
localReadResult(tieredData, Errors.NONE))));
+
+ ReplicaManagerLogReader logReader = new
ReplicaManagerLogReader(replicaManager);
+ LinkedHashMap<TopicIdPartition, LogReadResult> result =
+ logReader.readAsync(fetchParams(), Set.of(TOPIC_ID_PARTITION),
+ offsets(TOPIC_ID_PARTITION), maxBytes(TOPIC_ID_PARTITION),
false).get(10, TimeUnit.SECONDS);
+
+ LogReadResult logReadResult = result.get(TOPIC_ID_PARTITION);
+ // Tiered data is skipped (local read returned as-is), and the remote
tier is never consulted.
+ assertSame(tieredData, logReadResult.info());
+ assertEquals(Errors.NONE, logReadResult.error());
+ verify(replicaManager, never()).remoteLogManager();
+ }
+
+ @Test
+ public void testReadAsyncReturnsErrorFromLocalRead() throws Exception {
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ FetchDataInfo localData = localData();
+ stubReadFromLog(replicaManager,
+ List.of(new Tuple2<>(TOPIC_ID_PARTITION,
localReadResult(localData, Errors.UNKNOWN_SERVER_ERROR))));
+
+ ReplicaManagerLogReader logReader = new
ReplicaManagerLogReader(replicaManager);
+ LinkedHashMap<TopicIdPartition, LogReadResult> result =
+ logReader.readAsync(fetchParams(), Set.of(TOPIC_ID_PARTITION),
+ offsets(TOPIC_ID_PARTITION), maxBytes(TOPIC_ID_PARTITION),
true).get(10, TimeUnit.SECONDS);
+
+ LogReadResult logReadResult = result.get(TOPIC_ID_PARTITION);
+ assertSame(localData, logReadResult.info());
+ assertEquals(Errors.UNKNOWN_SERVER_ERROR, logReadResult.error());
+ verify(replicaManager, never()).remoteLogManager();
+ }
+
+ @Test
+ public void testReadAsyncReturnsErrorWhenRemoteReadFails() throws
Exception {
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ RemoteLogManager remoteLogManager = mock(RemoteLogManager.class);
+
when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));
+
+ FetchDataInfo tieredData = tieredData(remoteStorageFetchInfo());
+ stubReadFromLog(replicaManager,
+ List.of(new Tuple2<>(TOPIC_ID_PARTITION,
localReadResult(tieredData, Errors.NONE))));
+
+ doAnswer(invocation -> {
+ Consumer<RemoteLogReadResult> callback = invocation.getArgument(1);
+ callback.accept(new RemoteLogReadResult(Optional.empty(),
Optional.of(new RuntimeException("remote read failed"))));
+ return null;
+ }).when(remoteLogManager).asyncRead(any(), any());
+
+ ReplicaManagerLogReader logReader = new
ReplicaManagerLogReader(replicaManager);
+ LinkedHashMap<TopicIdPartition, LogReadResult> result =
+ logReader.readAsync(fetchParams(), Set.of(TOPIC_ID_PARTITION),
+ offsets(TOPIC_ID_PARTITION), maxBytes(TOPIC_ID_PARTITION),
true).get(10, TimeUnit.SECONDS);
+
+ LogReadResult logReadResult = result.get(TOPIC_ID_PARTITION);
+ // Partial-data tolerant: the read completes (does not throw) with the
local info and the error.
+ assertSame(tieredData, logReadResult.info());
+ assertEquals(Errors.UNKNOWN_SERVER_ERROR, logReadResult.error());
+ }
+
+ @Test
+ public void testReadAsyncResolvesPartitionsIndependently() throws
Exception {
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ RemoteLogManager remoteLogManager = mock(RemoteLogManager.class);
+
when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));
+
+ // Partition 1 is tiered (resolved from remote), partition 2 is
available locally.
+ RemoteStorageFetchInfo remoteStorageFetchInfo =
remoteStorageFetchInfo();
+ FetchDataInfo localData = localData();
+ stubReadFromLog(replicaManager, List.of(
+ new Tuple2<>(TOPIC_ID_PARTITION,
localReadResult(tieredData(remoteStorageFetchInfo), Errors.NONE)),
+ new Tuple2<>(TOPIC_ID_PARTITION_2, localReadResult(localData,
Errors.NONE))));
+
+ FetchDataInfo remoteData = new FetchDataInfo(new
LogOffsetMetadata(0L), MemoryRecords.EMPTY);
+ doAnswer(invocation -> {
+ Consumer<RemoteLogReadResult> callback = invocation.getArgument(1);
+ callback.accept(new RemoteLogReadResult(Optional.of(remoteData),
Optional.empty()));
+ return null;
+ }).when(remoteLogManager).asyncRead(eq(remoteStorageFetchInfo), any());
+
+ ReplicaManagerLogReader logReader = new
ReplicaManagerLogReader(replicaManager);
+ LinkedHashMap<TopicIdPartition, LogReadResult> result =
+ logReader.readAsync(fetchParams(), Set.of(TOPIC_ID_PARTITION,
TOPIC_ID_PARTITION_2),
+ offsets(TOPIC_ID_PARTITION, TOPIC_ID_PARTITION_2),
+ maxBytes(TOPIC_ID_PARTITION, TOPIC_ID_PARTITION_2),
true).get(10, TimeUnit.SECONDS);
+
+ assertEquals(Set.of(TOPIC_ID_PARTITION, TOPIC_ID_PARTITION_2),
result.keySet());
+ assertSame(remoteData, result.get(TOPIC_ID_PARTITION).info());
+ assertSame(localData, result.get(TOPIC_ID_PARTITION_2).info());
+ }
+}
diff --git a/server/src/main/java/org/apache/kafka/server/share/LogReader.java
b/server/src/main/java/org/apache/kafka/server/share/LogReader.java
index 60e453e9b86..3ae9500f9f4 100644
--- a/server/src/main/java/org/apache/kafka/server/share/LogReader.java
+++ b/server/src/main/java/org/apache/kafka/server/share/LogReader.java
@@ -22,6 +22,7 @@ import org.apache.kafka.storage.internals.log.LogReadResult;
import java.util.LinkedHashMap;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
/**
* Abstraction for reading records from log.
@@ -42,4 +43,35 @@ public interface LogReader {
Set<TopicIdPartition> partitionsToFetch,
LinkedHashMap<TopicIdPartition, Long> topicPartitionFetchOffsets,
LinkedHashMap<TopicIdPartition, Integer> partitionMaxBytes);
+
+ /**
+ * Read records for the given partitions starting at the specified
offsets, combining the local read
+ * and - when {@code readRemote} is true and the requested data has been
tiered off the local log - the
+ * follow-up remote read into a single call.
+ *
+ * <p>This is the asynchronous, remote-aware counterpart to {@link #read}:
it returns a single future
+ * holding one {@link LogReadResult} per requested partition. The future
completes once every partition
+ * has resolved - partitions available locally (or whose local read
failed) resolve immediately, while
+ * partitions whose data is in remote storage resolve later, once the
remote read finishes on the remote
+ * storage reader pool, so the caller's thread is never blocked on remote
IO. When {@code readRemote} is
+ * false, tiered offsets are simply omitted from the result rather than
fetched.
+ *
+ * <p>Each per-partition {@link LogReadResult} is partial-data tolerant:
the read never fails as a whole,
+ * allowing callers to use whatever records were retrieved (via {@link
LogReadResult#info()}) and skip
+ * the rest based on {@link LogReadResult#error()}.
+ *
+ * @param fetchParams The fetch parameters (isolation
level, maxBytes, etc.)
+ * @param partitionsToFetch The set of partitions to fetch
+ * @param topicPartitionFetchOffsets The fetch offset per partition
+ * @param partitionMaxBytes The max bytes per partition
+ * @param readRemote Whether to follow tiered offsets to
the remote tier; when false,
+ * tiered offsets are skipped.
+ * @return A future of a map from partition to that partition's {@link
LogReadResult}.
+ */
+ CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
readAsync(
+ FetchParams fetchParams,
+ Set<TopicIdPartition> partitionsToFetch,
+ LinkedHashMap<TopicIdPartition, Long> topicPartitionFetchOffsets,
+ LinkedHashMap<TopicIdPartition, Integer> partitionMaxBytes,
+ boolean readRemote);
}
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
new file mode 100644
index 00000000000..df6f381dec9
--- /dev/null
+++
b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcher.java
@@ -0,0 +1,226 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.server.share.dlq;
+
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.record.internal.Record;
+import org.apache.kafka.common.record.internal.RecordBatch;
+import org.apache.kafka.common.record.internal.Records;
+import org.apache.kafka.common.requests.FetchRequest;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.server.share.LogReader;
+import org.apache.kafka.server.storage.log.FetchIsolation;
+import org.apache.kafka.server.storage.log.FetchParams;
+import org.apache.kafka.storage.internals.log.LogReadResult;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Reads the original source records for the offset range described by a
{@link ShareGroupDLQRecordParameter}
+ * so they can be copied into a DLQ record. Reads are issued one batch at a
time in a loop via
+ * {@link LogReader#readAsync}, which combines the local read with the
follow-up remote read for any offset
+ * tiered off the local log. When a read is already complete the loop
continues in place; when it is still
+ * 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>Instances are single-use: create one fetcher per {@link #fetch()} call.
+ */
+public class ShareGroupDLQRecordFetcher {
+ private static final Logger log =
LoggerFactory.getLogger(ShareGroupDLQRecordFetcher.class);
+
+ private final LogReader logReader;
+ private final Time time;
+ private final ShareGroupDLQRecordParameter param;
+
+ private final TopicIdPartition tp;
+ private final long endOffset;
+ private final int recordCount;
+ private final long startTime;
+ private final Map<Long, Record> recordMap;
+ // 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 FetchParams fetchParams;
+
+ public ShareGroupDLQRecordFetcher(LogReader logReader, Time time,
ShareGroupDLQRecordParameter param, int maxFetchBytes) {
+ this.logReader = logReader;
+ this.time = time;
+ this.param = param;
+ this.tp = param.topicIdPartition();
+ this.endOffset = param.lastOffset();
+ this.recordCount = (int) (param.lastOffset() - param.firstOffset() +
1);
+ this.startTime = time.hiResClockMs();
+ this.recordMap = new HashMap<>(recordCount);
+ this.maxBytesMap.put(tp, maxFetchBytes);
+ this.fetchParams = new FetchParams(
+ FetchRequest.CONSUMER_REPLICA_ID, // -1, reading as a
consumer
+ -1, // replicaEpoch
+ 0L, // maxWaitMs - don't
block
+ 1, // minBytes
+ maxFetchBytes, // maxBytes
+ FetchIsolation.HIGH_WATERMARK, // committed only
+ Optional.empty() // clientMetadata
+ );
+ }
+
+ /**
+ * 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.
+ */
+ public CompletableFuture<Map<Long, Record>> fetch() {
+ try {
+ runFrom(param.firstOffset());
+ } catch (Exception e) {
+ // Never let an unexpected error escape; skip record copy entirely.
+ log.warn("Unexpected error fetching records for {}. Skipping
record copy.", param, e);
+ result.complete(Map.of());
+ }
+ return result;
+ }
+
+ /**
+ * Drives the reads in a loop via {@link LogReader#readAsync}. When the
per-offset read is already
+ * 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).
+ */
+ private void runFrom(long startFrom) {
+ long nextOffset = startFrom;
+ while (nextOffset <= endOffset) {
+ offsets.put(tp, nextOffset);
+
+ CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
future =
+ logReader.readAsync(fetchParams, Set.of(tp), offsets,
maxBytesMap, true);
+
+ if (!future.isDone()) {
+ // A remote read is in flight: resume from the callback so the
calling thread is unblocked.
+ long readFrom = nextOffset;
+ future.whenComplete((results, exception) -> resume(readFrom,
logReadResult(results), exception));
+ return;
+ }
+
+ // Safe (non-blocking) because the future is done. readAsync is
partial-data tolerant, so it
+ // 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
+ return;
+ }
+ nextOffset = advanced;
+ }
+ complete();
+ }
+
+ /**
+ * 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).
+ */
+ private LogReadResult logReadResult(LinkedHashMap<TopicIdPartition,
LogReadResult> results) {
+ return results == null ? null : results.get(tp);
+ }
+
+ /**
+ * Resumes the read loop after an asynchronous (remote) read completes.
Runs after runFrom() has already
+ * returned, so invoking runFrom() here does not grow the original call
stack.
+ */
+ private void resume(long readFrom, LogReadResult logReadResult, Throwable
exception) {
+ try {
+ if (exception != null) {
+ log.warn("Unable to read records at offset {} for {}. Skipping
it.", readFrom, param, exception);
+ complete();
+ return;
+ }
+ long advanced = collect(readFrom, logReadResult);
+ if (advanced <= readFrom) {
+ complete(); // no progress, stop
+ } else {
+ runFrom(advanced); // resume the loop
+ }
+ } catch (Exception e) {
+ log.warn("Unexpected error processing records for {}. Skipping
record copy.", param, e);
+ result.complete(Map.of());
+ }
+ }
+
+ /**
+ * Collects the records from a completed read into the map and returns the
offset to read from next.
+ * A read that failed (carries an error) or returned no usable data leaves
the offsets unread (skipped),
+ * which the loop treats as no progress.
+ */
+ private long collect(long readFrom, LogReadResult logReadResult) {
+ if (logReadResult == null) {
+ return readFrom;
+ }
+ if (logReadResult.error() != Errors.NONE) {
+ log.warn("Unable to read records at offset {} for {} due to error
{}. Skipping it.",
+ readFrom, param, logReadResult.error());
+ return readFrom;
+ }
+ return collectRecords(logReadResult.info().records, readFrom);
+ }
+
+ /**
+ * 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.
+ */
+ private long collectRecords(Records records, long readFrom) {
+ long nextOffset = readFrom;
+ for (RecordBatch batch : records.batches()) {
+ for (Record record : batch) {
+ // A fetch can return a batch whose base offset is below the
requested offset, so skip
+ // any record at or before the read position to avoid
re-processing and dragging
+ // nextOffset backwards.
+ if (record.offset() < readFrom) continue;
+ if (record.offset() > endOffset) return nextOffset;
+ recordMap.put(record.offset(), record);
+ nextOffset = Math.max(nextOffset, record.offset() + 1); //
never moves backwards
+ }
+ }
+ 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.
+ */
+ private void complete() {
+ log.trace("Log fetch took {} ms for {} records starting at {} for {}",
time.hiResClockMs() - startTime,
+ recordCount, param.firstOffset(), param);
+ if (recordCount != recordMap.size()) {
+ log.info("Total offsets requested: {}, Records found: {}",
recordCount, recordMap.size());
+ }
+ result.complete(Map.copyOf(recordMap));
+ }
+}
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 799346ebf8c..1a27dd7f0bc 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
@@ -37,12 +37,10 @@ import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.record.internal.MemoryRecords;
import org.apache.kafka.common.record.internal.Record;
-import org.apache.kafka.common.record.internal.RecordBatch;
import org.apache.kafka.common.record.internal.SimpleRecord;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.CreateTopicsRequest;
import org.apache.kafka.common.requests.CreateTopicsResponse;
-import org.apache.kafka.common.requests.FetchRequest;
import org.apache.kafka.common.requests.ProduceRequest;
import org.apache.kafka.common.requests.ProduceResponse;
import org.apache.kafka.common.utils.Time;
@@ -50,13 +48,10 @@ import
org.apache.kafka.common.utils.internals.ExponentialBackoffManager;
import org.apache.kafka.server.config.ServerConfigs;
import org.apache.kafka.server.share.LogReader;
import org.apache.kafka.server.share.metrics.ShareGroupMetrics;
-import org.apache.kafka.server.storage.log.FetchIsolation;
-import org.apache.kafka.server.storage.log.FetchParams;
import org.apache.kafka.server.util.InterBrokerSendThread;
import org.apache.kafka.server.util.RequestAndCompletionHandler;
import org.apache.kafka.server.util.timer.Timer;
import org.apache.kafka.server.util.timer.TimerTask;
-import org.apache.kafka.storage.internals.log.LogReadResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -190,7 +185,27 @@ public class ShareGroupDLQStateManager {
}
CompletableFuture<Void> future = new CompletableFuture<>();
ProduceRequestHandler requestHandler = new
ProduceRequestHandler(param, future, requestBackoffMs, requestBackoffMaxMs,
maxRequestAttempts);
- enqueue(requestHandler);
+
+ // Validate the DLQ configuration up front, synchronously on the
calling thread, so a
+ // misconfigured DLQ fails fast, and we never read source records for
one. enqueue() also
+ // re-validates so that retries re-check the (dynamic) config.
+ Optional<Throwable> validationError =
requestHandler.validateDlqTopic();
+ if (validationError.isPresent()) {
+ future.completeExceptionally(validationError.get());
+ return future;
+ }
+
+ // Resolve the source records once, 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
+ // configured (validated above); otherwise we enqueue immediately.
+ if (cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
+ requestHandler.resolveRecords().whenComplete((ignored,
ignoredError) -> enqueue(requestHandler));
+ } else {
+ enqueue(requestHandler);
+ }
return future;
}
@@ -221,9 +236,6 @@ public class ShareGroupDLQStateManager {
* @param handler The handler instance to add to the node map.
*/
private void addRequestToNodeMap(Node node, ProduceRequestHandler handler)
{
- if (!handler.isBatchable()) {
- return;
- }
synchronized (nodeMapLock) {
nodeRPCMap.computeIfAbsent(node, k -> new LinkedList<>())
.add(handler);
@@ -238,9 +250,18 @@ public class ShareGroupDLQStateManager {
private static final Logger LOG =
LoggerFactory.getLogger(ShareGroupDLQStateManager.ProduceRequestHandler.class);
private final ExponentialBackoffManager createTopicsBackoff;
private final ExponentialBackoffManager produceRequestBackoff;
- private Node dlqPartitionLeaderNode;
- private int dlqDestinationPartition;
- private ShareGroupDLQMetadataCacheHelper.TopicPartitionData
dlqTopicPartitionData;
+ // These DLQ topic fields are written by populateDLQTopicData() and
read while building the
+ // produce request - both on the sender thread (from
dlqTopicExists()/handleCreateTopicsResponse()).
+ // Kept volatile defensively.
+ 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.
+ private volatile Map<Long, Record> resolvedRecordData = Map.of();
public static final String HEADER_DLQ_ERRORS_TOPIC =
"__dlq.errors.topic";
public static final String HEADER_DLQ_ERRORS_PARTITION =
"__dlq.errors.partition";
@@ -296,19 +317,6 @@ public class ShareGroupDLQStateManager {
return "ProduceRequestHandler";
}
- /**
- * This method helps determine if the handler could
- * participate in batching (added to nodeMap). This will
- * be helpful if the RPCs which cannot be batched are included in
- * this class as well.
- *
- * @return Boolean indicating whether this handler can be coalesced
with others
- * to reduce number of RPCs sent.
- */
- boolean isBatchable() {
- return true;
- }
-
public void requestErrorResponse(Throwable exception) {
this.result.completeExceptionally(exception);
}
@@ -337,10 +345,6 @@ public class ShareGroupDLQStateManager {
.setTopics(topicCollection));
}
- public AbstractRequest.Builder<? extends AbstractRequest>
requestBuilder() {
- throw new RuntimeException("Produce requests are batchable, hence
individual requests not needed.");
- }
-
public void populateDLQTopicData() throws ConfigException {
Optional<String> dlqTopic =
cacheHelper.shareGroupDlqTopic(param.groupId());
if (dlqTopic.isEmpty()) {
@@ -372,7 +376,9 @@ public class ShareGroupDLQStateManager {
}
public ProduceRequestData.TopicProduceData topicProduceData() {
- Map<Long, Record> originalRecordData = maybeFetchRecordData();
+ // Records have already been resolved (including any remote
storage reads) before this
+ // handler was added to the node map, so no blocking fetch happens
on the sender thread here.
+ Map<Long, Record> originalRecordData = resolvedRecordData;
List<SimpleRecord> simpleRecords = new ArrayList<>();
for (long i = param.firstOffset(); i <= param.lastOffset(); i++) {
@@ -451,6 +457,7 @@ public class ShareGroupDLQStateManager {
} catch (ConfigException e) {
return false;
}
+ // Source records were already resolved before enqueue; just
add to the node map.
addRequestToNodeMap(dlqPartitionLeaderNode, this);
}
return isDlqTopicPresent;
@@ -545,11 +552,8 @@ public class ShareGroupDLQStateManager {
try {
populateDLQTopicData();
createTopicsBackoff.resetAttempts();
- if (this.isBatchable()) {
-
addRequestToNodeMap(this.dlqPartitionLeaderNode, this);
- } else {
- enqueue(this);
- }
+ // Source records were 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);
if (!createTopicsBackoff.canAttempt()) {
@@ -690,78 +694,37 @@ public class ShareGroupDLQStateManager {
}
}
- private Map<Long, Record> maybeFetchRecordData() {
- if
(!cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
- return Map.of();
- }
- long startTime = time.hiResClockMs();
- TopicIdPartition tp = param.topicIdPartition();
-
- FetchParams fetchParams = new FetchParams(
- FetchRequest.CONSUMER_REPLICA_ID, // -1, reading as
a consumer
- -1, // replicaEpoch
- 0L, // maxWaitMs -
don't block
- 1, // minBytes
- DLQ_MAX_FETCH_BYTES, // maxBytes
- FetchIsolation.HIGH_WATERMARK, // committed only
- Optional.empty() // clientMetadata
- );
-
- long nextOffset = param.firstOffset();
- long endOffset = param.lastOffset();
- int recordCount = (int) (param.lastOffset() - param.firstOffset()
+ 1);
-
- Map<Long, Record> recordMap = new HashMap<>(recordCount);
- LinkedHashMap<TopicIdPartition, Long> offsets = new
LinkedHashMap<>();
- LinkedHashMap<TopicIdPartition, Integer> maxBytesMap = new
LinkedHashMap<>();
- maxBytesMap.put(tp, DLQ_MAX_FETCH_BYTES);
-
- // 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 in offsets map within the while loop.
- while (nextOffset <= endOffset) {
- long readFrom = nextOffset; // offset requested for this
iteration
- offsets.put(tp, readFrom);
-
- LinkedHashMap<TopicIdPartition, LogReadResult> result =
- logReader.read(fetchParams, Set.of(tp), offsets,
maxBytesMap);
-
- LogReadResult res = result.get(param.topicIdPartition());
- if (res == null || res.error().code() != Errors.NONE.code()) {
- log.warn("Unable to fetch actual record at offset {} for
handler {}.", readFrom, this);
- return Map.of();
- }
-
- res.info().delayedRemoteStorageFetch.ifPresent(data ->
log.info(
- "Some offset data in is in remote storage. Skipping it."));
-
- for (RecordBatch batch : res.info().records.batches()) {
- for (Record record : batch) {
- // A fetch can return a batch whose base offset is
below the requested
- // offset, so skip any record at or before the read
position to avoid
- // re-processing and dragging nextOffset backwards.
- if (record.offset() < readFrom) continue;
- if (record.offset() > param.lastOffset()) {
- log.trace("Preempted log fetch took {} ms for {}
records starting at {} for {}", time.hiResClockMs() - startTime,
- recordCount, param.firstOffset(), this);
- return Map.copyOf(recordMap);
- }
- recordMap.put(record.offset(), record);
- nextOffset = Math.max(nextOffset, record.offset() +
1); // never moves backwards
- }
+ /**
+ * 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.
+ *
+ * <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.
+ *
+ * @return A future that always completes normally, once resolution
has finished.
+ */
+ CompletableFuture<Void> resolveRecords() {
+ 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;
}
+ resolved.complete(null);
+ });
+ return resolved;
+ }
- // If the read position did not advance this iteration we have
made no progress
- // (reached HWM/LEO or only stale records were returned). Bail
out to guarantee
- // termination rather than re-fetching the same offset forever.
- if (nextOffset <= readFrom) break;
+ private CompletableFuture<Map<Long, Record>> maybeFetchRecordData() {
+ if
(!cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
+ return CompletableFuture.completedFuture(Map.of());
}
- log.trace("Full log fetch took {} ms for {} records starting at {}
for {}", time.hiResClockMs() - startTime,
- recordCount, param.firstOffset(), this);
- log.info("Total offsets fetched: {}, Records found: {}",
recordCount, recordMap.size());
- return Map.copyOf(recordMap);
+ return new ShareGroupDLQRecordFetcher(logReader, time, param,
DLQ_MAX_FETCH_BYTES).fetch();
}
}
@@ -806,16 +769,9 @@ public class ShareGroupDLQStateManager {
log.error("Unable to create topic request for handler
{}.", handler, exp);
handler.requestErrorResponse(Errors.INVALID_CONFIG.exception());
}
- } else {
- if (!handler.isBatchable()) {
- requests.add(new RequestAndCompletionHandler(
- time.milliseconds(),
- handler.dlqPartitionLeaderNode(),
- handler.requestBuilder(),
- handler
- ));
- }
}
+ // When the DLQ topic already exists, the handler is added to
the node map for produce
+ // coalescing (asynchronously, once its records are resolved),
so nothing more to do here.
}
// {
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
new file mode 100644
index 00000000000..6779e9617cc
--- /dev/null
+++
b/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcherTest.java
@@ -0,0 +1,291 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.server.share.dlq;
+
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.compress.Compression;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.record.internal.MemoryRecords;
+import org.apache.kafka.common.record.internal.Record;
+import org.apache.kafka.common.record.internal.Records;
+import org.apache.kafka.common.record.internal.SimpleRecord;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.share.LogReader;
+import org.apache.kafka.server.util.MockTime;
+import org.apache.kafka.storage.internals.log.FetchDataInfo;
+import org.apache.kafka.storage.internals.log.LogReadResult;
+
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.OptionalLong;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anySet;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class ShareGroupDLQRecordFetcherTest {
+
+ private static final MockTime MOCK_TIME = new MockTime();
+ private static final String GROUP_ID = "test-group";
+ private static final TopicIdPartition TOPIC_ID_PARTITION =
+ new TopicIdPartition(Uuid.randomUuid(), 0, "source-topic");
+ private static final int MAX_FETCH_BYTES = 1024 * 1024;
+
+ private final LogReader logReader = mock(LogReader.class);
+
+ private static ShareGroupDLQRecordParameter param(long firstOffset, long
lastOffset) {
+ return new ShareGroupDLQRecordParameter(
+ GROUP_ID, TOPIC_ID_PARTITION, firstOffset, lastOffset,
Optional.empty(), Optional.empty());
+ }
+
+ private ShareGroupDLQRecordFetcher fetcher(ShareGroupDLQRecordParameter
param) {
+ return new ShareGroupDLQRecordFetcher(logReader, MOCK_TIME, param,
MAX_FETCH_BYTES);
+ }
+
+ private Map<Long, Record> fetch(ShareGroupDLQRecordParameter param) throws
Exception {
+ return fetcher(param).fetch().get(10, TimeUnit.SECONDS);
+ }
+
+ // ---- helpers ----
+
+ // 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);
+ }
+
+ // A successful read carrying the given records (offsets assigned from 0
by MemoryRecords#withRecords).
+ private static LogReadResult success(SimpleRecord... records) {
+ return logReadResult(
+ new FetchDataInfo(null,
MemoryRecords.withRecords(Compression.NONE, records)), Errors.NONE);
+ }
+
+ // A failed read - partial-data tolerant, so it still carries a (here
empty) FetchDataInfo.
+ private static LogReadResult failure(Errors error) {
+ return logReadResult(new FetchDataInfo(null, MemoryRecords.EMPTY),
error);
+ }
+
+ private static LinkedHashMap<TopicIdPartition, LogReadResult>
resultMap(LogReadResult result) {
+ LinkedHashMap<TopicIdPartition, LogReadResult> map = new
LinkedHashMap<>();
+ map.put(TOPIC_ID_PARTITION, result);
+ return map;
+ }
+
+ private static CompletableFuture<LinkedHashMap<TopicIdPartition,
LogReadResult>> done(LogReadResult result) {
+ return CompletableFuture.completedFuture(resultMap(result));
+ }
+
+ @SafeVarargs
+ private void
whenReadAsync(CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
first,
+
CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>... rest) {
+ var stub = when(logReader.readAsync(any(), anySet(), any(), any(),
anyBoolean())).thenReturn(first);
+ for (CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
future : rest) {
+ stub = stub.thenReturn(future);
+ }
+ }
+
+ private static SimpleRecord record(String key, String value) {
+ return new SimpleRecord(MOCK_TIME.milliseconds(),
+ key.getBytes(StandardCharsets.UTF_8),
value.getBytes(StandardCharsets.UTF_8));
+ }
+
+ private static void assertRecord(Map<Long, Record> result, long offset,
String key, String value) {
+ Record record = result.get(offset);
+ assertTrue(record != null, "Expected a record at offset " + offset);
+ assertArrayEquals(key.getBytes(StandardCharsets.UTF_8),
toArray(record.key()));
+ assertArrayEquals(value.getBytes(StandardCharsets.UTF_8),
toArray(record.value()));
+ }
+
+ private static byte[] toArray(ByteBuffer buffer) {
+ return Utils.toArray(buffer);
+ }
+
+ @Test
+ public void testFetchAllRecordsInSingleRead() throws Exception {
+ whenReadAsync(done(success(record("k0", "v0"), record("k1", "v1"),
record("k2", "v2"))));
+
+ Map<Long, Record> result = fetch(param(0L, 2L));
+
+ assertEquals(3, result.size());
+ assertRecord(result, 0L, "k0", "v0");
+ assertRecord(result, 1L, "k1", "v1");
+ assertRecord(result, 2L, "k2", "v2");
+ // The DLQ copy always asks readAsync to follow tiered offsets to the
remote tier.
+ verify(logReader).readAsync(any(), anySet(), any(), any(), eq(true));
+ }
+
+ @Test
+ public void testFetchRecordsAcrossMultipleReads() throws Exception {
+ // First read returns only the first two offsets; the next read
returns the batch containing the
+ // remaining offset (records at or before the already-read position
are skipped by the fetcher).
+ whenReadAsync(
+ done(success(record("k0", "v0"), record("k1", "v1"))),
+ done(success(record("k0", "v0"), record("k1", "v1"), record("k2",
"v2"))));
+
+ Map<Long, Record> result = fetch(param(0L, 2L));
+
+ assertEquals(3, result.size());
+ assertRecord(result, 0L, "k0", "v0");
+ assertRecord(result, 1L, "k1", "v1");
+ assertRecord(result, 2L, "k2", "v2");
+ verify(logReader, times(2)).readAsync(any(), anySet(), any(), any(),
eq(true));
+ }
+
+ @Test
+ public void testReadErrorYieldsNoRecords() throws Exception {
+ whenReadAsync(done(failure(Errors.UNKNOWN_SERVER_ERROR)));
+
+ Map<Long, Record> result = fetch(param(0L, 2L));
+
+ assertTrue(result.isEmpty());
+ }
+
+ @Test
+ public void testReadErrorMidRangeReturnsRecordsReadSoFar() throws
Exception {
+ // The first read succeeds and the second fails; the records already
collected are still returned.
+ whenReadAsync(
+ done(success(record("k0", "v0"), record("k1", "v1"))),
+ done(failure(Errors.UNKNOWN_SERVER_ERROR)));
+
+ Map<Long, Record> result = fetch(param(0L, 2L));
+
+ assertEquals(2, result.size());
+ assertRecord(result, 0L, "k0", "v0");
+ assertRecord(result, 1L, "k1", "v1");
+ assertNull(result.get(2L));
+ }
+
+ @Test
+ public void testMissingPartitionResultYieldsNoRecords() throws Exception {
+ // readAsync returns no entry for the partition - nothing can be read.
+ when(logReader.readAsync(any(), anySet(), any(), any(), anyBoolean()))
+ .thenReturn(CompletableFuture.completedFuture(new
LinkedHashMap<>()));
+
+ Map<Long, Record> result = fetch(param(0L, 2L));
+
+ assertTrue(result.isEmpty());
+ }
+
+ @Test
+ public void testNoProgressYieldsNoRecords() throws Exception {
+ // A read that returns no records does not advance the read position,
so the loop terminates.
+ whenReadAsync(done(success()));
+
+ Map<Long, Record> result = fetch(param(0L, 2L));
+
+ assertTrue(result.isEmpty());
+ }
+
+ @Test
+ public void testSingleOffsetFetch() throws Exception {
+ whenReadAsync(done(success(record("k0", "v0"))));
+
+ Map<Long, Record> result = fetch(param(0L, 0L));
+
+ assertEquals(1, result.size());
+ assertRecord(result, 0L, "k0", "v0");
+ }
+
+ @Test
+ public void testRecordsBeyondEndOffsetIgnored() throws Exception {
+ // The read returns more records than the requested range [0, 1];
offsets beyond endOffset are ignored.
+ whenReadAsync(done(success(record("k0", "v0"), record("k1", "v1"),
record("k2", "v2"))));
+
+ Map<Long, Record> result = fetch(param(0L, 1L));
+
+ assertEquals(2, result.size());
+ assertRecord(result, 0L, "k0", "v0");
+ assertRecord(result, 1L, "k1", "v1");
+ assertNull(result.get(2L));
+ }
+
+ @Test
+ public void testAsyncReadResumesLoopWhenPending() throws Exception {
+ // A not-yet-complete future (e.g. an in-flight remote read) suspends
the loop; the fetch returns
+ // before the read completes and resumes from the callback.
+ 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();
+ 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);
+ assertEquals(3, result.size());
+ assertRecord(result, 0L, "k0", "v0");
+ assertRecord(result, 1L, "k1", "v1");
+ assertRecord(result, 2L, "k2", "v2");
+ }
+
+ @Test
+ public void testAsyncReadResumeWithNoProgressCompletesEmpty() throws
Exception {
+ 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();
+ 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());
+ }
+
+ @Test
+ public void testAsyncReadResumeCompletesEmptyWhenProcessingThrows() throws
Exception {
+ 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();
+
+ // 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());
+ }
+
+ @Test
+ public void testFetchCompletesEmptyWhenReadAsyncThrows() throws Exception {
+ // An unexpected error from the log reader must not escape; the copy
is skipped entirely.
+ when(logReader.readAsync(any(), anySet(), any(), any(),
anyBoolean())).thenThrow(new RuntimeException("boom"));
+
+ Map<Long, Record> result = fetch(param(0L, 2L));
+
+ assertTrue(result.isEmpty());
+ }
+}
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 2fd2cf2b76b..a1b96b88660 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
@@ -64,6 +64,7 @@ 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.ExecutionException;
@@ -90,6 +91,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anySet;
import static org.mockito.ArgumentMatchers.anyString;
@@ -1571,6 +1573,33 @@ class ShareGroupDLQStateManagerTest {
// --- DLQ record with copy record enabled ---
+ private static FetchDataInfo recordsInfo(SimpleRecord... records) {
+ return new FetchDataInfo(null,
MemoryRecords.withRecords(Compression.NONE, records));
+ }
+
+ // 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(
+ TopicIdPartition topicIdPartition, LogReadResult result) {
+ LinkedHashMap<TopicIdPartition, LogReadResult> map = new
LinkedHashMap<>();
+ map.put(topicIdPartition, result);
+ return CompletableFuture.completedFuture(map);
+ }
+
+ // Stubs logReader.readAsync to return, in order, the given per-call
results for the partition,
+ // each wrapped in an already-complete future.
+ private static void whenReadAsync(LogReader logReader, TopicIdPartition
topicIdPartition,
+ LogReadResult first, LogReadResult...
rest) {
+ var stub = when(logReader.readAsync(any(), anySet(), any(), any(),
anyBoolean()))
+ .thenReturn(asyncReadMap(topicIdPartition, first));
+ for (LogReadResult result : rest) {
+ stub = stub.thenReturn(asyncReadMap(topicIdPartition, result));
+ }
+ }
+
@Test
public void testDLQRecordCopyEnabled() throws Exception {
MockClient client = new MockClient(MOCK_TIME);
@@ -1595,21 +1624,10 @@ class ShareGroupDLQStateManagerTest {
byte[] keyData3 = "key3".getBytes(StandardCharsets.UTF_8);
byte[] valueData3 = "value3".getBytes(StandardCharsets.UTF_8);
LogReader logReader = mock(LogReader.class);
- LogReadResult readResult = mock(LogReadResult.class);
- when(readResult.error()).thenReturn(Errors.NONE);
- when(readResult.info()).thenReturn(new FetchDataInfo(
- null,
- MemoryRecords.withRecords(
- Compression.NONE,
- new SimpleRecord(MOCK_TIME.milliseconds(), keyData1,
valueData1),
- new SimpleRecord(MOCK_TIME.milliseconds(), keyData2,
valueData2),
- new SimpleRecord(MOCK_TIME.milliseconds(), keyData3,
valueData3)
- )
- ));
- LinkedHashMap<TopicIdPartition, LogReadResult> readResultMap = new
LinkedHashMap<>();
- readResultMap.put(param.topicIdPartition(), readResult);
- when(logReader.read(any(), anySet(), any(), any()))
- .thenReturn(readResultMap);
+ whenReadAsync(logReader, param.topicIdPartition(),
logReadResult(recordsInfo(
+ new SimpleRecord(MOCK_TIME.milliseconds(), keyData1, valueData1),
+ new SimpleRecord(MOCK_TIME.milliseconds(), keyData2, valueData2),
+ new SimpleRecord(MOCK_TIME.milliseconds(), keyData3, valueData3)),
Errors.NONE));
ShareGroupDLQMetadataCacheHelper cacheHelper =
cacheHelper(DEFAULT_LEADER);
when(cacheHelper.isShareGroupDlqCopyRecordEnabled(any())).thenReturn(true);
@@ -1632,6 +1650,91 @@ class ShareGroupDLQStateManagerTest {
verify(mockMetrics, never()).recordDLQProduceFailed(any());
}
+ @Test
+ public void testDLQRecordCopyEnabledFetchesOnceAcrossProduceRetries()
throws Exception {
+ int maxAttempts = 3;
+ // Real timer with tiny backoffs lets the produce retries fire within
milliseconds.
+ Timer realTimer = new SystemTimerReaper("shareGroupDLQTestTimer",
+ new SystemTimer("shareGroupDLQTestTimer"));
+ try {
+ // param() spans offsets 0..2, so return all three records in a
single read; one
+ // resolution then maps to exactly one logReader.readAsync() call.
+ ShareGroupDLQRecordParameter param = param();
+ LogReader logReader = mock(LogReader.class);
+ whenReadAsync(logReader, param.topicIdPartition(),
logReadResult(recordsInfo(
+ new SimpleRecord(MOCK_TIME.milliseconds(),
"key0".getBytes(StandardCharsets.UTF_8),
"value0".getBytes(StandardCharsets.UTF_8)),
+ new SimpleRecord(MOCK_TIME.milliseconds(),
"key1".getBytes(StandardCharsets.UTF_8),
"value1".getBytes(StandardCharsets.UTF_8)),
+ new SimpleRecord(MOCK_TIME.milliseconds(),
"key2".getBytes(StandardCharsets.UTF_8),
"value2".getBytes(StandardCharsets.UTF_8))), Errors.NONE));
+
+ ShareGroupDLQMetadataCacheHelper cacheHelper =
cacheHelper(DEFAULT_LEADER);
+
when(cacheHelper.isShareGroupDlqCopyRecordEnabled(any())).thenReturn(true);
+
+ MockClient client = new MockClient(MOCK_TIME);
+ // Two produce disconnects (retriable), then a successful produce.
+ client.prepareResponseFrom(body -> body instanceof ProduceRequest,
null, DEFAULT_LEADER, true);
+ client.prepareResponseFrom(body -> body instanceof ProduceRequest,
null, DEFAULT_LEADER, true);
+ 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(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.
+ verify(logReader, times(1)).readAsync(any(), anySet(), any(),
any(), anyBoolean());
+ verify(mockMetrics, times(maxAttempts)).recordDLQProduce(GROUP_ID);
+ verify(mockMetrics).recordDLQRecordWrite(GROUP_ID, 3);
+ verify(mockMetrics, never()).recordDLQProduceFailed(any());
+ } finally {
+ Utils.closeQuietly(realTimer, "shareGroupDLQTestTimer");
+ }
+ }
+
+ @Test
+ public void testDLQRecordCopyEnabledButInvalidConfigSkipsFetch() throws
Exception {
+ LogReader logReader = mock(LogReader.class);
+ ShareGroupDLQMetadataCacheHelper cacheHelper =
mock(ShareGroupDLQMetadataCacheHelper.class);
+ // Misconfigured DLQ (no topic configured) - validation must fail
before any record copy.
+
when(cacheHelper.shareGroupDlqTopic(GROUP_ID)).thenReturn(Optional.empty());
+
when(cacheHelper.shareGroupDlqTopicPrefix()).thenReturn(Optional.empty());
+
when(cacheHelper.isShareGroupDlqCopyRecordEnabled(any())).thenReturn(true);
+
+ stateManager =
builder().withLogReader(logReader).withCacheHelper(cacheHelper).build();
+ stateManager.start();
+ Throwable cause = getCause(stateManager.dlq(param()));
+ assertInstanceOf(ConfigException.class, cause);
+ // Even though record copy is enabled, an invalid DLQ config fails
fast and the source log
+ // is never read.
+ verifyNoInteractions(logReader);
+ verifyNoInteractions(mockMetrics);
+ }
+
+ @Test
+ public void testDLQRecordCopyDisabledSkipsFetch() throws Exception {
+ MockClient client = new MockClient(MOCK_TIME);
+ client.prepareResponseFrom(
+ body -> body instanceof ProduceRequest,
+ successfulProduceResponse(0),
+ DEFAULT_LEADER
+ );
+ LogReader logReader = mock(LogReader.class);
+
+ // Default cacheHelper has record copy disabled.
+ stateManager =
builder().withClient(client).withLogReader(logReader).build();
+ stateManager.start();
+ assertNull(stateManager.dlq(param()).get(10, TimeUnit.SECONDS));
+
+ // Copy disabled => the source log is never read; the DLQ record
carries headers only.
+ verifyNoInteractions(logReader);
+ verify(mockMetrics).recordDLQRecordWrite(GROUP_ID, 3);
+ verify(mockMetrics, never()).recordDLQProduceFailed(any());
+ }
+
@Test
public void testDLQRecordCopyEnabledWithPartialLogReaderRecords() throws
Exception {
MockClient client = new MockClient(MOCK_TIME);
@@ -1656,37 +1759,15 @@ class ShareGroupDLQStateManagerTest {
byte[] keyData3 = "key3".getBytes(StandardCharsets.UTF_8);
byte[] valueData3 = "value3".getBytes(StandardCharsets.UTF_8);
LogReader logReader = mock(LogReader.class);
- LogReadResult readResult1 = mock(LogReadResult.class);
- when(readResult1.error()).thenReturn(Errors.NONE);
- // Return 2 records only
- when(readResult1.info()).thenReturn(new FetchDataInfo(
- null,
- MemoryRecords.withRecords(
- Compression.NONE,
+ // First read returns 2 records; the next read contains the 3rd offset
as well.
+ whenReadAsync(logReader, param.topicIdPartition(),
+ logReadResult(recordsInfo(
new SimpleRecord(MOCK_TIME.milliseconds(), keyData1,
valueData1),
- new SimpleRecord(MOCK_TIME.milliseconds(), keyData2,
valueData2)
- )
- ));
- LogReadResult readResult2 = mock(LogReadResult.class);
- when(readResult2.error()).thenReturn(Errors.NONE);
- // Next read contains the 3rd offset as well
- when(readResult2.info()).thenReturn(new FetchDataInfo(
- null,
- MemoryRecords.withRecords(
- Compression.NONE,
+ new SimpleRecord(MOCK_TIME.milliseconds(), keyData2,
valueData2)), Errors.NONE),
+ logReadResult(recordsInfo(
new SimpleRecord(MOCK_TIME.milliseconds(), keyData1,
valueData1),
new SimpleRecord(MOCK_TIME.milliseconds(), keyData2,
valueData2),
- new SimpleRecord(MOCK_TIME.milliseconds(), keyData3,
valueData3)
- )
- ));
-
- LinkedHashMap<TopicIdPartition, LogReadResult> readResultMap1 = new
LinkedHashMap<>();
- LinkedHashMap<TopicIdPartition, LogReadResult> readResultMap2 = new
LinkedHashMap<>();
- readResultMap1.put(param.topicIdPartition(), readResult1);
- readResultMap2.put(param.topicIdPartition(), readResult2);
- when(logReader.read(any(), anySet(), any(), any()))
- .thenReturn(readResultMap1)
- .thenReturn(readResultMap2);
+ new SimpleRecord(MOCK_TIME.milliseconds(), keyData3,
valueData3)), Errors.NONE));
ShareGroupDLQMetadataCacheHelper cacheHelper =
cacheHelper(DEFAULT_LEADER);
when(cacheHelper.isShareGroupDlqCopyRecordEnabled(any())).thenReturn(true);
@@ -1731,21 +1812,11 @@ class ShareGroupDLQStateManagerTest {
byte[] keyData2 = "key2".getBytes(StandardCharsets.UTF_8);
byte[] valueData2 = "value2".getBytes(StandardCharsets.UTF_8);
LogReader logReader = mock(LogReader.class);
- LogReadResult readResult = mock(LogReadResult.class);
- when(readResult.error()).thenReturn(Errors.NONE);
- // Return 2 records only
- when(readResult.info()).thenReturn(new FetchDataInfo(
- null,
- MemoryRecords.withRecords(
- Compression.NONE,
- new SimpleRecord(MOCK_TIME.milliseconds(), keyData1,
valueData1),
- new SimpleRecord(MOCK_TIME.milliseconds(), keyData2,
valueData2)
- )
- ));
- LinkedHashMap<TopicIdPartition, LogReadResult> readResultMap = new
LinkedHashMap<>();
- readResultMap.put(param.topicIdPartition(), readResult);
- when(logReader.read(any(), anySet(), any(), any()))
- .thenReturn(readResultMap);
+ // Every read returns only 2 records (offsets 0 and 1); the 3rd offset
is never read, so the
+ // loop makes no further progress and the record is produced with
headers only.
+ whenReadAsync(logReader, param.topicIdPartition(),
logReadResult(recordsInfo(
+ new SimpleRecord(MOCK_TIME.milliseconds(), keyData1, valueData1),
+ new SimpleRecord(MOCK_TIME.milliseconds(), keyData2, valueData2)),
Errors.NONE));
ShareGroupDLQMetadataCacheHelper cacheHelper =
cacheHelper(DEFAULT_LEADER);
when(cacheHelper.isShareGroupDlqCopyRecordEnabled(any())).thenReturn(true);
@@ -1786,12 +1857,143 @@ class ShareGroupDLQStateManagerTest {
ShareGroupDLQRecordParameter param = param();
LogReader logReader = mock(LogReader.class);
- LogReadResult readResult = mock(LogReadResult.class);
- when(readResult.error()).thenReturn(Errors.UNKNOWN_SERVER_ERROR);
- LinkedHashMap<TopicIdPartition, LogReadResult> readResultMap = new
LinkedHashMap<>();
- readResultMap.put(param.topicIdPartition(), readResult);
- when(logReader.read(any(), anySet(), any(), any()))
- .thenReturn(readResultMap);
+ // The read fails; record copy is skipped and the DLQ record is
produced with headers only.
+ whenReadAsync(logReader, param.topicIdPartition(),
+ logReadResult(recordsInfo(), Errors.UNKNOWN_SERVER_ERROR));
+
+ ShareGroupDLQMetadataCacheHelper cacheHelper =
cacheHelper(DEFAULT_LEADER);
+
when(cacheHelper.isShareGroupDlqCopyRecordEnabled(any())).thenReturn(true);
+ stateManager =
builder().withClient(client).withLogReader(logReader).withCacheHelper(cacheHelper).build();
+ stateManager.start();
+ assertNull(stateManager.dlq(param).get(10, TimeUnit.SECONDS));
+
+ assertEquals(1, capturedProduces.size());
+ assertDlqProduceRecordHeaders(capturedProduces.get(0), Map.of(
+ 0, new ExpectedDlqPartition(0L, 2L, 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"
+ ), List.of(), List.of())
+ ));
+ verify(mockMetrics).recordDLQProduce(GROUP_ID);
+ verify(mockMetrics).recordDLQRecordWrite(GROUP_ID, 3);
+ verify(mockMetrics, never()).recordDLQProduceFailed(any());
+ }
+
+ @Test
+ public void testDLQRecordCopyEnabledWithRemoteFetch() throws Exception {
+ MockClient client = new MockClient(MOCK_TIME);
+ List<ProduceRequest> capturedProduces = new ArrayList<>();
+ client.prepareResponseFrom(
+ captureProduce(capturedProduces),
+ successfulProduceResponse(0),
+ DEFAULT_LEADER
+ );
+
+ ShareGroupDLQRecordParameter param = param(); // 3 offsets requested
+ byte[] keyData1 = "key1".getBytes(StandardCharsets.UTF_8);
+ byte[] valueData1 = "value1".getBytes(StandardCharsets.UTF_8);
+ byte[] keyData2 = "key2".getBytes(StandardCharsets.UTF_8);
+ byte[] valueData2 = "value2".getBytes(StandardCharsets.UTF_8);
+ byte[] keyData3 = "key3".getBytes(StandardCharsets.UTF_8);
+ byte[] valueData3 = "value3".getBytes(StandardCharsets.UTF_8);
+
+ LogReader logReader = mock(LogReader.class);
+ // readAsync resolves the (possibly tiered) records and returns them
in a single read.
+ whenReadAsync(logReader, param.topicIdPartition(),
logReadResult(recordsInfo(
+ new SimpleRecord(MOCK_TIME.milliseconds(), keyData1, valueData1),
+ new SimpleRecord(MOCK_TIME.milliseconds(), keyData2, valueData2),
+ new SimpleRecord(MOCK_TIME.milliseconds(), keyData3, valueData3)),
Errors.NONE));
+
+ ShareGroupDLQMetadataCacheHelper cacheHelper =
cacheHelper(DEFAULT_LEADER);
+
when(cacheHelper.isShareGroupDlqCopyRecordEnabled(any())).thenReturn(true);
+ stateManager =
builder().withClient(client).withLogReader(logReader).withCacheHelper(cacheHelper).build();
+ stateManager.start();
+ assertNull(stateManager.dlq(param).get(10, TimeUnit.SECONDS));
+
+ assertEquals(1, capturedProduces.size());
+ assertDlqProduceRecordHeaders(capturedProduces.get(0), Map.of(
+ 0, new ExpectedDlqPartition(0L, 2L, 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"
+ ), List.of(keyData1, keyData2, keyData3), List.of(valueData1,
valueData2, valueData3))
+ ));
+ verify(mockMetrics).recordDLQProduce(GROUP_ID);
+ verify(mockMetrics).recordDLQRecordWrite(GROUP_ID, 3);
+ verify(mockMetrics, never()).recordDLQProduceFailed(any());
+ }
+
+ @Test
+ public void testDLQRecordCopyEnabledWithMixedLocalAndRemoteFetch() throws
Exception {
+ MockClient client = new MockClient(MOCK_TIME);
+ List<ProduceRequest> capturedProduces = new ArrayList<>();
+ client.prepareResponseFrom(
+ captureProduce(capturedProduces),
+ successfulProduceResponse(0),
+ DEFAULT_LEADER
+ );
+
+ ShareGroupDLQRecordParameter param = param(); // 3 offsets requested
(0, 1, 2)
+ byte[] keyData1 = "key1".getBytes(StandardCharsets.UTF_8);
+ byte[] valueData1 = "value1".getBytes(StandardCharsets.UTF_8);
+ byte[] keyData2 = "key2".getBytes(StandardCharsets.UTF_8);
+ byte[] valueData2 = "value2".getBytes(StandardCharsets.UTF_8);
+ byte[] keyData3 = "key3".getBytes(StandardCharsets.UTF_8);
+ byte[] valueData3 = "value3".getBytes(StandardCharsets.UTF_8);
+
+ LogReader logReader = mock(LogReader.class);
+ // First read returns offset 0; the next read returns the batch
covering the remaining offsets
+ // (records at or before the already-read position are skipped by the
fetcher).
+ whenReadAsync(logReader, param.topicIdPartition(),
+ logReadResult(recordsInfo(
+ new SimpleRecord(MOCK_TIME.milliseconds(), keyData1,
valueData1)), Errors.NONE),
+ logReadResult(recordsInfo(
+ new SimpleRecord(MOCK_TIME.milliseconds(), keyData1,
valueData1),
+ new SimpleRecord(MOCK_TIME.milliseconds(), keyData2,
valueData2),
+ new SimpleRecord(MOCK_TIME.milliseconds(), keyData3,
valueData3)), Errors.NONE));
+
+ ShareGroupDLQMetadataCacheHelper cacheHelper =
cacheHelper(DEFAULT_LEADER);
+
when(cacheHelper.isShareGroupDlqCopyRecordEnabled(any())).thenReturn(true);
+ stateManager =
builder().withClient(client).withLogReader(logReader).withCacheHelper(cacheHelper).build();
+ stateManager.start();
+ assertNull(stateManager.dlq(param).get(10, TimeUnit.SECONDS));
+
+ assertEquals(1, capturedProduces.size());
+ assertDlqProduceRecordHeaders(capturedProduces.get(0), Map.of(
+ 0, new ExpectedDlqPartition(0L, 2L, 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"
+ ), List.of(keyData1, keyData2, keyData3), List.of(valueData1,
valueData2, valueData3))
+ ));
+ verify(mockMetrics).recordDLQProduce(GROUP_ID);
+ verify(mockMetrics).recordDLQRecordWrite(GROUP_ID, 3);
+ verify(mockMetrics, never()).recordDLQProduceFailed(any());
+ }
+
+ @Test
+ public void testDLQRecordCopyEnabledWithRemoteFetchFailureSkipsRecords()
throws Exception {
+ MockClient client = new MockClient(MOCK_TIME);
+ List<ProduceRequest> capturedProduces = new ArrayList<>();
+ client.prepareResponseFrom(
+ captureProduce(capturedProduces),
+ successfulProduceResponse(0),
+ DEFAULT_LEADER
+ );
+
+ ShareGroupDLQRecordParameter param = param();
+ LogReader logReader = mock(LogReader.class);
+ // readAsync cannot read the (tiered) data; the offsets are gracefully
skipped and the DLQ
+ // record is produced with headers only.
+ whenReadAsync(logReader, param.topicIdPartition(),
+ logReadResult(recordsInfo(), Errors.UNKNOWN_SERVER_ERROR));
ShareGroupDLQMetadataCacheHelper cacheHelper =
cacheHelper(DEFAULT_LEADER);
when(cacheHelper.isShareGroupDlqCopyRecordEnabled(any())).thenReturn(true);
@@ -1814,6 +2016,16 @@ class ShareGroupDLQStateManagerTest {
verify(mockMetrics, never()).recordDLQProduceFailed(any());
}
+ private static MockClient.RequestMatcher
captureProduce(List<ProduceRequest> capturedProduces) {
+ return body -> {
+ if (body instanceof ProduceRequest pr) {
+ capturedProduces.add(pr);
+ return true;
+ }
+ return false;
+ };
+ }
+
private static ShareGroupDLQStateManager.ProduceRequestHandler
newHandlerForCoalesceTest(
ShareGroupDLQStateManager manager,
String groupId,
