You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "smjn (via GitHub)" <gi...@apache.org> on 2026/06/04 22:15:11 UTC
[PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
smjn opened a new pull request, #22479:
URL: https://github.com/apache/kafka/pull/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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3406366451
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -653,6 +686,73 @@ private void handleProduceResponse(ClientResponse response) {
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) {
+ offsets.put(tp, nextOffset);
+
+ 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 {}.", nextOffset, this);
+ return Map.of();
+ }
+
+ res.info().delayedRemoteStorageFetch.ifPresent(data -> log.info(
+ "Some offset data in is in remote storage. Skipping it."));
+
+ boolean continueFetch = false;
+ for (RecordBatch batch : res.info().records.batches()) {
+ for (Record record : batch) {
+ if (record.offset() < param.firstOffset()) 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 Collections.unmodifiableMap(recordMap);
Review Comment:
Will add but it would unnecessarily use would be less efficient.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3379838146
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -653,6 +684,80 @@ private void handleProduceResponse(ClientResponse response) {
requestErrorResponse(clientResponseError.exception());
}
}
+
+ private void maybeFetchRecordData() {
+ if (cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
+ // A non-null originalRecordData indicates that the data for the offsets was
+ // already fetched at a previous time. This could happen in case there was
+ // a retriable exception in a previous produce request, and it is being re-sent.
+ // This optimization will help in reducing LogReader.read calls. Note that an
+ // empty (but non-null) map means a previous fetch found no records in range
+ // (e.g. all offsets compacted away), so we still skip re-fetching in that case.
+ if (originalRecordData != null) {
+ return;
+ }
Review Comment:
When the dlq call is made by the dlqManager - it implies all records being DLQ'ed have already present in the topic which is read only. So if a previous call has indeed found some data related to the dlqManager call, the next time that call happens (maybe retry), the data can only reduce (perhaps compaction) but not increase.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #22479:
URL: https://github.com/apache/kafka/pull/22479#issuecomment-4679602806
@apoorvmittal10
> Sorry, to ask again. Are the originalDataRecords be long lived in broker memory, if yes then how long?
The originalRecordData map is a per handler (created as part of external DLQ request) member and will be short-lived (until the RPC fails catastrophically or succeeds).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3367014782
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -345,10 +366,20 @@ public void populateDLQTopicData() throws ConfigException {
}
public ProduceRequestData.TopicProduceData topicProduceData() {
+ 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)));
+ int recordIndex = (int) (i - param.firstOffset());
+ ByteBuffer key = null;
+ ByteBuffer value = null;
+ if (originalRecordData.size() > recordIndex) {
Review Comment:
I realised, there could be a bug here if the offsets are sparsely filled due to compaction on the user topic - replacing this DS with a hashmap and adding the suggested non-null check should be sufficient.
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -345,10 +366,20 @@ public void populateDLQTopicData() throws ConfigException {
}
public ProduceRequestData.TopicProduceData topicProduceData() {
+ 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)));
+ int recordIndex = (int) (i - param.firstOffset());
+ ByteBuffer key = null;
+ ByteBuffer value = null;
+ if (originalRecordData.size() > recordIndex) {
Review Comment:
I realised there could be a bug here if the offsets are sparsely filled due to compaction on the user topic - replacing this DS with a hashmap and adding the suggested non-null check should be sufficient.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3395097661
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -214,6 +231,7 @@ class ProduceRequestHandler implements RequestCompletionHandler {
private Node dlqPartitionLeaderNode;
private int dlqDestinationPartition;
private ShareGroupDLQMetadataCacheHelper.TopicPartitionData dlqTopicPartitionData;
+ private Map<Long, Record> originalRecordData;
Review Comment:
Yes - we do have have the ability to resend the DLQ requests in case retryable error is received from PRODUCE RPC (upto 5 times based on exponential backoff). This will be more helpful when RemoteLogReader is introduced.
If we are retrying, there is no need to keep refetch the records associated with the call and hence keeping them cached would be more efficient. We fetch records only once when the DLQ PRODUCE RPC is sent for the first time.
The `originalRecordData` map is a per handler (created as part of external DLQ request) member and will be short-lived (until the RPC fails catastrophically or succeeds)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3379851288
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -121,10 +134,16 @@ public 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.maxFetchBytes = maxFetchBytes;
+ this.logReader = logReader;
Review Comment:
We'll keep best effort approach for now and a follow up PR for gathering data from remote storage. https://issues.apache.org/jira/browse/KAFKA-20678
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3406061294
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -653,6 +686,73 @@ private void handleProduceResponse(ClientResponse response) {
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) {
+ offsets.put(tp, nextOffset);
+
+ 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 {}.", nextOffset, this);
+ return Map.of();
+ }
+
+ res.info().delayedRemoteStorageFetch.ifPresent(data -> log.info(
+ "Some offset data in is in remote storage. Skipping it."));
+
+ boolean continueFetch = false;
+ for (RecordBatch batch : res.info().records.batches()) {
+ for (Record record : batch) {
+ if (record.offset() < param.firstOffset()) 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 Collections.unmodifiableMap(recordMap);
Review Comment:
nit: You can use Map.copyOf as what we use elsewhere instead of Collections.unmodifiableMap
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -653,6 +686,73 @@ private void handleProduceResponse(ClientResponse response) {
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) {
+ offsets.put(tp, nextOffset);
+
+ 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 {}.", nextOffset, this);
+ return Map.of();
+ }
+
+ res.info().delayedRemoteStorageFetch.ifPresent(data -> log.info(
+ "Some offset data in is in remote storage. Skipping it."));
+
+ boolean continueFetch = false;
+ for (RecordBatch batch : res.info().records.batches()) {
+ for (Record record : batch) {
+ if (record.offset() < param.firstOffset()) 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 Collections.unmodifiableMap(recordMap);
+ }
+ recordMap.put(record.offset(), record);
+ nextOffset = record.offset() + 1;
+ continueFetch = true;
+ }
+ }
+ if (!continueFetch) break; // no more records available (reached HWM/LEO)
+ }
+ log.trace("Full log fetch took {} ms for {} records starting at {} for {}", time.hiResClockMs() - startTime,
+ recordCount, param.firstOffset(), this);
+ return Collections.unmodifiableMap(recordMap);
+ }
Review Comment:
Do you think an info log is needed when recordMap count for data filled is not same as recordCount needed?
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -653,6 +686,73 @@ private void handleProduceResponse(ClientResponse response) {
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) {
+ offsets.put(tp, nextOffset);
+
+ 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 {}.", nextOffset, this);
+ return Map.of();
+ }
+
+ res.info().delayedRemoteStorageFetch.ifPresent(data -> log.info(
+ "Some offset data in is in remote storage. Skipping it."));
+
+ boolean continueFetch = false;
+ for (RecordBatch batch : res.info().records.batches()) {
+ for (Record record : batch) {
+ if (record.offset() < param.firstOffset()) 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 Collections.unmodifiableMap(recordMap);
+ }
+ recordMap.put(record.offset(), record);
+ nextOffset = record.offset() + 1;
+ continueFetch = true;
+ }
+ }
+ if (!continueFetch) break; // no more records available (reached HWM/LEO)
+ }
+ log.trace("Full log fetch took {} ms for {} records starting at {} for {}", time.hiResClockMs() - startTime,
+ recordCount, param.firstOffset(), this);
+ return Collections.unmodifiableMap(recordMap);
Review Comment:
Same here.
##########
server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java:
##########
@@ -1389,6 +1447,116 @@ public void testCoalesceProduceRequestsSkipsHandlerWhoseTopicProduceDataThrows()
assertEquals(DLQ_TOPIC_ID, request.data().topicData().iterator().next().topicId());
}
+ // --- DLQ record with copy record enabled ---
+
+ @Test
+ public void testHappyPathWithDLQRecordCopyEnabled() throws Exception {
Review Comment:
nit
```suggestion
public void testDLQRecordCopyEnabled() throws Exception {
```
##########
server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java:
##########
@@ -1389,6 +1447,116 @@ public void testCoalesceProduceRequestsSkipsHandlerWhoseTopicProduceDataThrows()
assertEquals(DLQ_TOPIC_ID, request.data().topicData().iterator().next().topicId());
}
+ // --- DLQ record with copy record enabled ---
+
+ @Test
+ public void testHappyPathWithDLQRecordCopyEnabled() 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);
+
Review Comment:
nit: remove unnecessary line break, seems 2 consective line breaks.
##########
server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java:
##########
@@ -1389,6 +1447,116 @@ public void testCoalesceProduceRequestsSkipsHandlerWhoseTopicProduceDataThrows()
assertEquals(DLQ_TOPIC_ID, request.data().topicData().iterator().next().topicId());
}
+ // --- DLQ record with copy record enabled ---
+
+ @Test
+ public void testHappyPathWithDLQRecordCopyEnabled() 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 = happyCacheHelper(DEFAULT_LEADER);
Review Comment:
nit: I would suggest to rename `happyCacheHelper` as `cacheHelper` and write in comments for the method regarding it's purpose. Also I couldn't find one unhappy cache helper method, hence ;)
##########
server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java:
##########
@@ -1389,6 +1447,116 @@ public void testCoalesceProduceRequestsSkipsHandlerWhoseTopicProduceDataThrows()
assertEquals(DLQ_TOPIC_ID, request.data().topicData().iterator().next().topicId());
}
+ // --- DLQ record with copy record enabled ---
+
+ @Test
+ public void testHappyPathWithDLQRecordCopyEnabled() 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 = happyCacheHelper(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 testErrorPathWithDLQRecordCopyEnabled() 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 = happyCacheHelper(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());
+ }
+
Review Comment:
Can you please add more tests
1. When the record map do not return all the offsets requested.
2. When log reader requires multiple while loop iterations to fetch all required data.
##########
server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java:
##########
@@ -1389,6 +1447,116 @@ public void testCoalesceProduceRequestsSkipsHandlerWhoseTopicProduceDataThrows()
assertEquals(DLQ_TOPIC_ID, request.data().topicData().iterator().next().topicId());
}
+ // --- DLQ record with copy record enabled ---
+
+ @Test
+ public void testHappyPathWithDLQRecordCopyEnabled() 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 = happyCacheHelper(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 testErrorPathWithDLQRecordCopyEnabled() throws Exception {
Review Comment:
nit
```suggestion
public void testDLQRecordCopyEnabledWithErrorOnLogRead() throws Exception {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3398461167
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -653,6 +684,80 @@ private void handleProduceResponse(ClientResponse response) {
requestErrorResponse(clientResponseError.exception());
}
}
+
+ private void maybeFetchRecordData() {
+ if (cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
+ // A non-null originalRecordData indicates that the data for the offsets was
+ // already fetched at a previous time. This could happen in case there was
+ // a retriable exception in a previous produce request, and it is being re-sent.
+ // This optimization will help in reducing LogReader.read calls. Note that an
+ // empty (but non-null) map means a previous fetch found no records in range
+ // (e.g. all offsets compacted away), so we still skip re-fetching in that case.
+ if (originalRecordData != null) {
+ return;
+ }
Review Comment:
No, not long lived. The duration will be the lifespan of a single DLQ call (until success or final failure).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #22479:
URL: https://github.com/apache/kafka/pull/22479#issuecomment-4695882580
@apoorvmittal10 Thanks for the review, inc comments
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #22479:
URL: https://github.com/apache/kafka/pull/22479#issuecomment-4690489114
@apoorvmittal10 Removing `originalRecordData` as member variable over broker mem concern.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3397475127
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -214,6 +231,7 @@ class ProduceRequestHandler implements RequestCompletionHandler {
private Node dlqPartitionLeaderNode;
private int dlqDestinationPartition;
private ShareGroupDLQMetadataCacheHelper.TopicPartitionData dlqTopicPartitionData;
+ private Map<Long, Record> originalRecordData;
Review Comment:
@apoorvmittal10
No, `originalDataRecords` was always a member variable of the handler class - not the top level ShareGroupDLQStateManager. New handler class object is created for each new DLQ request from the SharePartition. `originalDataRecords` is NOT shared between multiple DLQ requests.
Approach is same - its just that it is initialized lazily.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3412137054
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -653,6 +686,73 @@ private void handleProduceResponse(ClientResponse response) {
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) {
+ offsets.put(tp, nextOffset);
+
+ 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 {}.", nextOffset, this);
+ return Map.of();
+ }
+
+ res.info().delayedRemoteStorageFetch.ifPresent(data -> log.info(
+ "Some offset data in is in remote storage. Skipping it."));
+
+ boolean continueFetch = false;
+ for (RecordBatch batch : res.info().records.batches()) {
+ for (Record record : batch) {
+ if (record.offset() < param.firstOffset()) 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 Collections.unmodifiableMap(recordMap);
+ }
+ recordMap.put(record.offset(), record);
+ nextOffset = record.offset() + 1;
+ continueFetch = true;
+ }
+ }
+ if (!continueFetch) break; // no more records available (reached HWM/LEO)
+ }
+ log.trace("Full log fetch took {} ms for {} records starting at {} for {}", time.hiResClockMs() - startTime,
+ recordCount, param.firstOffset(), this);
+ return Collections.unmodifiableMap(recordMap);
+ }
Review Comment:
The info log is not always needed I think but only when the size is not same. Currently it prints on every DLQ record which is not adding much value.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3398453004
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -653,6 +684,80 @@ private void handleProduceResponse(ClientResponse response) {
requestErrorResponse(clientResponseError.exception());
}
}
+
+ private void maybeFetchRecordData() {
+ if (cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
+ // A non-null originalRecordData indicates that the data for the offsets was
+ // already fetched at a previous time. This could happen in case there was
+ // a retriable exception in a previous produce request, and it is being re-sent.
+ // This optimization will help in reducing LogReader.read calls. Note that an
+ // empty (but non-null) map means a previous fetch found no records in range
+ // (e.g. all offsets compacted away), so we still skip re-fetching in that case.
+ if (originalRecordData != null) {
+ return;
+ }
+
+ 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
+ maxFetchBytes, // 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, maxFetchBytes);
+
+ // 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) {
+ offsets.put(tp, nextOffset);
+
+ 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 {}.", nextOffset, this);
+ return;
+ }
+
+ boolean done = false;
+ for (RecordBatch batch : res.info().records.batches()) {
+ for (Record record : batch) {
Review Comment:
General integ tests JIRA - https://issues.apache.org/jira/browse/KAFKA-20686
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3376398859
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -121,10 +134,16 @@ public 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.maxFetchBytes = maxFetchBytes;
+ this.logReader = logReader;
Review Comment:
Generally yes but are we always going to read from local log and not from remote storage for data? If remote fetches are required then this approach alone will not work, we require handling to fetch messages from remote storage.
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -653,6 +684,80 @@ private void handleProduceResponse(ClientResponse response) {
requestErrorResponse(clientResponseError.exception());
}
}
+
+ private void maybeFetchRecordData() {
+ if (cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
+ // A non-null originalRecordData indicates that the data for the offsets was
+ // already fetched at a previous time. This could happen in case there was
+ // a retriable exception in a previous produce request, and it is being re-sent.
+ // This optimization will help in reducing LogReader.read calls. Note that an
+ // empty (but non-null) map means a previous fetch found no records in range
+ // (e.g. all offsets compacted away), so we still skip re-fetching in that case.
+ if (originalRecordData != null) {
+ return;
+ }
+
+ 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
+ maxFetchBytes, // 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, maxFetchBytes);
+
+ // 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) {
+ offsets.put(tp, nextOffset);
+
+ 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 {}.", nextOffset, this);
+ return;
+ }
+
+ boolean done = false;
+ for (RecordBatch batch : res.info().records.batches()) {
+ for (Record record : batch) {
Review Comment:
Can it happen that while fetching the offsets were not compacted but when DLQ is triggered then offsets were compacted and not found? Can we have a test for same?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3379805488
##########
core/src/main/scala/kafka/server/BrokerServer.scala:
##########
@@ -777,7 +777,9 @@ class BrokerServer(
new ShareCoordinatorMetadataCacheHelperImpl(metadataCache, key => shareCoordinator.partitionFor(key), config.interBrokerListenerName, groupConfigManager),
Time.SYSTEM,
shareGroupTimer,
- shareGroupMetrics
+ shareGroupMetrics,
+ config.getInt(ServerConfigs.FETCH_MAX_BYTES_CONFIG),
Review Comment:
Ok, then I will keep a non-configurable 1 MB max bytes limit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3379864096
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -653,6 +684,80 @@ private void handleProduceResponse(ClientResponse response) {
requestErrorResponse(clientResponseError.exception());
}
}
+
+ private void maybeFetchRecordData() {
+ if (cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
+ // A non-null originalRecordData indicates that the data for the offsets was
+ // already fetched at a previous time. This could happen in case there was
+ // a retriable exception in a previous produce request, and it is being re-sent.
+ // This optimization will help in reducing LogReader.read calls. Note that an
+ // empty (but non-null) map means a previous fetch found no records in range
+ // (e.g. all offsets compacted away), so we still skip re-fetching in that case.
+ if (originalRecordData != null) {
+ return;
+ }
+
+ 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
+ maxFetchBytes, // 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, maxFetchBytes);
+
+ // 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) {
+ offsets.put(tp, nextOffset);
+
+ 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 {}.", nextOffset, this);
+ return;
+ }
+
+ boolean done = false;
+ for (RecordBatch batch : res.info().records.batches()) {
+ for (Record record : batch) {
Review Comment:
will add a unit test but an integ test will be better place to test this, which will be added in subsequent PRs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3397475127
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -214,6 +231,7 @@ class ProduceRequestHandler implements RequestCompletionHandler {
private Node dlqPartitionLeaderNode;
private int dlqDestinationPartition;
private ShareGroupDLQMetadataCacheHelper.TopicPartitionData dlqTopicPartitionData;
+ private Map<Long, Record> originalRecordData;
Review Comment:
@apoorvmittal10
No, `originalDataRecords` was always a member variable of the handler class - not the top level ShareGroupDLQStateManager.
Approach is same - its just that it is initialized lazily.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #22479:
URL: https://github.com/apache/kafka/pull/22479#issuecomment-4626633061
### Sample output
```
$ bin/kafka-features.sh --bootstrap-server localhost:9092 upgrade --feature share.version=2
share.version was upgraded to 2.
$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config "share.auto.offset.reset=earliest" --entity-type groups --entity-name gs1
Completed updating config for group gs1.
$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config "errors.deadletterqueue.topic.name=dlq.gs1dlqtopic" --entity-type groups --entity-name gs1
Completed updating config for group gs1.
$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config "errors.deadletterqueue.copy.record.enable=true" --entity-type groups --entity-name gs1
Completed updating config for group gs1.
$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic t1
Created topic t1.
$ bin/kafka-producer-perf-test.sh --bootstrap-server localhost:9092 --topic t1 --num-records 5 --payload-monotonic --throughput -1
5 records sent, 15.723270 records/sec (0.00 MB/sec), 180.80 ms avg latency, 308.00 ms max latency, 149 ms 50th, 308 ms 95th, 308 ms 99th, 308 ms 99.9th.
$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic dlq.gs1dlqtopic --from-beginning --property print.key=true --property print.headers=true --property print.partition=true --property print.timestamp=true --timeout-ms 6000
Option --property is deprecated and will be removed in a future version. Use --formatter-property instead.
The consumer rebalance protocol (KIP-848) is production-ready! Set group.protocol=consumer to try it out. See https://kafka.apache.org/documentation/#consumer_rebalance_protocol
CreateTime:491928092 Partition:0 __dlq.errors.topic:t1,__dlq.errors.partition:0,__dlq.errors.offset:0,__dlq.errors.group:gs1,__dlq.errors.delivery.count:1,__dlq.errors.message:Offset rejected by client. null 0
CreateTime:491928092 Partition:0 __dlq.errors.topic:t1,__dlq.errors.partition:0,__dlq.errors.offset:1,__dlq.errors.group:gs1,__dlq.errors.delivery.count:1,__dlq.errors.message:Offset rejected by client. null 1
CreateTime:491928092 Partition:0 __dlq.errors.topic:t1,__dlq.errors.partition:0,__dlq.errors.offset:2,__dlq.errors.group:gs1,__dlq.errors.delivery.count:1,__dlq.errors.message:Offset rejected by client. null 2
CreateTime:491928092 Partition:0 __dlq.errors.topic:t1,__dlq.errors.partition:0,__dlq.errors.offset:3,__dlq.errors.group:gs1,__dlq.errors.delivery.count:1,__dlq.errors.message:Offset rejected by client. null 3
CreateTime:491928092 Partition:0 __dlq.errors.topic:t1,__dlq.errors.partition:0,__dlq.errors.offset:4,__dlq.errors.group:gs1,__dlq.errors.delivery.count:1,__dlq.errors.message:Offset rejected by client. null 4
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3367019403
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -653,6 +684,69 @@ private void handleProduceResponse(ClientResponse response) {
requestErrorResponse(clientResponseError.exception());
}
}
+
+ private void maybeFetchRecordData() {
+ if (cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
+ if (!originalRecordData.isEmpty()) {
+ return;
+ }
+
+ 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
+ maxFetchBytes, // 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);
+
+ List<Record> records = new ArrayList<>(recordCount);
+ LinkedHashMap<TopicIdPartition, Long> offsets = new LinkedHashMap<>();
+ LinkedHashMap<TopicIdPartition, Integer> maxBytesMap = new LinkedHashMap<>();
+ maxBytesMap.put(tp, maxFetchBytes);
+
+ while (nextOffset <= endOffset) {
Review Comment:
Hi,
We already have a guard against that - if no data arrives `for (RecordBatch batch : res.info().records.batches())` never executes and the check
```
if (!done) break; // no more records available (reached HWM/LEO)
```
will break the while loop.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3367019403
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -653,6 +684,69 @@ private void handleProduceResponse(ClientResponse response) {
requestErrorResponse(clientResponseError.exception());
}
}
+
+ private void maybeFetchRecordData() {
+ if (cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
+ if (!originalRecordData.isEmpty()) {
+ return;
+ }
+
+ 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
+ maxFetchBytes, // 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);
+
+ List<Record> records = new ArrayList<>(recordCount);
+ LinkedHashMap<TopicIdPartition, Long> offsets = new LinkedHashMap<>();
+ LinkedHashMap<TopicIdPartition, Integer> maxBytesMap = new LinkedHashMap<>();
+ maxBytesMap.put(tp, maxFetchBytes);
+
+ while (nextOffset <= endOffset) {
Review Comment:
Hi,
We already have a guard against that - if no data arrives the for loop:
```
for (RecordBatch batch : res.info().records.batches())
```
never executes and the check
```
if (!done) break; // no more records available (reached HWM/LEO)
```
will break the while loop.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3376326611
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -653,6 +684,80 @@ private void handleProduceResponse(ClientResponse response) {
requestErrorResponse(clientResponseError.exception());
}
}
+
+ private void maybeFetchRecordData() {
+ if (cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
+ // A non-null originalRecordData indicates that the data for the offsets was
+ // already fetched at a previous time. This could happen in case there was
+ // a retriable exception in a previous produce request, and it is being re-sent.
+ // This optimization will help in reducing LogReader.read calls. Note that an
+ // empty (but non-null) map means a previous fetch found no records in range
+ // (e.g. all offsets compacted away), so we still skip re-fetching in that case.
+ if (originalRecordData != null) {
+ return;
+ }
Review Comment:
And how does it guarantee that the data fetched in some previous iteration would be same for next DLQ call?
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -653,6 +684,80 @@ private void handleProduceResponse(ClientResponse response) {
requestErrorResponse(clientResponseError.exception());
}
}
+
+ private void maybeFetchRecordData() {
+ if (cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
+ // A non-null originalRecordData indicates that the data for the offsets was
+ // already fetched at a previous time. This could happen in case there was
+ // a retriable exception in a previous produce request, and it is being re-sent.
+ // This optimization will help in reducing LogReader.read calls. Note that an
+ // empty (but non-null) map means a previous fetch found no records in range
+ // (e.g. all offsets compacted away), so we still skip re-fetching in that case.
+ if (originalRecordData != null) {
+ return;
+ }
+
+ 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
+ maxFetchBytes, // 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, maxFetchBytes);
+
+ // 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) {
+ offsets.put(tp, nextOffset);
+
+ 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 {}.", nextOffset, this);
+ return;
+ }
+
+ boolean done = false;
+ for (RecordBatch batch : res.info().records.batches()) {
+ for (Record record : batch) {
+ if (record.offset() < param.firstOffset()) 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);
+ originalRecordData = recordMap;
+ return;
+ }
+ recordMap.put(record.offset(), record);
+ nextOffset = record.offset() + 1;
+ done = true;
Review Comment:
Seems a bit opposite to me as `done` was very implicit for me that the process completed but it's the opposite in the code.
```suggestion
continueFetch = true;
```
##########
core/src/main/scala/kafka/server/BrokerServer.scala:
##########
@@ -777,7 +777,9 @@ class BrokerServer(
new ShareCoordinatorMetadataCacheHelperImpl(metadataCache, key => shareCoordinator.partitionFor(key), config.interBrokerListenerName, groupConfigManager),
Time.SYSTEM,
shareGroupTimer,
- shareGroupMetrics
+ shareGroupMetrics,
+ config.getInt(ServerConfigs.FETCH_MAX_BYTES_CONFIG),
Review Comment:
It's kind of reading through a buffer where you have small default. I would suggest same that instead of going with 55MB default you have some lower value as most rejects will be for a specific offset. And you anyways have a while loop in implementation to iterate for further offsets.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3379864096
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -653,6 +684,80 @@ private void handleProduceResponse(ClientResponse response) {
requestErrorResponse(clientResponseError.exception());
}
}
+
+ private void maybeFetchRecordData() {
+ if (cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
+ // A non-null originalRecordData indicates that the data for the offsets was
+ // already fetched at a previous time. This could happen in case there was
+ // a retriable exception in a previous produce request, and it is being re-sent.
+ // This optimization will help in reducing LogReader.read calls. Note that an
+ // empty (but non-null) map means a previous fetch found no records in range
+ // (e.g. all offsets compacted away), so we still skip re-fetching in that case.
+ if (originalRecordData != null) {
+ return;
+ }
+
+ 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
+ maxFetchBytes, // 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, maxFetchBytes);
+
+ // 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) {
+ offsets.put(tp, nextOffset);
+
+ 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 {}.", nextOffset, this);
+ return;
+ }
+
+ boolean done = false;
+ for (RecordBatch batch : res.info().records.batches()) {
+ for (Record record : batch) {
Review Comment:
integ test will be better place to test this, which will be added in subsequent PRs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3395097661
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -214,6 +231,7 @@ class ProduceRequestHandler implements RequestCompletionHandler {
private Node dlqPartitionLeaderNode;
private int dlqDestinationPartition;
private ShareGroupDLQMetadataCacheHelper.TopicPartitionData dlqTopicPartitionData;
+ private Map<Long, Record> originalRecordData;
Review Comment:
Yes - we do have have the ability to resend the DLQ requests in case retryable error is received from PRODUCE RPC (upto 5 times based on exponential backoff).
If we are retrying, there is no need to keep refetch the records associated with the call and hence keeping them cached would be more efficient. We fetch records only once when the DLQ PRODUCE RPC is sent for the first time.
The `originalRecordData` map is a per handler (created as part of external DLQ request) member and will be short-lived (until the RPC fails catastrophically or succeeds)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3389620766
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -82,12 +91,14 @@ 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;
+ private static final int DLQ_MAX_FETCH_BYTES = 1024 * 1024;
Review Comment:
Can you please write the code comment here about the choice of 1 MB as what we discussed in other comments.
##########
core/src/main/scala/kafka/server/BrokerServer.scala:
##########
@@ -772,13 +777,7 @@ class BrokerServer(
if (config.shareGroupConfig.shareGroupDLQManagerClassName.nonEmpty) {
val klass = Utils.loadClass(config.shareGroupConfig.shareGroupDLQManagerClassName, classOf[Object]).asInstanceOf[Class[ShareGroupDLQManager]]
if (klass.getName.equals(classOf[DefaultShareGroupDLQManager].getName)) {
- DefaultShareGroupDLQManager.instance(
- NetworkUtils.buildNetworkClient("ShareGroupDLQManager", config, metrics, Time.SYSTEM, new LogContext(s"[ShareGroupDLQManager broker=${config.brokerId}]")),
- new ShareCoordinatorMetadataCacheHelperImpl(metadataCache, key => shareCoordinator.partitionFor(key), config.interBrokerListenerName, groupConfigManager),
- Time.SYSTEM,
- shareGroupTimer,
- shareGroupMetrics
- )
+ DefaultShareGroupDLQManager.instance(NetworkUtils.buildNetworkClient("ShareGroupDLQManager", config, metrics, Time.SYSTEM, new LogContext(s"[ShareGroupDLQManager broker=${config.brokerId}]")), new ShareCoordinatorMetadataCacheHelperImpl(metadataCache, key => shareCoordinator.partitionFor(key), config.interBrokerListenerName, groupConfigManager), Time.SYSTEM, shareGroupTimer, shareGroupMetrics, shareGroupLogReader)
Review Comment:
Can we write the parameters in separate line as it was earlier for readability.
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -653,6 +684,80 @@ private void handleProduceResponse(ClientResponse response) {
requestErrorResponse(clientResponseError.exception());
}
}
+
+ private void maybeFetchRecordData() {
+ if (cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
+ // A non-null originalRecordData indicates that the data for the offsets was
+ // already fetched at a previous time. This could happen in case there was
+ // a retriable exception in a previous produce request, and it is being re-sent.
+ // This optimization will help in reducing LogReader.read calls. Note that an
+ // empty (but non-null) map means a previous fetch found no records in range
+ // (e.g. all offsets compacted away), so we still skip re-fetching in that case.
+ if (originalRecordData != null) {
+ return;
+ }
Review Comment:
Sorry, to ask again. Are the originalDataRecords be long lived in broker memory, if yes then how long?
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -214,6 +231,7 @@ class ProduceRequestHandler implements RequestCompletionHandler {
private Node dlqPartitionLeaderNode;
private int dlqDestinationPartition;
private ShareGroupDLQMetadataCacheHelper.TopicPartitionData dlqTopicPartitionData;
+ private Map<Long, Record> originalRecordData;
Review Comment:
Can you help me understand the benefit of keeping the originalRecordData as instance level and not fetching the data for offsets locally while processing a DLQ message from a queue?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3366977909
##########
core/src/main/scala/kafka/server/BrokerServer.scala:
##########
@@ -777,7 +777,9 @@ class BrokerServer(
new ShareCoordinatorMetadataCacheHelperImpl(metadataCache, key => shareCoordinator.partitionFor(key), config.interBrokerListenerName, groupConfigManager),
Time.SYSTEM,
shareGroupTimer,
- shareGroupMetrics
+ shareGroupMetrics,
+ config.getInt(ServerConfigs.FETCH_MAX_BYTES_CONFIG),
Review Comment:
DLQ request could be for a single offset or multiple (if batch REJECTs). So one readCall could serve multiple offsets. Since we are reading user topic data - there is no good default value, hence opting for server default.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3366981087
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -231,6 +251,7 @@ public ProduceRequestHandler(
) {
this.param = param;
this.result = result;
+ this.originalRecordData = new ArrayList<>();
Review Comment:
Hi,
Yes, it is lazy. The actual records will only be fetched when the record copy config is enabled and share group is configured correctly. Opted to create empty ArrayList to prevent checking nulls.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3379851288
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -121,10 +134,16 @@ public 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.maxFetchBytes = maxFetchBytes;
+ this.logReader = logReader;
Review Comment:
We'll keep best effort approach for now and a follow up PR for gathering data from remote storage.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3395097661
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -214,6 +231,7 @@ class ProduceRequestHandler implements RequestCompletionHandler {
private Node dlqPartitionLeaderNode;
private int dlqDestinationPartition;
private ShareGroupDLQMetadataCacheHelper.TopicPartitionData dlqTopicPartitionData;
+ private Map<Long, Record> originalRecordData;
Review Comment:
Yes - we do have have the ability to resend the DLQ requests in case retryable error is received from PRODUCE RPC (upto 5 times based on exponential backoff).
If we are retrying, there is no need to keep refetch the records associated with the call and hence keeping them cached would be more efficient.
The `originalRecordData` map is a per handler (created as part of external DLQ request) member and will be short-lived (until the RPC fails catastrophically or succeeds)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3396540692
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -214,6 +231,7 @@ class ProduceRequestHandler implements RequestCompletionHandler {
private Node dlqPartitionLeaderNode;
private int dlqDestinationPartition;
private ShareGroupDLQMetadataCacheHelper.TopicPartitionData dlqTopicPartitionData;
+ private Map<Long, Record> originalRecordData;
Review Comment:
To be sure we have changed the approach now as what you mentioned from earlier being held at instnce level, correct?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3366981087
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -231,6 +251,7 @@ public ProduceRequestHandler(
) {
this.param = param;
this.result = result;
+ this.originalRecordData = new ArrayList<>();
Review Comment:
Hi,
I realised there could be a bug here if the offsets are sparsely filled due to compaction on the user topic - replacing this DS with a hashmap and adding the suggested non-null check should be sufficient.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3366981087
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -231,6 +251,7 @@ public ProduceRequestHandler(
) {
this.param = param;
this.result = result;
+ this.originalRecordData = new ArrayList<>();
Review Comment:
Hi,
I realised there could be a bug here if the offsets are sparsely filled due to compaction on the user topic - replacing this DS with a hashmap and initing lazily.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3361829394
##########
core/src/main/scala/kafka/server/BrokerServer.scala:
##########
@@ -777,7 +777,9 @@ class BrokerServer(
new ShareCoordinatorMetadataCacheHelperImpl(metadataCache, key => shareCoordinator.partitionFor(key), config.interBrokerListenerName, groupConfigManager),
Time.SYSTEM,
shareGroupTimer,
- shareGroupMetrics
+ shareGroupMetrics,
+ config.getInt(ServerConfigs.FETCH_MAX_BYTES_CONFIG),
+ new ReplicaManagerLogReader(replicaManager)
Review Comment:
There should already be one ReplicaManagerLogReader created for SharePartitionManager, why not to re-use?
##########
core/src/main/scala/kafka/server/BrokerServer.scala:
##########
@@ -777,7 +777,9 @@ class BrokerServer(
new ShareCoordinatorMetadataCacheHelperImpl(metadataCache, key => shareCoordinator.partitionFor(key), config.interBrokerListenerName, groupConfigManager),
Time.SYSTEM,
shareGroupTimer,
- shareGroupMetrics
+ shareGroupMetrics,
+ config.getInt(ServerConfigs.FETCH_MAX_BYTES_CONFIG),
Review Comment:
Are the reads going to be for multiple offsets or 1 at a time? I am expecting a specific offset data will be fetched, correct? If that's the case then do we need to 55MB as max bytes config?
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -345,10 +366,20 @@ public void populateDLQTopicData() throws ConfigException {
}
public ProduceRequestData.TopicProduceData topicProduceData() {
+ 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)));
+ int recordIndex = (int) (i - param.firstOffset());
+ ByteBuffer key = null;
+ ByteBuffer value = null;
+ if (originalRecordData.size() > recordIndex) {
Review Comment:
What's the significance of this check, isn't you expect that if originalRecordData is not empty then all offsets data should be present? So if (!originalRecordData.isEmpty()) check should be sufficient?
Also where do we validate that originalRecordData strictly has all offset data which is requested?
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -231,6 +251,7 @@ public ProduceRequestHandler(
) {
this.param = param;
this.result = result;
+ this.originalRecordData = new ArrayList<>();
Review Comment:
Shouldn't we lazily initialize the arraylist when original record data capture is enabled?
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -653,6 +684,69 @@ private void handleProduceResponse(ClientResponse response) {
requestErrorResponse(clientResponseError.exception());
}
}
+
+ private void maybeFetchRecordData() {
+ if (cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
+ if (!originalRecordData.isEmpty()) {
+ return;
+ }
Review Comment:
When this can happen, a comment in code will be helpful?
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -653,6 +684,69 @@ private void handleProduceResponse(ClientResponse response) {
requestErrorResponse(clientResponseError.exception());
}
}
+
+ private void maybeFetchRecordData() {
+ if (cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
+ if (!originalRecordData.isEmpty()) {
+ return;
+ }
+
+ 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
+ maxFetchBytes, // 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);
+
+ List<Record> records = new ArrayList<>(recordCount);
+ LinkedHashMap<TopicIdPartition, Long> offsets = new LinkedHashMap<>();
+ LinkedHashMap<TopicIdPartition, Integer> maxBytesMap = new LinkedHashMap<>();
+ maxBytesMap.put(tp, maxFetchBytes);
+
+ while (nextOffset <= endOffset) {
Review Comment:
Though it make sense to keep fetching till all offsets are fetched but I was wondering if there can be an issue where no new data arrives and the while loop continues so shall we have some safe check?
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -653,6 +684,69 @@ private void handleProduceResponse(ClientResponse response) {
requestErrorResponse(clientResponseError.exception());
}
}
+
+ private void maybeFetchRecordData() {
+ if (cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
+ if (!originalRecordData.isEmpty()) {
+ return;
+ }
+
+ 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
+ maxFetchBytes, // 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);
+
+ List<Record> records = new ArrayList<>(recordCount);
+ LinkedHashMap<TopicIdPartition, Long> offsets = new LinkedHashMap<>();
+ LinkedHashMap<TopicIdPartition, Integer> maxBytesMap = new LinkedHashMap<>();
+ maxBytesMap.put(tp, maxFetchBytes);
+
+ while (nextOffset <= endOffset) {
+ offsets.put(tp, nextOffset);
Review Comment:
Would be good to write a comment that the fetch is for same topic partition again hence just update the nextOffset without clearing the map.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #22479:
URL: https://github.com/apache/kafka/pull/22479#discussion_r3389609501
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
@@ -653,6 +684,80 @@ private void handleProduceResponse(ClientResponse response) {
requestErrorResponse(clientResponseError.exception());
}
}
+
+ private void maybeFetchRecordData() {
+ if (cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
+ // A non-null originalRecordData indicates that the data for the offsets was
+ // already fetched at a previous time. This could happen in case there was
+ // a retriable exception in a previous produce request, and it is being re-sent.
+ // This optimization will help in reducing LogReader.read calls. Note that an
+ // empty (but non-null) map means a previous fetch found no records in range
+ // (e.g. all offsets compacted away), so we still skip re-fetching in that case.
+ if (originalRecordData != null) {
+ return;
+ }
+
+ 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
+ maxFetchBytes, // 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, maxFetchBytes);
+
+ // 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) {
+ offsets.put(tp, nextOffset);
+
+ 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 {}.", nextOffset, this);
+ return;
+ }
+
+ boolean done = false;
+ for (RecordBatch batch : res.info().records.batches()) {
+ for (Record record : batch) {
Review Comment:
Do we have a jira for same to track?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20613: Add code to support copy record in DLQ state manager. [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 merged PR #22479:
URL: https://github.com/apache/kafka/pull/22479
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org