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 a7f9e519f25 KAFKA-20613: Add code to support copy record in DLQ state
manager. (#22479)
a7f9e519f25 is described below
commit a7f9e519f25a4dabf4512723d2d5f290c63e2b3d
Author: Sushant Mahajan <[email protected]>
AuthorDate: Mon Jun 15 16:14:25 2026 +0530
KAFKA-20613: Add code to support copy record in DLQ state manager. (#22479)
* In this PR, we have added code to read original record key/value data
from the offset being DLQ'ed. The data is read via the `LogReader`
interface.
* New tests have been added to check the functionality.
Reviewers: Apoorv Mittal <[email protected]>
---
.../src/main/scala/kafka/server/BrokerServer.scala | 12 +-
.../share/dlq/DefaultShareGroupDLQManager.java | 11 +-
.../share/dlq/ShareGroupDLQStateManager.java | 111 ++++++-
.../share/dlq/ShareGroupDLQStateManagerTest.java | 354 +++++++++++++++++++--
4 files changed, 460 insertions(+), 28 deletions(-)
diff --git a/core/src/main/scala/kafka/server/BrokerServer.scala
b/core/src/main/scala/kafka/server/BrokerServer.scala
index 8467b285f33..dce48fa521d 100644
--- a/core/src/main/scala/kafka/server/BrokerServer.scala
+++ b/core/src/main/scala/kafka/server/BrokerServer.scala
@@ -23,7 +23,7 @@ import kafka.log.LogManager
import kafka.network.SocketServer
import kafka.raft.KafkaRaftManager
import kafka.server.metadata._
-import kafka.server.share.{ReplicaManagerPartitionMetadataProvider,
ReplicaManagerLogReader, ShareCoordinatorMetadataCacheHelperImpl,
SharePartitionManager}
+import kafka.server.share.{ReplicaManagerLogReader,
ReplicaManagerPartitionMetadataProvider,
ShareCoordinatorMetadataCacheHelperImpl, SharePartitionManager}
import org.apache.kafka.common.config.ConfigException
import org.apache.kafka.common.internals.Plugin
import org.apache.kafka.common.message.ApiMessageType.ListenerType
@@ -177,6 +177,8 @@ class BrokerServer(
private var shareGroupDLQManager: ShareGroupDLQManager = _
+ private var shareGroupLogReader: ReplicaManagerLogReader = _
+
private def maybeChangeStatus(from: ProcessStatus, to: ProcessStatus):
Boolean = {
lock.lock()
try {
@@ -396,6 +398,9 @@ class BrokerServer(
/* create metrics object to be shared with share DLQ manager share
partition manager*/
shareGroupMetrics = new ShareGroupMetrics(time)
+ /* create log reader object to share with share group DLQ manager and
SharePartitionManager */
+ shareGroupLogReader = new ReplicaManagerLogReader(replicaManager)
+
/* create share group DLQ manager */
shareGroupDLQManager = createShareGroupDLQManager()
@@ -467,7 +472,7 @@ class BrokerServer(
sharePartitionManager = new SharePartitionManager(
replicaManager,
- new ReplicaManagerLogReader(replicaManager),
+ shareGroupLogReader,
new ReplicaManagerPartitionMetadataProvider(replicaManager),
(key: DelayedShareFetchKey) =>
replicaManager.completeDelayedShareFetchRequest(key),
time,
@@ -777,7 +782,8 @@ class BrokerServer(
new ShareCoordinatorMetadataCacheHelperImpl(metadataCache, key =>
shareCoordinator.partitionFor(key), config.interBrokerListenerName,
groupConfigManager),
Time.SYSTEM,
shareGroupTimer,
- shareGroupMetrics
+ shareGroupMetrics,
+ shareGroupLogReader
)
} else if
(klass.getName.equals(classOf[NoOpShareGroupDLQManager].getName)) {
info("Using no-op share group DLQ manager")
diff --git
a/server/src/main/java/org/apache/kafka/server/share/dlq/DefaultShareGroupDLQManager.java
b/server/src/main/java/org/apache/kafka/server/share/dlq/DefaultShareGroupDLQManager.java
index 32e242b0ed1..73559b3cebf 100644
---
a/server/src/main/java/org/apache/kafka/server/share/dlq/DefaultShareGroupDLQManager.java
+++
b/server/src/main/java/org/apache/kafka/server/share/dlq/DefaultShareGroupDLQManager.java
@@ -19,6 +19,7 @@ package org.apache.kafka.server.share.dlq;
import org.apache.kafka.clients.KafkaClient;
import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.server.share.LogReader;
import org.apache.kafka.server.share.metrics.ShareGroupMetrics;
import org.apache.kafka.server.util.timer.Timer;
@@ -45,9 +46,10 @@ public class DefaultShareGroupDLQManager implements
ShareGroupDLQManager {
ShareGroupDLQMetadataCacheHelper cacheHelper,
Time time,
Timer timer,
- ShareGroupMetrics metrics
+ ShareGroupMetrics shareGroupMetrics,
+ LogReader logReader
) {
- DefaultShareGroupDLQManager instance = new
DefaultShareGroupDLQManager(client, cacheHelper, time, timer, metrics);
+ DefaultShareGroupDLQManager instance = new
DefaultShareGroupDLQManager(client, cacheHelper, time, timer,
shareGroupMetrics, logReader);
instance.start();
return instance;
}
@@ -57,9 +59,10 @@ public class DefaultShareGroupDLQManager implements
ShareGroupDLQManager {
ShareGroupDLQMetadataCacheHelper cacheHelper,
Time time,
Timer timer,
- ShareGroupMetrics shareGroupMetrics
+ ShareGroupMetrics shareGroupMetrics,
+ LogReader logReader
) {
- this.stateManager = new ShareGroupDLQStateManager(client, cacheHelper,
time, timer, shareGroupMetrics);
+ this.stateManager = new ShareGroupDLQStateManager(client, cacheHelper,
time, timer, shareGroupMetrics, logReader);
}
private void start() {
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 b54878d7187..6bd1863aca8 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
@@ -36,30 +36,39 @@ import org.apache.kafka.common.message.ProduceResponseData;
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;
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;
+import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
@@ -82,12 +91,21 @@ public class ShareGroupDLQStateManager {
private final Time time;
private final Timer timer;
private final ShareGroupDLQMetadataCacheHelper cacheHelper;
+ private final LogReader logReader;
private final ShareGroupMetrics shareGroupMetrics;
public static final long REQUEST_BACKOFF_MS = 1_000L;
public static final long REQUEST_BACKOFF_MAX_MS = 30_000L;
private static final int MAX_REQUEST_ATTEMPTS = 5;
private static final int RETRY_BACKOFF_EXP_BASE =
CommonClientConfigs.RETRY_BACKOFF_EXP_BASE;
private static final double RETRY_BACKOFF_JITTER =
CommonClientConfigs.RETRY_BACKOFF_JITTER;
+
+ /**
+ * In most cases we expect the records getting DLQ'ed will be single
offsets and
+ * not complete batches. Hence, using a large upper limit while reading
from the log
+ * would be fruitless in most cases. Therefore, the value of 1 MB has been
chosen
+ * for the DLQ related log reads.
+ */
+ private static final int DLQ_MAX_FETCH_BYTES = 1024 * 1024;
private static final Logger log =
LoggerFactory.getLogger(ShareGroupDLQStateManager.class);
private final Set<Node> inFlight = new HashSet<>();
@@ -99,7 +117,8 @@ public class ShareGroupDLQStateManager {
ShareGroupDLQMetadataCacheHelper cacheHelper,
Time time,
Timer timer,
- ShareGroupMetrics shareGroupMetrics
+ ShareGroupMetrics shareGroupMetrics,
+ LogReader logReader
) {
if (client == null) {
throw new IllegalArgumentException("Kafkaclient must not be
null.");
@@ -121,10 +140,15 @@ public class ShareGroupDLQStateManager {
throw new IllegalArgumentException("ShareGroupMetrics must not be
null.");
}
+ if (logReader == null) {
+ throw new IllegalArgumentException("LogReader must not be null.");
+ }
+
this.time = time;
this.timer = timer;
this.cacheHelper = cacheHelper;
this.shareGroupMetrics = shareGroupMetrics;
+ this.logReader = logReader;
this.sender = new SendThread(
"ShareGroupDLQSendThread",
client,
@@ -345,10 +369,19 @@ public class ShareGroupDLQStateManager {
}
public ProduceRequestData.TopicProduceData topicProduceData() {
+ Map<Long, Record> originalRecordData = maybeFetchRecordData();
+
List<SimpleRecord> simpleRecords = new ArrayList<>();
for (long i = param.firstOffset(); i <= param.lastOffset(); i++) {
long timestamp = time.hiResClockMs();
- simpleRecords.add(new SimpleRecord(timestamp, (byte[]) null,
null, headers(i)));
+ ByteBuffer key = null;
+ ByteBuffer value = null;
+ Record record = originalRecordData.get(i);
+ if (record != null) {
+ key = record.hasKey() ? record.key() : null;
+ value = record.hasValue() ? record.value() : null;
+ }
+ simpleRecords.add(new SimpleRecord(timestamp, key, value,
headers(i)));
}
MemoryRecords records = MemoryRecords.withRecords(
@@ -653,6 +686,80 @@ public class ShareGroupDLQStateManager {
requestErrorResponse(clientResponseError.exception());
}
}
+
+ 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
+ }
+ }
+
+ // 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;
+ }
+ 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);
+ }
}
private class SendThread extends InterBrokerSendThread {
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 6eb5b659250..2da51e250d3 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
@@ -22,6 +22,7 @@ import org.apache.kafka.clients.MockClient;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.compress.Compression;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.message.CreateTopicsResponseData;
@@ -30,12 +31,14 @@ import org.apache.kafka.common.message.ProduceResponseData;
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.SimpleRecord;
import org.apache.kafka.common.requests.CreateTopicsRequest;
import org.apache.kafka.common.requests.CreateTopicsResponse;
import org.apache.kafka.common.requests.ProduceRequest;
import org.apache.kafka.common.requests.ProduceResponse;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.share.LogReader;
import
org.apache.kafka.server.share.dlq.ShareGroupDLQMetadataCacheHelper.TopicPartitionData;
import org.apache.kafka.server.share.metrics.ShareGroupMetrics;
import org.apache.kafka.server.util.MockTime;
@@ -43,6 +46,8 @@ import org.apache.kafka.server.util.timer.MockTimer;
import org.apache.kafka.server.util.timer.SystemTimer;
import org.apache.kafka.server.util.timer.SystemTimerReaper;
import org.apache.kafka.server.util.timer.Timer;
+import org.apache.kafka.storage.internals.log.FetchDataInfo;
+import org.apache.kafka.storage.internals.log.LogReadResult;
import org.apache.kafka.test.TestUtils;
import org.junit.jupiter.api.AfterEach;
@@ -51,8 +56,10 @@ import org.mockito.Mockito;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -72,6 +79,7 @@ import static
org.apache.kafka.server.share.dlq.ShareGroupDLQStateManager.Produc
import static
org.apache.kafka.server.share.dlq.ShareGroupDLQStateManager.ProduceRequestHandler.HEADER_DLQ_ERRORS_OFFSET;
import static
org.apache.kafka.server.share.dlq.ShareGroupDLQStateManager.ProduceRequestHandler.HEADER_DLQ_ERRORS_PARTITION;
import static
org.apache.kafka.server.share.dlq.ShareGroupDLQStateManager.ProduceRequestHandler.HEADER_DLQ_ERRORS_TOPIC;
+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.assertInstanceOf;
@@ -82,6 +90,7 @@ 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.anyInt;
+import static org.mockito.ArgumentMatchers.anySet;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
@@ -100,6 +109,7 @@ class ShareGroupDLQStateManagerTest {
private static final Uuid DLQ_TOPIC_ID = Uuid.randomUuid();
private static final Uuid SOURCE_TOPIC_ID = Uuid.randomUuid();
private static final Node DEFAULT_LEADER = new Node(0, HOST, PORT);
+ private static final LogReader MOCK_LOG_READER = mock(LogReader.class);
private final MockTimer mockTimer = new MockTimer(MOCK_TIME);
private final ShareGroupMetrics mockMetrics =
mock(ShareGroupMetrics.class);
@@ -118,6 +128,7 @@ class ShareGroupDLQStateManagerTest {
private Timer timer;
private ShareGroupDLQMetadataCacheHelper cacheHelper;
private ShareGroupMetrics shareGroupMetrics;
+ private LogReader logReader;
Builder withClient(KafkaClient client) {
this.client = client;
@@ -144,15 +155,21 @@ class ShareGroupDLQStateManagerTest {
return this;
}
+ Builder withLogReader(LogReader logReader) {
+ this.logReader = logReader;
+ return this;
+ }
+
ShareGroupDLQStateManager build() {
// Default to the test-class mockMetrics field so tests can verify
interactions
// without having to thread a custom metrics mock through the
builder.
return new ShareGroupDLQStateManager(
client != null ? client : new MockClient(MOCK_TIME),
- cacheHelper != null ? cacheHelper :
happyCacheHelper(DEFAULT_LEADER),
+ cacheHelper != null ? cacheHelper :
cacheHelper(DEFAULT_LEADER),
time,
timer != null ? timer : mockTimer,
- shareGroupMetrics != null ? shareGroupMetrics : mockMetrics
+ shareGroupMetrics != null ? shareGroupMetrics : mockMetrics,
+ logReader == null ? MOCK_LOG_READER : logReader
);
}
}
@@ -172,7 +189,13 @@ class ShareGroupDLQStateManagerTest {
);
}
- private static ShareGroupDLQMetadataCacheHelper happyCacheHelper(Node
leader) {
+ /**
+ * Util method to populate DLQ metadata cache with sensible defaults.
+ *
+ * @param leader Node representing leader node for DLQ partition.
+ * @return Populated DLQ cache helper object.
+ */
+ private static ShareGroupDLQMetadataCacheHelper cacheHelper(Node leader) {
ShareGroupDLQMetadataCacheHelper helper =
mock(ShareGroupDLQMetadataCacheHelper.class);
when(helper.shareGroupDlqTopic(GROUP_ID)).thenReturn(Optional.of(DLQ_TOPIC));
when(helper.shareGroupDlqTopicPrefix()).thenReturn(Optional.empty());
@@ -187,6 +210,7 @@ class ShareGroupDLQStateManagerTest {
Optional.of(DLQ_TOPIC_ID),
List.of(leader)
));
+
when(helper.isShareGroupDlqCopyRecordEnabled(GROUP_ID)).thenReturn(false);
return helper;
}
@@ -208,7 +232,8 @@ class ShareGroupDLQStateManagerTest {
* request. {@code sharedHeaders} are expected to be identical on every
record in that partition;
* the offset header is built per-record from {@code firstOffset}..{@code
lastOffset}.
*/
- private record ExpectedDlqPartition(long firstOffset, long lastOffset,
Map<String, String> sharedHeaders) {
+ private record ExpectedDlqPartition(long firstOffset, long lastOffset,
Map<String, String> sharedHeaders,
+ List<byte[]> keys, List<byte[]>
values) {
}
/**
@@ -243,6 +268,19 @@ class ShareGroupDLQStateManagerTest {
expectedHeaders.put(HEADER_DLQ_ERRORS_OFFSET,
Long.toString(expectedOffset));
assertEquals(expectedHeaders, actualHeaders,
"Partition " + partition.index() + " record at offset " +
expectedOffset + " has unexpected headers");
+
+ if (!expected.keys().isEmpty()) {
+ assertKeyValue(record, expected.keys.get(recordCount),
true);
+ } else {
+ assertFalse(record.hasKey());
+ }
+
+ if (!expected.values().isEmpty()) {
+ assertKeyValue(record, expected.values.get(recordCount),
false);
+ } else {
+ assertFalse(record.hasValue());
+ }
+
expectedOffset++;
recordCount++;
}
@@ -251,20 +289,46 @@ class ShareGroupDLQStateManagerTest {
}
}
+ private static void assertKeyValue(Record record, byte[] expectedData,
boolean isKey) {
+ if (expectedData == null) {
+ if (isKey) {
+ assertFalse(record.hasKey());
+ } else {
+ assertFalse(record.hasValue());
+ }
+ } else {
+ if (isKey && !record.hasValue()) {
+ fail("record key not found");
+ }
+
+ if (!isKey && !record.hasValue()) {
+ fail("record value not found");
+ }
+
+ byte[] actualChunk = new byte[expectedData.length];
+ if (isKey) {
+ record.key().get(actualChunk);
+ } else {
+ record.value().get(actualChunk);
+ }
+ assertArrayEquals(expectedData, actualChunk);
+ }
+ }
+
// ---- Constructor null-check tests ----
@Test
public void testConstructorRejectsNullClient() {
ShareGroupDLQMetadataCacheHelper cacheHelper =
mock(ShareGroupDLQMetadataCacheHelper.class);
assertThrows(IllegalArgumentException.class,
- () -> new ShareGroupDLQStateManager(null, cacheHelper, MOCK_TIME,
mockTimer, mockMetrics));
+ () -> new ShareGroupDLQStateManager(null, cacheHelper, MOCK_TIME,
mockTimer, mockMetrics, MOCK_LOG_READER));
}
@Test
public void testConstructorRejectsNullCacheHelper() {
KafkaClient client = mock(KafkaClient.class);
assertThrows(IllegalArgumentException.class,
- () -> new ShareGroupDLQStateManager(client, null, MOCK_TIME,
mockTimer, mockMetrics));
+ () -> new ShareGroupDLQStateManager(client, null, MOCK_TIME,
mockTimer, mockMetrics, MOCK_LOG_READER));
}
@Test
@@ -272,7 +336,7 @@ class ShareGroupDLQStateManagerTest {
KafkaClient client = mock(KafkaClient.class);
ShareGroupDLQMetadataCacheHelper cacheHelper =
mock(ShareGroupDLQMetadataCacheHelper.class);
assertThrows(IllegalArgumentException.class,
- () -> new ShareGroupDLQStateManager(client, cacheHelper, null,
mockTimer, mockMetrics));
+ () -> new ShareGroupDLQStateManager(client, cacheHelper, null,
mockTimer, mockMetrics, MOCK_LOG_READER));
}
@Test
@@ -280,7 +344,7 @@ class ShareGroupDLQStateManagerTest {
KafkaClient client = mock(KafkaClient.class);
ShareGroupDLQMetadataCacheHelper cacheHelper =
mock(ShareGroupDLQMetadataCacheHelper.class);
assertThrows(IllegalArgumentException.class,
- () -> new ShareGroupDLQStateManager(client, cacheHelper,
MOCK_TIME, null, mockMetrics));
+ () -> new ShareGroupDLQStateManager(client, cacheHelper,
MOCK_TIME, null, mockMetrics, MOCK_LOG_READER));
}
@Test
@@ -288,7 +352,15 @@ class ShareGroupDLQStateManagerTest {
KafkaClient client = mock(KafkaClient.class);
ShareGroupDLQMetadataCacheHelper cacheHelper =
mock(ShareGroupDLQMetadataCacheHelper.class);
assertThrows(IllegalArgumentException.class,
- () -> new ShareGroupDLQStateManager(client, cacheHelper,
MOCK_TIME, mockTimer, null));
+ () -> new ShareGroupDLQStateManager(client, cacheHelper,
MOCK_TIME, mockTimer, null, MOCK_LOG_READER));
+ }
+
+ @Test
+ public void testConstructorRejectsNullLogReader() {
+ KafkaClient client = mock(KafkaClient.class);
+ ShareGroupDLQMetadataCacheHelper cacheHelper =
mock(ShareGroupDLQMetadataCacheHelper.class);
+ assertThrows(IllegalArgumentException.class,
+ () -> new ShareGroupDLQStateManager(client, cacheHelper,
MOCK_TIME, mockTimer, mockMetrics, null));
}
// ---- Lifecycle tests ----
@@ -428,7 +500,7 @@ class ShareGroupDLQStateManagerTest {
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);
@@ -481,7 +553,7 @@ class ShareGroupDLQStateManagerTest {
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);
@@ -539,7 +611,7 @@ class ShareGroupDLQStateManagerTest {
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);
@@ -694,7 +766,7 @@ class ShareGroupDLQStateManagerTest {
HEADER_DLQ_ERRORS_GROUP, GROUP_ID,
HEADER_DLQ_ERRORS_DELIVERY_COUNT, "1",
HEADER_DLQ_ERRORS_MESSAGE, "simulated cause"
- ))
+ ), List.of(), List.of())
));
// CreateTopics retried, but produce only ran once (after the
eventual create success).
verify(mockMetrics).recordDLQProduce(GROUP_ID);
@@ -742,7 +814,7 @@ class ShareGroupDLQStateManagerTest {
HEADER_DLQ_ERRORS_GROUP, GROUP_ID,
HEADER_DLQ_ERRORS_DELIVERY_COUNT, "1",
HEADER_DLQ_ERRORS_MESSAGE, "simulated cause"
- ))
+ ), List.of(), List.of())
);
for (ProduceRequest pr : capturedProduces) {
assertDlqProduceRecordHeaders(pr, expectedByPartition);
@@ -991,12 +1063,12 @@ class ShareGroupDLQStateManagerTest {
HEADER_DLQ_ERRORS_TOPIC, "source-topic",
HEADER_DLQ_ERRORS_PARTITION, "0",
HEADER_DLQ_ERRORS_GROUP, GROUP_ID
- ));
+ ), List.of(), List.of());
ExpectedDlqPartition expectedDlqPartition1 = new
ExpectedDlqPartition(0L, 0L, Map.of(
HEADER_DLQ_ERRORS_TOPIC, "source-topic",
HEADER_DLQ_ERRORS_PARTITION, "1",
HEADER_DLQ_ERRORS_GROUP, GROUP_ID
- ));
+ ), List.of(), List.of());
if (capturedProduces.size() == 1) {
assertDlqProduceRecordHeaders(capturedProduces.get(0), Map.of(
0, expectedDlqPartition0,
@@ -1054,7 +1126,7 @@ class ShareGroupDLQStateManagerTest {
HEADER_DLQ_ERRORS_TOPIC, "source-topic",
HEADER_DLQ_ERRORS_PARTITION, "0",
HEADER_DLQ_ERRORS_GROUP, GROUP_ID
- ))
+ ), List.of(), List.of())
));
verify(mockMetrics).recordDLQProduce(GROUP_ID);
verify(mockMetrics).recordDLQRecordWrite(GROUP_ID, 1);
@@ -1182,7 +1254,7 @@ class ShareGroupDLQStateManagerTest {
client.prepareResponseFrom(body -> true, null, DEFAULT_LEADER);
- stateManager =
builder().withClient(client).withCacheHelper(happyCacheHelper(DEFAULT_LEADER)).build();
+ stateManager =
builder().withClient(client).withCacheHelper(cacheHelper(DEFAULT_LEADER)).build();
Future<Boolean> done = executor.submit(() -> {
long start = System.currentTimeMillis();
@@ -1389,6 +1461,251 @@ class ShareGroupDLQStateManagerTest {
assertEquals(DLQ_TOPIC_ID,
request.data().topicData().iterator().next().topicId());
}
+ // --- DLQ record with copy record enabled ---
+
+ @Test
+ public void testDLQRecordCopyEnabled() throws Exception {
+ MockClient client = new MockClient(MOCK_TIME);
+ List<ProduceRequest> capturedProduces = new ArrayList<>();
+ client.prepareResponseFrom(
+ body -> {
+ if (body instanceof ProduceRequest pr) {
+ capturedProduces.add(pr);
+ return true;
+ }
+ return false;
+ },
+ successfulProduceResponse(0),
+ DEFAULT_LEADER
+ );
+
+ ShareGroupDLQRecordParameter param = param();
+ 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);
+ 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);
+
+ 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 testDLQRecordCopyEnabledWithPartialLogReaderRecords() throws
Exception {
+ MockClient client = new MockClient(MOCK_TIME);
+ List<ProduceRequest> capturedProduces = new ArrayList<>();
+ client.prepareResponseFrom(
+ body -> {
+ if (body instanceof ProduceRequest pr) {
+ capturedProduces.add(pr);
+ return true;
+ }
+ return false;
+ },
+ successfulProduceResponse(0),
+ DEFAULT_LEADER
+ );
+
+ ShareGroupDLQRecordParameter param = param(); // 3 offset 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);
+ 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,
+ 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(), 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);
+
+ 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"
+ ), Arrays.asList(keyData1, keyData2, keyData3),
Arrays.asList(valueData1, valueData2, valueData3))
+ ));
+ verify(mockMetrics).recordDLQProduce(GROUP_ID);
+ verify(mockMetrics).recordDLQRecordWrite(GROUP_ID, 3);
+ verify(mockMetrics, never()).recordDLQProduceFailed(any());
+ }
+
+ @Test
+ public void testDLQRecordCopyEnabledWithMultipleLogReaderIterations()
throws Exception {
+ MockClient client = new MockClient(MOCK_TIME);
+ List<ProduceRequest> capturedProduces = new ArrayList<>();
+ client.prepareResponseFrom(
+ body -> {
+ if (body instanceof ProduceRequest pr) {
+ capturedProduces.add(pr);
+ return true;
+ }
+ return false;
+ },
+ 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);
+ 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);
+
+ 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"
+ ), Arrays.asList(keyData1, keyData2, null),
Arrays.asList(valueData1, valueData2, null))
+ ));
+ verify(mockMetrics).recordDLQProduce(GROUP_ID);
+ verify(mockMetrics).recordDLQRecordWrite(GROUP_ID, 3);
+ verify(mockMetrics, never()).recordDLQProduceFailed(any());
+ }
+
+ @Test
+ public void testDLQRecordCopyEnabledWithErrorOnLogRead() throws Exception {
+ MockClient client = new MockClient(MOCK_TIME);
+ List<ProduceRequest> capturedProduces = new ArrayList<>();
+ client.prepareResponseFrom(
+ body -> {
+ if (body instanceof ProduceRequest pr) {
+ capturedProduces.add(pr);
+ return true;
+ }
+ return false;
+ },
+ successfulProduceResponse(0),
+ DEFAULT_LEADER
+ );
+
+ 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);
+
+ 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());
+ }
+
private static ShareGroupDLQStateManager.ProduceRequestHandler
newHandlerForCoalesceTest(
ShareGroupDLQStateManager manager,
String groupId,
@@ -1407,7 +1724,6 @@ class ShareGroupDLQStateManagerTest {
3);
}
-
// ---- Response builder helpers ----
private static ProduceResponse successfulProduceResponse(int partition) {
