This is an automated email from the ASF dual-hosted git repository.
AndrewJSchofield pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 93d0db45199 KAFKA-20723: Merge records while coalescing share DLQ
requests. [2/N] (#22662)
93d0db45199 is described below
commit 93d0db451995652a5d0c45c544a436cf2fa1588a
Author: Sushant Mahajan <[email protected]>
AuthorDate: Wed Jun 24 21:08:11 2026 +0530
KAFKA-20723: Merge records while coalescing share DLQ requests. [2/N]
(#22662)
Current code in `ShareGroupDLQStateManager.coalesceProduceRequests`
combines requests by appending records in a list. For example, consider
2 DLQ requests `r1` and `r2`:
```
r1: (DLQPartition: 0, records: [a, b, c]),
r2: (DLQPartition: 0, records: [p, q, r])
would result in
produce request: (...., partitionData: [
(0, [a,b,c]),
(0, [p,q,r])
]
```
This is incorrect as the API (`KafkaApis.handleProduceRequest`) will
overwrite the record entries for the same `(topicId, partition)`
ignoring `(0, [a,b,c])`. To fix this the caller must send merged records
in
the request, which should be:
```
produce request: (...., partitionData: [
(0, [a,b,c,p,q,r])
]
```
Headers for records are preserved in both cases.
This PR fixes the issue caller side (`ShareGroupDLQStateManager`) and
adds a new unit test to verify the same.
Reviewers: Andrew Schofield <[email protected]>, Apoorv Mittal
<[email protected]>
---
.../share/dlq/ShareGroupDLQStateManager.java | 57 ++++++++++++---
.../share/dlq/ShareGroupDLQStateManagerTest.java | 85 +++++++++++++++++++++-
2 files changed, 131 insertions(+), 11 deletions(-)
diff --git
a/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
index 8a0b3e04d26..799346ebf8c 100644
---
a/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
+++
b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
@@ -924,19 +924,24 @@ public class ShareGroupDLQStateManager {
// Visibility for tests
static CoalesceResults coalesceProduceRequests(List<ProduceRequestHandler>
handlers) {
// Above handlers are destined for the same broker node - it could be
for different DLQ topics and partitions
- // but the same broker node. Now the produce request requires each
topic data request to be
- // scoped to a specific topic/topicId and the partition data could
have all the record information
- // and the destination DLQ partition. To accomplish this, we will map
handlers by DLQ topic id.
- Map<Uuid, ProduceRequestData.TopicProduceData> produceHandlerMap = new
HashMap<>();
+ // but the same broker node. The produce request requires each topic
data request to be scoped to a
+ // specific topic/topicId, and within a topic each partition must
appear at most once (the broker keys
+ // partitions by (topicId, index) and would otherwise drop all but one
entry). So we first collect the
+ // records into a map keyed by DLQ topic id and then DLQ partition -
merging the records of all handlers
+ // that target the same (topic, partition) - and then build a single
produce request from that map.
+ Map<Uuid, String> topicNames = new HashMap<>();
+ Map<Uuid, Map<Integer, List<MemoryRecords>>>
recordsByTopicAndPartition = new LinkedHashMap<>();
List<ProduceRequestHandler> liveHandlers = new
ArrayList<>(handlers.size());
handlers.forEach(handler -> {
try {
ProduceRequestData.TopicProduceData topicProduceData =
handler.topicProduceData();
- produceHandlerMap.computeIfAbsent(topicProduceData.topicId(),
topicId ->
- new ProduceRequestData.TopicProduceData()
- .setName(topicProduceData.name())
- .setTopicId(topicId)
- ).partitionData().addAll(topicProduceData.partitionData());
+ Uuid topicId = topicProduceData.topicId();
+ topicNames.putIfAbsent(topicId, topicProduceData.name());
+ Map<Integer, List<MemoryRecords>> partitionRecords =
+ recordsByTopicAndPartition.computeIfAbsent(topicId, k ->
new LinkedHashMap<>());
+ topicProduceData.partitionData().forEach(partitionData ->
+ partitionRecords.computeIfAbsent(partitionData.index(), k
-> new ArrayList<>())
+ .add((MemoryRecords) partitionData.records()));
liveHandlers.add(handler);
} catch (Exception exception) {
log.error("Unable to coalesce ProduceRequestData for handler
{}. It will be skipped from DLQ.", handler, exception);
@@ -944,8 +949,21 @@ public class ShareGroupDLQStateManager {
}
});
+ ProduceRequestData.TopicProduceDataCollection topicData = new
ProduceRequestData.TopicProduceDataCollection();
+ recordsByTopicAndPartition.forEach((topicId, partitionRecords) -> {
+ List<ProduceRequestData.PartitionProduceData> partitionData = new
ArrayList<>(partitionRecords.size());
+ partitionRecords.forEach((partitionIndex, records) ->
+ partitionData.add(new ProduceRequestData.PartitionProduceData()
+ .setIndex(partitionIndex)
+ .setRecords(mergeRecords(records))));
+ topicData.add(new ProduceRequestData.TopicProduceData()
+ .setName(topicNames.get(topicId))
+ .setTopicId(topicId)
+ .setPartitionData(partitionData));
+ });
+
ProduceRequestData data = new ProduceRequestData()
- .setTopicData(new
ProduceRequestData.TopicProduceDataCollection(produceHandlerMap.values().iterator()))
+ .setTopicData(topicData)
.setAcks((short) -1) // all replicas
.setTimeoutMs(ServerConfigs.REQUEST_TIMEOUT_MS_DEFAULT);
@@ -954,4 +972,23 @@ public class ShareGroupDLQStateManager {
liveHandlers
);
}
+
+ /**
+ * Merges the records of all handlers that target the same DLQ partition
into a single {@link MemoryRecords}
+ * (one record batch). The partition must appear only once in the
coalesced produce request, and a produce
+ * request is only allowed one record batch per partition - so when more
than one handler contributes records
+ * for a partition, they are combined into a single batch.
+ */
+ private static MemoryRecords mergeRecords(List<MemoryRecords> recordsList)
{
+ if (recordsList.size() == 1) {
+ return recordsList.get(0);
+ }
+ List<SimpleRecord> simpleRecords = new ArrayList<>();
+ for (MemoryRecords records : recordsList) {
+ for (Record record : records.records()) {
+ simpleRecords.add(new SimpleRecord(record.timestamp(),
record.key(), record.value(), record.headers()));
+ }
+ }
+ return MemoryRecords.withRecords(Compression.NONE,
simpleRecords.toArray(new SimpleRecord[0]));
+ }
}
diff --git
a/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java
b/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java
index a1767ab634b..2fd2cf2b76b 100644
---
a/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java
+++
b/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java
@@ -59,6 +59,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -1401,6 +1402,79 @@ class ShareGroupDLQStateManagerTest {
verify(mockMetrics, times(2)).recordDLQProduce(GROUP_ID);
}
+ @Test
+ public void testCoalesceProduceRequestsMergesRecordsForSameDlqPartition()
throws Exception {
+ // One share group with two source topics, two partitions each (four
source topic-partitions). The DLQ
+ // topic has a single partition (default cache helper), so all four
handlers target the same DLQ
+ // (topic, partition). Their records must be merged into a single
partition entry / single record batch,
+ // preserving every record - rather than producing duplicate partition
entries that the broker would
+ // collapse, dropping records.
+ Uuid sourceTopicAId = Uuid.randomUuid();
+ Uuid sourceTopicBId = Uuid.randomUuid();
+ List<TopicIdPartition> sources = List.of(
+ new TopicIdPartition(sourceTopicAId, 0, "source-topic-a"),
+ new TopicIdPartition(sourceTopicAId, 1, "source-topic-a"),
+ new TopicIdPartition(sourceTopicBId, 0, "source-topic-b"),
+ new TopicIdPartition(sourceTopicBId, 1, "source-topic-b"));
+
+ stateManager = builder().build();
+ List<ShareGroupDLQStateManager.ProduceRequestHandler> handlers = new
ArrayList<>();
+ for (TopicIdPartition source : sources) {
+ ShareGroupDLQStateManager.ProduceRequestHandler handler =
+ newHandlerForCoalesceTest(stateManager, GROUP_ID, source);
+ handler.populateDLQTopicData();
+ handlers.add(handler);
+ }
+
+ ShareGroupDLQStateManager.CoalesceResults result =
+ ShareGroupDLQStateManager.coalesceProduceRequests(handlers);
+
+ assertEquals(handlers, result.liveHandlers());
+
+ ProduceRequest request = ((ProduceRequest.Builder)
result.request()).build();
+ assertEquals(1, request.data().topicData().size(), "All records target
the single DLQ topic");
+ ProduceRequestData.TopicProduceData topic =
request.data().topicData().iterator().next();
+ assertEquals(DLQ_TOPIC_ID, topic.topicId());
+ assertEquals(1, topic.partitionData().size(),
+ "All four handlers target the same DLQ partition and must merge
into a single partition entry");
+ ProduceRequestData.PartitionProduceData partition =
topic.partitionData().get(0);
+ assertEquals(0, partition.index());
+
+ // The single merged batch must contain one record per source
topic-partition, with each record's
+ // source topic + partition headers intact.
+ MemoryRecords records = (MemoryRecords) partition.records();
+ assertEquals(sources.size(), recordCount(records));
+ Set<String> headers = new HashSet<>();
+ for (Record record : records.records()) {
+ headers.add(headerValue(record, HEADER_DLQ_ERRORS_TOPIC)
+ + "-" + headerValue(record, HEADER_DLQ_ERRORS_PARTITION));
+ }
+ assertEquals(
+ Set.of("source-topic-a-0", "source-topic-a-1", "source-topic-b-0",
"source-topic-b-1"),
+ headers);
+
+ verify(mockMetrics, times(sources.size())).recordDLQProduce(GROUP_ID);
+ }
+
+ private static int recordCount(MemoryRecords records) {
+ int count = 0;
+ Iterator<Record> iterator = records.records().iterator();
+ while (iterator.hasNext()) {
+ count++;
+ iterator.next();
+ }
+ return count;
+ }
+
+ private static String headerValue(Record record, String key) {
+ for (Header header : record.headers()) {
+ if (header.key().equals(key)) {
+ return new String(header.value(), StandardCharsets.UTF_8);
+ }
+ }
+ return null;
+ }
+
@Test
public void testCoalesceProduceRequestsKeepsDifferentDlqTopicsSeparate()
throws Exception {
String groupA = "group-a";
@@ -1744,10 +1818,19 @@ class ShareGroupDLQStateManagerTest {
ShareGroupDLQStateManager manager,
String groupId,
int sourcePartition
+ ) {
+ return newHandlerForCoalesceTest(manager, groupId,
+ new TopicIdPartition(SOURCE_TOPIC_ID, sourcePartition,
"source-topic"));
+ }
+
+ private static ShareGroupDLQStateManager.ProduceRequestHandler
newHandlerForCoalesceTest(
+ ShareGroupDLQStateManager manager,
+ String groupId,
+ TopicIdPartition source
) {
ShareGroupDLQRecordParameter param = new ShareGroupDLQRecordParameter(
groupId,
- new TopicIdPartition(SOURCE_TOPIC_ID, sourcePartition,
"source-topic"),
+ source,
0L, 0L,
Optional.empty(), Optional.empty());
return manager.new ProduceRequestHandler(
