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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
