[jira] [Updated] (KAFKA-14029) Consumer response serialization could block other response handlers at scale

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

     [ https://issues.apache.org/jira/browse/KAFKA-14029?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

BugFinder updated KAFKA-14029:
------------------------------
    Priority: Minor  (was: Blocker)

> Consumer response serialization could block other response handlers at scale
> ----------------------------------------------------------------------------
>
>                 Key: KAFKA-14029
>                 URL: https://issues.apache.org/jira/browse/KAFKA-14029
>             Project: Kafka
>          Issue Type: Improvement
>          Components: consumer
>    Affects Versions: 3.2.0
>            Reporter: BugFinder
>            Priority: Minor
>
> Hi,
> We have been using our in-house tools to test Kafka's scalability to have an idea of how a large-scale deployment will work and where are the bottlenecks. For now, we are looking at version 3.2 and focused in a many-consumers scenario.
> Consumer-wise, we want to report a possible issue and eventually propose a solution, aiming to build our expertise in the system. When adding a new consumer to a group, the code path
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator$JoinGroupResponseHandler.handle *// (has a synchronized block)*
>  org.apache.kafka.clients.consumer.internals.AbstractCoordinator.onLeaderElected
>    org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeaderElected
>     org.apache.kafka.clients.consumer.internals.ConsumerProtocol.serializeAssignment
>      org.apache.kafka.clients.consumer.internals.ConsumerProtocol.serializeAssignment
>       org.apache.kafka.common.protocol.MessageUtil.toVersionPrefixedByteBuffer *// linear on size of message*
> could end up being costly when the size of the message is large. toVersionPrefixedByteBuffer seems to be linear in the size of the message, and albeit writing an array in linear time is not unreasonable at all, {*}under certain conditions, e.g. when under locks{*}, it can cause {*}undesired contention{*}. In this case, its invoked to serialize the assigment when adding a new consumer (on where there is another loop wraping up this path that seems to depend on the number of assignments, which could be another problematic dimension if growing causing undesired nesting), here
> [org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeaderElected]
> // ...
> Map<String, ByteBuffer> groupAssignment = new HashMap<>();
>         for (Map.Entry<String, Assignment> assignmentEntry : assignments.entrySet()) {
>             ByteBuffer buffer = *ConsumerProtocol.serializeAssignment(assignmentEntry.getValue()); // calls toVersionPrefixedByteBuffer* 
>             groupAssignment.put(assignmentEntry.getKey(), buffer);
>         }
> // ...
> The question here is, {*}is there a need to serialize the assignment inside the synchronized block{*}? if that assignment is too large, it could easily add a few seconds to the request and block others that use the same lock, like HeartbeatResponseHandler.
>  



--
This message was sent by Atlassian Jira
(v8.20.7#820007)