[
https://issues.apache.org/jira/browse/KAFKA-20803?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Shekhar Prasad Rajak updated KAFKA-20803:
-----------------------------------------
Labels: queue queues-for-kafka share-group (was: )
> 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
> Assignee: Shekhar Prasad Rajak
> Priority: Major
> Labels: queue, queues-for-kafka, share-group
> 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)
