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 81447c182a0 KAFKA-20678: Add timeout in replica manager log reader
remote read. (#22707)
81447c182a0 is described below
commit 81447c182a02667e40467e652effa50ad833d6f9
Author: Sushant Mahajan <[email protected]>
AuthorDate: Fri Jul 3 01:59:33 2026 +0530
KAFKA-20678: Add timeout in replica manager log reader remote read. (#22707)
ReplicaManagerLogReader.readAsync fetches data from remote storage, if
enabled. The response is async. Current code does not bound this call to
remote storage. This PR uses the existing `remoteFetchMaxWaitMs` to
alleviate the situation by scheduling a timed killer task into the
replica manager's timer wheel.
A new integ test has been added to ShareConsumerDLQTest to verify e-2-e
tiering based DLQ.
Reviewers: Apoorv Mittal <[email protected]>, Abhinav Dixit
<[email protected]>
---
build.gradle | 1 +
.../clients/consumer/ShareConsumerDLQTest.java | 175 ++++++++++++++++++++-
.../server/share/ReplicaManagerLogReader.java | 68 +++++---
.../server/share/ReplicaManagerLogReaderTest.java | 70 +++++++--
.../share/dlq/ShareGroupDLQRecordFetcher.java | 10 +-
5 files changed, 281 insertions(+), 43 deletions(-)
diff --git a/build.gradle b/build.gradle
index d4b484f0dfb..1267c5600e1 100644
--- a/build.gradle
+++ b/build.gradle
@@ -2182,6 +2182,7 @@ project(':clients:clients-integration-tests') {
testImplementation project(':raft')
testImplementation project(':server')
testImplementation project(':storage')
+ testImplementation testFixtures(project(':storage'))
testImplementation project(':core').sourceSets.test.output
testImplementation testFixtures(project(':clients'))
implementation project(':server-common')
diff --git
a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerDLQTest.java
b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerDLQTest.java
index b3f7eb09afb..1cbf5d16e02 100644
---
a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerDLQTest.java
+++
b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerDLQTest.java
@@ -20,6 +20,7 @@ import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.Config;
import org.apache.kafka.clients.admin.ConfigEntry;
import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.admin.OffsetSpec;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
@@ -267,6 +268,134 @@ public class ShareConsumerDLQTest extends
ShareConsumerTestBase {
verifyDlqMetrics(groupId, recordCount);
}
+ /**
+ * End-to-end DLQ copy when the source records have been tiered to remote
storage. The source topic enables
+ * tiered storage and rolls a segment per record, with a 45s total
retention and a 5s local retention so the
+ * early offsets are offloaded to remote storage and then deleted locally
(well before the remote segments
+ * expire). Once the local segments are gone (verified via the
earliest-local offset advancing past them), the
+ * records are rejected with record copy enabled - so the DLQ record
fetcher must read the original records
+ * back from remote storage. The resulting DLQ records carrying the
original key/value confirm the fetcher
+ * successfully pulled them from remote storage.
+ *
+ * <p>Tiered storage is backed by the local-filesystem {@code
LocalTieredStorage} RSM and the default
+ * {@code TopicBasedRemoteLogMetadataManager}; short task/cleanup
intervals keep the offload + local-delete
+ * cycle quick.
+ */
+ @ClusterTest(
+ serverProperties = {
+ @ClusterConfigProperty(key = "remote.log.storage.system.enable",
value = "true"),
+ @ClusterConfigProperty(key =
"remote.log.storage.manager.class.name",
+ value =
"org.apache.kafka.server.log.remote.storage.LocalTieredStorage"),
+ @ClusterConfigProperty(key =
"remote.log.manager.task.interval.ms", value = "500"),
+ @ClusterConfigProperty(key =
"remote.log.metadata.manager.listener.name", value = "EXTERNAL"),
+ @ClusterConfigProperty(key =
"rlmm.config.remote.log.metadata.topic.replication.factor", value = "1"),
+ @ClusterConfigProperty(key =
"rlmm.config.remote.log.metadata.topic.num.partitions", value = "1"),
+ @ClusterConfigProperty(key = "log.retention.check.interval.ms",
value = "500"),
+ @ClusterConfigProperty(key = "log.initial.task.delay.ms", value =
"100")
+ }
+ )
+ public void testDlqCopiesRecordsReadFromRemoteStorage() throws Exception {
+ String groupId = "dlq-remote-group";
+ // The broker's default share-group DLQ topic prefix is "dlq.", so the
topic name must start with it.
+ String dlqTopic = "dlq.remote";
+ String sourceTopic = "dlq-remote-source";
+ int recordCount = 5;
+
+ alterShareAutoOffsetReset(groupId, "earliest");
+ createDlqTopic(dlqTopic);
+ alterShareGroupConfig(groupId,
GroupConfig.ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG, dlqTopic);
+ // Record copy enabled: the DLQ fetcher must read the original records
back to copy their key/value.
+ alterShareGroupConfig(groupId,
GroupConfig.ERRORS_DEADLETTERQUEUE_COPY_RECORD_ENABLE_CONFIG, "true");
+
+ // Tiered source topic: one segment per record, 45s total retention
and 5s local retention so inactive
+ // segments are offloaded to remote storage and then deleted locally
shortly after, while the remote
+ // segments comfortably survive the rest of the test.
+ createRemoteStorageSourceTopic(sourceTopic, 45_000L, 5_000L);
+
+ produceTo(sourceTopic, 0, recordCount);
+
+ // Wait until the early offsets have been offloaded to remote storage
AND removed locally - i.e. the
+ // earliest *local* offset has advanced past them, so reading those
offsets must now hit remote storage.
+ // The last record stays in the active (never-offloaded) segment, so
the earliest local offset should
+ // reach recordCount - 1. A generous timeout (well inside the 45s
remote retention) absorbs remote-log
+ // metadata-manager startup; it normally resolves a few seconds after
the 5s local retention elapses.
+ waitForCondition(() -> earliestLocalOffset(sourceTopic, 0) >=
recordCount - 1,
+ 30_000L, 500L,
+ () -> "Source records were not tiered to remote storage and
removed locally in time");
+
+ // Reject every record. Both the share fetch (to deliver them) and the
DLQ record fetcher (to copy them)
+ // must read the tiered offsets back from remote storage.
+ rejectRecords(groupId, sourceTopic, recordCount);
+
+ // Record copy is enabled, so every DLQ record must carry the original
key/value. For the tiered offsets
+ // (no longer present locally) that is only possible if the DLQ
fetcher pulled them from remote storage.
+ verifyDlqTopicRecords(dlqTopic, groupId, sourceTopic, 0,
expectedSourceOffsets(recordCount), true);
+ verifyDlqMetrics(groupId, recordCount);
+ }
+
+ @ClusterTest(
+ serverProperties = {
+ @ClusterConfigProperty(key = "remote.log.storage.system.enable",
value = "true"),
+ @ClusterConfigProperty(key =
"remote.log.storage.manager.class.name",
+ value =
"org.apache.kafka.server.log.remote.storage.LocalTieredStorage"),
+ @ClusterConfigProperty(key =
"remote.log.manager.task.interval.ms", value = "500"),
+ @ClusterConfigProperty(key =
"remote.log.metadata.manager.listener.name", value = "EXTERNAL"),
+ @ClusterConfigProperty(key =
"rlmm.config.remote.log.metadata.topic.replication.factor", value = "1"),
+ @ClusterConfigProperty(key =
"rlmm.config.remote.log.metadata.topic.num.partitions", value = "1"),
+ @ClusterConfigProperty(key = "log.retention.check.interval.ms",
value = "500"),
+ @ClusterConfigProperty(key = "log.initial.task.delay.ms", value =
"100"),
+ @ClusterConfigProperty(key =
"group.share.min.heartbeat.interval.ms", value = "1500"),
+ @ClusterConfigProperty(key = "group.share.heartbeat.interval.ms",
value = "1500")
+ }
+ )
+ public void testDlqCopiesRecordsReadFromRemoteAndLocalStorage() throws
Exception {
+ String groupId = "dlq-remote-and-local-group";
+ // The broker's default share-group DLQ topic prefix is "dlq.", so the
topic name must start with it.
+ String dlqTopic = "dlq.remote-and-local-topic";
+ String sourceTopic = "dlq-remote-and-local-source";
+ int recordCount = 5;
+
+ alterShareAutoOffsetReset(groupId, "earliest");
+ createDlqTopic(dlqTopic);
+ alterShareGroupConfig(groupId,
GroupConfig.ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG, dlqTopic);
+ // Record copy enabled: the DLQ fetcher must read the original records
back to copy their key/value.
+ alterShareGroupConfig(groupId,
GroupConfig.ERRORS_DEADLETTERQUEUE_COPY_RECORD_ENABLE_CONFIG, "true");
+
+ // Tiered source topic: one segment per record, 45s total retention
and 5s local retention so inactive
+ // segments are offloaded to remote storage and then deleted locally
shortly after, while the remote
+ // segments comfortably survive the rest of the test.
+ createRemoteStorageSourceTopic(sourceTopic, 45_000L, 10_000L);
+
+ produceTo(sourceTopic, 0, recordCount);
+
+ // Wait until the early offsets have been offloaded to remote storage
AND removed locally - i.e. the
+ // earliest *local* offset has advanced past them, so reading those
offsets must now hit remote storage.
+ // The last record stays in the active (never-offloaded) segment, so
the earliest local offset should
+ // reach recordCount - 1. A generous timeout (well inside the 45s
remote retention) absorbs remote-log
+ // metadata-manager startup; it normally resolves a few seconds after
the 5s local retention elapses.
+ waitForCondition(() -> earliestLocalOffset(sourceTopic, 0) >=
recordCount - 1,
+ 30_000L, 500L,
+ () -> "Source records were not tiered to remote storage and
removed locally in time");
+
+ // Produce some more which stay in local.
+ produceTo(sourceTopic, 0, recordCount);
+
+ // Reject every record. Both the share fetch (to deliver them) and the
DLQ record fetcher (to copy them)
+ // must read the tiered offsets back from remote storage.
+ rejectRecords(groupId, sourceTopic, recordCount * 2);
+
+ // Record copy is enabled, so every DLQ record must carry the original
key/value. For the tiered offsets
+ // (no longer present locally) that is only possible if the DLQ
fetcher pulled them from remote storage.
+ verifyDlqTopicRecords(dlqTopic, groupId, sourceTopic, 0,
expectedSourceOffsets(recordCount * 2), true);
+
+ // Make sure not all offsets from second produce are tiered.
+ waitForCondition(() -> earliestLocalOffset(sourceTopic, 0) <
recordCount * 2 - 1,
+ 30_000L, 500L,
+ () -> "Offsets from second produce were tiered");
+
+ verifyDlqMetrics(groupId, recordCount * 2);
+ }
+
/**
* Rejects records from a multi-partition source topic and verifies they
are routed to the correct DLQ
* partition. The destination partition is {@code sourcePartition %
numDlqPartitions}; with a DLQ topic that
@@ -537,13 +666,23 @@ public class ShareConsumerDLQTest extends
ShareConsumerTestBase {
*/
private void verifyDlqTopicRecords(String dlqTopic, String groupId,
Set<Long> expectedSourceOffsets,
boolean copyEnabled) throws
InterruptedException {
+ verifyDlqTopicRecords(dlqTopic, groupId, tp.topic(), tp.partition(),
expectedSourceOffsets, copyEnabled);
+ }
+
+ /**
+ * As {@link #verifyDlqTopicRecords(String, String, Set, boolean)}, but
for an explicit source topic-partition
+ * (rather than the base {@code tp}). The DLQ records' context headers
must reference {@code sourceTopic} /
+ * {@code sourcePartition}.
+ */
+ private void verifyDlqTopicRecords(String dlqTopic, String groupId, String
sourceTopic, int sourcePartition,
+ Set<Long> expectedSourceOffsets,
boolean copyEnabled) throws InterruptedException {
List<ConsumerRecord<byte[], byte[]>> dlqRecords =
readDlqPartition(dlqTopic, 0, expectedSourceOffsets.size());
assertEquals(expectedSourceOffsets.size(), dlqRecords.size(),
"Unexpected number of records on the DLQ topic");
Set<Long> actualSourceOffsets = new HashSet<>();
for (ConsumerRecord<byte[], byte[]> record : dlqRecords) {
if (copyEnabled) {
- // produceMessages() produces records with key "key" and value
"value".
+ // produceMessages()/produceTo() produce records with key
"key" and value "value".
assertEquals("key", new
String(Objects.requireNonNull(record.key()), StandardCharsets.UTF_8),
"DLQ record key should be copied when record copy is
enabled");
assertEquals("value", new
String(Objects.requireNonNull(record.value()), StandardCharsets.UTF_8),
@@ -553,8 +692,8 @@ public class ShareConsumerDLQTest extends
ShareConsumerTestBase {
assertNull(record.value(), "DLQ record value should be null
when record copy is disabled");
}
assertEquals(groupId, headerValue(record,
HEADER_DLQ_ERRORS_GROUP));
- assertEquals(tp.topic(), headerValue(record,
HEADER_DLQ_ERRORS_TOPIC));
- assertEquals(Integer.toString(tp.partition()), headerValue(record,
HEADER_DLQ_ERRORS_PARTITION));
+ assertEquals(sourceTopic, headerValue(record,
HEADER_DLQ_ERRORS_TOPIC));
+ assertEquals(Integer.toString(sourcePartition),
headerValue(record, HEADER_DLQ_ERRORS_PARTITION));
actualSourceOffsets.add(Long.parseLong(Objects.requireNonNull(headerValue(record,
HEADER_DLQ_ERRORS_OFFSET))));
}
assertEquals(expectedSourceOffsets, actualSourceOffsets, "DLQ records
should cover every expected source offset");
@@ -591,6 +730,36 @@ public class ShareConsumerDLQTest extends
ShareConsumerTestBase {
}, "Failed to create DLQ topic");
}
+ // Creates a single-partition source topic with tiered storage enabled and
one log segment per record (via
+ // per-record index entries). A short local retention (`localRetentionMs`)
deletes inactive segments from
+ // local storage soon after they are offloaded to remote storage, so
reading those offsets must hit remote
+ // storage; the total retention (`retentionMs`) is kept generous so the
remote segments are not deleted while
+ // the test is still running.
+ private void createRemoteStorageSourceTopic(String topic, long
retentionMs, long localRetentionMs) {
+ assertDoesNotThrow(() -> {
+ try (Admin admin = createAdminClient()) {
+ Map<String, String> configs = Map.of(
+ TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true",
+ TopicConfig.RETENTION_MS_CONFIG,
Long.toString(retentionMs),
+ TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG,
Long.toString(localRetentionMs),
+ // Roll a segment for every record so each inactive
segment can be offloaded then deleted locally.
+ TopicConfig.INDEX_INTERVAL_BYTES_CONFIG, "1",
+ TopicConfig.SEGMENT_INDEX_BYTES_CONFIG, "12");
+ admin.createTopics(Set.of(new NewTopic(topic, 1, (short)
1).configs(configs))).all().get();
+ }
+ }, "Failed to create remote-storage source topic");
+ }
+
+ // The earliest offset still held in local storage. Offsets below this
have been removed locally (e.g. after
+ // being offloaded to remote storage), so reading them must hit remote
storage. Used to confirm tiering.
+ private long earliestLocalOffset(String topic, int partition) throws
Exception {
+ TopicPartition topicPartition = new TopicPartition(topic, partition);
+ try (Admin admin = createAdminClient()) {
+ return admin.listOffsets(Map.of(topicPartition,
OffsetSpec.earliestLocal()))
+ .partitionResult(topicPartition).get().offset();
+ }
+ }
+
// Produces `count` records (key "key", value "value") to a specific
topic-partition.
private void produceTo(String topic, int partition, int count) {
try (Producer<byte[], byte[]> producer = createProducer()) {
diff --git a/core/src/main/java/kafka/server/share/ReplicaManagerLogReader.java
b/core/src/main/java/kafka/server/share/ReplicaManagerLogReader.java
index acebdb6a735..70e798b6eb2 100644
--- a/core/src/main/java/kafka/server/share/ReplicaManagerLogReader.java
+++ b/core/src/main/java/kafka/server/share/ReplicaManagerLogReader.java
@@ -25,6 +25,7 @@ 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.server.util.timer.TimerTask;
import org.apache.kafka.storage.internals.log.FetchDataInfo;
import org.apache.kafka.storage.internals.log.LogReadResult;
import org.apache.kafka.storage.internals.log.RemoteStorageFetchInfo;
@@ -37,6 +38,7 @@ import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
+import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import scala.Tuple2;
@@ -116,16 +118,10 @@ public class ReplicaManagerLogReader implements LogReader
{
LinkedHashMap<TopicIdPartition, LogReadResult> localReadResults =
read(fetchParams, partitionsToFetch, topicPartitionFetchOffsets,
partitionMaxBytes);
- // One future per partition; combined into a single future once every
partition has resolved.
+ // Only look at partitions with non-null read results.
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;
@@ -144,10 +140,8 @@ public class ReplicaManagerLogReader implements LogReader {
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);
+ // Sending an error here to caller is not useful so not
setting it in the result. Above log should suffice.
+ return logReadResult;
}
return withInfoAndError(logReadResult, remoteFetchDataInfo,
Errors.NONE);
}));
@@ -185,18 +179,19 @@ public class ReplicaManagerLogReader implements LogReader
{
* 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).
+ * configured, the read could not be completed, or the read does not
finish within the configured
+ * remote fetch timeout ({@code remote.fetch.max.wait.ms}). Used
internally by readAsync
+ * (package-private so it remains unit-testable).
*/
// Visibility for testing
CompletableFuture<FetchDataInfo> readRemote(RemoteStorageFetchInfo
remoteStorageFetchInfo) {
- CompletableFuture<FetchDataInfo> future = new CompletableFuture<>();
+ CompletableFuture<FetchDataInfo> remoteFuture = new
CompletableFuture<>();
Optional<RemoteLogManager> remoteLogManager =
OptionConverters.toJava(replicaManager.remoteLogManager());
if (remoteLogManager.isEmpty()) {
- future.completeExceptionally(new IllegalStateException(
+ remoteFuture.completeExceptionally(new IllegalStateException(
"Cannot read " + remoteStorageFetchInfo + " from remote
storage as remote log manager is not configured."));
- return future;
+ return remoteFuture;
}
try {
@@ -204,20 +199,51 @@ public class ReplicaManagerLogReader implements LogReader
{
// 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());
+ remoteFuture.completeExceptionally(result.error().get());
} else if (result.fetchDataInfo().isPresent()) {
- future.complete(result.fetchDataInfo().get());
+ remoteFuture.complete(result.fetchDataInfo().get());
} else {
- future.completeExceptionally(new IllegalStateException(
+ remoteFuture.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);
+ remoteFuture.completeExceptionally(e);
}
- return future;
+ // Bound the wait on the remote read so a stalled remote tier cannot
pin the read (and the
+ // resources held while it is pending) indefinitely. Use the broker's
timer wheel - as
+ // DelayedShareFetch does for its remote fetch - rather than
CompletableFuture#orTimeout. On
+ // expiry the future completes exceptionally with a TimeoutException,
which the caller treats
+ // as a (skippable) read error.
+ if (!remoteFuture.isDone()) {
+ long timeoutMs = remoteFetchMaxWaitMs();
+ TimerTask timeoutTask = new TimerTask(timeoutMs) {
+ @Override
+ public void run() {
+ remoteFuture.completeExceptionally(new TimeoutException(
+ "Remote read for " + remoteStorageFetchInfo + " did
not complete within " + timeoutMs + " ms."));
+ }
+ };
+ // Cancel the timer task once the read completes (either outcome)
so it does not linger in the wheel.
+ remoteFuture.whenComplete((info, exception) ->
timeoutTask.cancel());
+
+ // ReplicaManager has a dedicated timer for consumption by the
share fetch
+ // code, and it is exposed addShareFetchTimerRequest. The same is
also used
+ // in DelayedShareFetch. Hence, re-using it.
+ replicaManager.addShareFetchTimerRequest(timeoutTask);
+ }
+
+ return remoteFuture;
+ }
+
+ /**
+ * The maximum time to wait for a remote-tier read, taken from the broker's
+ * {@code remote.fetch.max.wait.ms}. Read per call since the config is
dynamically reconfigurable.
+ */
+ private long remoteFetchMaxWaitMs() {
+ return
replicaManager.config().remoteLogManagerConfig().remoteFetchMaxWaitMs();
}
}
diff --git
a/core/src/test/java/kafka/server/share/ReplicaManagerLogReaderTest.java
b/core/src/test/java/kafka/server/share/ReplicaManagerLogReaderTest.java
index f0f75b329f2..fb93cba205b 100644
--- a/core/src/test/java/kafka/server/share/ReplicaManagerLogReaderTest.java
+++ b/core/src/test/java/kafka/server/share/ReplicaManagerLogReaderTest.java
@@ -16,16 +16,20 @@
*/
package kafka.server.share;
+import kafka.server.KafkaConfig;
import kafka.server.ReplicaManager;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.config.AbstractConfig;
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.log.remote.storage.RemoteLogManagerConfig;
import org.apache.kafka.server.storage.log.FetchIsolation;
import org.apache.kafka.server.storage.log.FetchParams;
+import org.apache.kafka.server.util.timer.TimerTask;
import org.apache.kafka.storage.internals.log.FetchDataInfo;
import org.apache.kafka.storage.internals.log.LogOffsetMetadata;
import org.apache.kafka.storage.internals.log.LogReadResult;
@@ -36,12 +40,14 @@ import org.junit.jupiter.api.Test;
import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
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.concurrent.TimeoutException;
import java.util.function.Consumer;
import scala.Option;
@@ -70,6 +76,18 @@ public class ReplicaManagerLogReaderTest {
new TopicIdPartition(Uuid.randomUuid(), 0, "topic");
private static final TopicIdPartition TOPIC_ID_PARTITION_2 =
new TopicIdPartition(Uuid.randomUuid(), 1, "topic");
+ // A real config with defaults; readRemote() reads
remote.fetch.max.wait.ms from it for its timeout.
+ private static final RemoteLogManagerConfig REMOTE_LOG_MANAGER_CONFIG =
+ new RemoteLogManagerConfig(new
AbstractConfig(RemoteLogManagerConfig.configDef(), Map.of(), false));
+
+ // A ReplicaManager mock whose config() chain resolves
remote.fetch.max.wait.ms, as readRemote() needs.
+ private static ReplicaManager mockReplicaManager() {
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ KafkaConfig kafkaConfig = mock(KafkaConfig.class);
+
when(kafkaConfig.remoteLogManagerConfig()).thenReturn(REMOTE_LOG_MANAGER_CONFIG);
+ when(replicaManager.config()).thenReturn(kafkaConfig);
+ return replicaManager;
+ }
private static RemoteStorageFetchInfo remoteStorageFetchInfo() {
return new RemoteStorageFetchInfo(
@@ -124,7 +142,7 @@ public class ReplicaManagerLogReaderTest {
@Test
public void testReadReturnsEmptyWhenNoPartitionsToFetch() {
- ReplicaManager replicaManager = mock(ReplicaManager.class);
+ ReplicaManager replicaManager = mockReplicaManager();
ReplicaManagerLogReader logReader = new
ReplicaManagerLogReader(replicaManager);
LinkedHashMap<TopicIdPartition, LogReadResult> result =
@@ -136,7 +154,7 @@ public class ReplicaManagerLogReaderTest {
@Test
public void testReadReturnsResultsFromReplicaManager() {
- ReplicaManager replicaManager = mock(ReplicaManager.class);
+ ReplicaManager replicaManager = mockReplicaManager();
LogReadResult logReadResult = mock(LogReadResult.class);
Seq<Tuple2<TopicIdPartition, LogReadResult>> readFromLogResult =
CollectionConverters.asScala(List.of(new
Tuple2<>(TOPIC_ID_PARTITION, logReadResult))).toSeq();
@@ -157,7 +175,7 @@ public class ReplicaManagerLogReaderTest {
@Test
public void
testReadRemoteCompletesExceptionallyWhenRemoteLogManagerNotConfigured() {
- ReplicaManager replicaManager = mock(ReplicaManager.class);
+ ReplicaManager replicaManager = mockReplicaManager();
when(replicaManager.remoteLogManager()).thenReturn(Option.empty());
ReplicaManagerLogReader logReader = new
ReplicaManagerLogReader(replicaManager);
@@ -170,7 +188,7 @@ public class ReplicaManagerLogReaderTest {
@Test
public void testReadRemoteCompletesWithFetchedData() throws Exception {
- ReplicaManager replicaManager = mock(ReplicaManager.class);
+ ReplicaManager replicaManager = mockReplicaManager();
RemoteLogManager remoteLogManager = mock(RemoteLogManager.class);
when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));
@@ -190,7 +208,7 @@ public class ReplicaManagerLogReaderTest {
@Test
public void testReadRemoteCompletesExceptionallyWhenReadResultHasError() {
- ReplicaManager replicaManager = mock(ReplicaManager.class);
+ ReplicaManager replicaManager = mockReplicaManager();
RemoteLogManager remoteLogManager = mock(RemoteLogManager.class);
when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));
@@ -211,7 +229,7 @@ public class ReplicaManagerLogReaderTest {
@Test
public void testReadRemoteCompletesExceptionallyWhenReadResultIsEmpty() {
- ReplicaManager replicaManager = mock(ReplicaManager.class);
+ ReplicaManager replicaManager = mockReplicaManager();
RemoteLogManager remoteLogManager = mock(RemoteLogManager.class);
when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));
@@ -231,7 +249,7 @@ public class ReplicaManagerLogReaderTest {
@Test
public void testReadRemoteCompletesExceptionallyWhenSchedulingRejected() {
- ReplicaManager replicaManager = mock(ReplicaManager.class);
+ ReplicaManager replicaManager = mockReplicaManager();
RemoteLogManager remoteLogManager = mock(RemoteLogManager.class);
when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));
@@ -246,9 +264,29 @@ public class ReplicaManagerLogReaderTest {
assertSame(rejected, exception.getCause());
}
+ @Test
+ public void testReadRemoteTimesOutWhenRemoteReadNeverCompletes() {
+ ReplicaManager replicaManager = mockReplicaManager();
+ RemoteLogManager remoteLogManager = mock(RemoteLogManager.class);
+
when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));
+ // asyncRead never invokes the callback, so the remote read never
completes on its own.
+ // Simulate the timer wheel firing by running the scheduled timeout
task immediately.
+ doAnswer(invocation -> {
+ invocation.<TimerTask>getArgument(0).run();
+ return null;
+ }).when(replicaManager).addShareFetchTimerRequest(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(TimeoutException.class, exception.getCause());
+ }
+
@Test
public void testReadAsyncReturnsEmptyWhenNoPartitionsToFetch() {
- ReplicaManager replicaManager = mock(ReplicaManager.class);
+ ReplicaManager replicaManager = mockReplicaManager();
ReplicaManagerLogReader logReader = new
ReplicaManagerLogReader(replicaManager);
LinkedHashMap<TopicIdPartition, LogReadResult> result =
@@ -260,7 +298,7 @@ public class ReplicaManagerLogReaderTest {
@Test
public void testReadAsyncReturnsLocalDataWhenNotTiered() throws Exception {
- ReplicaManager replicaManager = mock(ReplicaManager.class);
+ ReplicaManager replicaManager = mockReplicaManager();
FetchDataInfo localData = localData();
stubReadFromLog(replicaManager,
List.of(new Tuple2<>(TOPIC_ID_PARTITION,
localReadResult(localData, Errors.NONE))));
@@ -280,7 +318,7 @@ public class ReplicaManagerLogReaderTest {
@Test
public void testReadAsyncFollowsRemoteWhenTieredAndReadRemoteTrue() throws
Exception {
- ReplicaManager replicaManager = mock(ReplicaManager.class);
+ ReplicaManager replicaManager = mockReplicaManager();
RemoteLogManager remoteLogManager = mock(RemoteLogManager.class);
when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));
@@ -308,7 +346,7 @@ public class ReplicaManagerLogReaderTest {
@Test
public void testReadAsyncSkipsRemoteWhenReadRemoteFalse() throws Exception
{
- ReplicaManager replicaManager = mock(ReplicaManager.class);
+ ReplicaManager replicaManager = mockReplicaManager();
FetchDataInfo tieredData = tieredData(remoteStorageFetchInfo());
stubReadFromLog(replicaManager,
List.of(new Tuple2<>(TOPIC_ID_PARTITION,
localReadResult(tieredData, Errors.NONE))));
@@ -327,7 +365,7 @@ public class ReplicaManagerLogReaderTest {
@Test
public void testReadAsyncReturnsErrorFromLocalRead() throws Exception {
- ReplicaManager replicaManager = mock(ReplicaManager.class);
+ ReplicaManager replicaManager = mockReplicaManager();
FetchDataInfo localData = localData();
stubReadFromLog(replicaManager,
List.of(new Tuple2<>(TOPIC_ID_PARTITION,
localReadResult(localData, Errors.UNKNOWN_SERVER_ERROR))));
@@ -344,8 +382,8 @@ public class ReplicaManagerLogReaderTest {
}
@Test
- public void testReadAsyncReturnsErrorWhenRemoteReadFails() throws
Exception {
- ReplicaManager replicaManager = mock(ReplicaManager.class);
+ public void
testReadAsyncReturnsNoneErrorWhenRemoteReadFailsButLocalSucceeds() throws
Exception {
+ ReplicaManager replicaManager = mockReplicaManager();
RemoteLogManager remoteLogManager = mock(RemoteLogManager.class);
when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));
@@ -367,12 +405,12 @@ public class ReplicaManagerLogReaderTest {
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());
+ assertEquals(Errors.NONE, logReadResult.error());
}
@Test
public void testReadAsyncResolvesPartitionsIndependently() throws
Exception {
- ReplicaManager replicaManager = mock(ReplicaManager.class);
+ ReplicaManager replicaManager = mockReplicaManager();
RemoteLogManager remoteLogManager = mock(RemoteLogManager.class);
when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));
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 df6f381dec9..4d832423b02 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
@@ -126,8 +126,12 @@ public class ShareGroupDLQRecordFetcher {
if (!future.isDone()) {
// A remote read is in flight: resume from the callback so the
calling thread is unblocked.
+ // Uses whenCompleteAsync (not whenComplete) to close a race
between the isDone() check
+ // above and callback registration: if the future completes in
that gap, whenComplete
+ // would run the callback inline on this thread, letting
resume() recurse into runFrom()
+ // on the same stack. whenCompleteAsync always dispatches
off-thread, so that can't happen.
long readFrom = nextOffset;
- future.whenComplete((results, exception) -> resume(readFrom,
logReadResult(results), exception));
+ future.whenCompleteAsync((results, exception) ->
resume(readFrom, logReadResult(results), exception));
return;
}
@@ -169,8 +173,8 @@ public class ShareGroupDLQRecordFetcher {
runFrom(advanced); // resume the loop
}
} catch (Exception e) {
- log.warn("Unexpected error processing records for {}. Skipping
record copy.", param, e);
- result.complete(Map.of());
+ log.warn("Unexpected error processing records for {}. Returning
records fetched so far.", param, e);
+ complete();
}
}
