Shekhar Prasad Rajak created KAFKA-20803:
--------------------------------------------
Summary: Share-group persister: O(N^2) response demultiplexing in
PersisterStateManager batched RPCs
Key: KAFKA-20803
URL: https://issues.apache.org/jira/browse/KAFKA-20803
Project: Kafka
Issue Type: Improvement
Affects Versions: 4.3.1
Reporter: Shekhar Prasad Rajak
Fix For: 4.3.2
When the share-group persister coalesces many partitions into a single
WRITE/READ/READ_SUMMARY/INITIALIZE/DELETE_SHARE_GROUP_STATE RPC, the response
is demultiplexed in O(N ^ 2) time, where N is the number of
partitions batched into that RPC. This runs on the network send thread, so it
also stalls unrelated in-flight RPCs.
*Current behavior*
Request coalescing is fine — RequestCoalescerHelper.coalesceWrites
(PersisterStateManager.java) folds all N per-partition handlers into one
request in a single O(N) pass.
The response side is the problem. The completion callback hands the entire
combined response to every handler:
// PersisterStateManager.generateRequests()
handlersPerGroup.forEach(handler1 -> handler1.onComplete(response));
Each handler then linearly scans the full response to locate its own single
PartitionResult:
// WriteStateHandler.handleRequestResponse(),
for (WriteStateResult r : combinedResponse.data().results()) // walk
all topics
if (r.topicId().equals(partitionKey().topicId()))
r.partitions().stream()
.filter(p -> p.partition() == partitionKey().partition()) // walk
all partitions
.findFirst();
N handlers × O(N) scan each = O(N ^ 2). The five handler types
(WriteStateHandler, ReadStateHandler, ReadStateSummaryHandler,
InitializeStateHandler, DeleteStateHandler) all share this identical pattern.
*Why it matters*
- Share groups typically subscribe to a topic with many partitions, and many
partitions hash to the same share-coordinator node, so N is large in practice
(hundreds).
- The demux executes inside the completion callback on the
NetworkClient/InterBrokerSendThread, so the O(N^2) work blocks the send loop
and delays every other queued RPC (reads, writes to other coordinators,
find-coordinator). wakeup() only fires after the scan completes.
*Proposed fix*
Demultiplex the combined response once into a Map<Uuid topicId, Map<Integer
partition, PartitionResult>> (O(N)), then hand each handler only its own slice
for an O(1) lookup — replacing the per-handler linear
scan.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
