[jira] [Commented] (KAFKA-20803) Share-group persister: O(N^2) response demultiplexing in PersisterStateManager batched RPCs

["Omnia Ibrahim (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-20803?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18104465#comment-18104465 ] 

Omnia Ibrahim commented on KAFKA-20803:
---------------------------------------

Moving to 4.5 as we are now in 4.4.0 code freeze
 

> 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.4.0
>
>
> 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)