[
https://issues.apache.org/jira/browse/KAFKA-4979?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15968922#comment-15968922
]
Nacho Munoz edited comment on KAFKA-4979 at 4/14/17 11:37 AM:
--------------------------------------------------------------
I was playing with the Kafka protocol from Clojure leveraging the AdminClient
send method. I'm happy to open a PR to fix this if it's considered an issue.
Please see below a handcrafted JoinGroupRequest that triggers the exception
described:
{noformat}
(send-to admin-client
(Node. 1 "localhost" 9092)
ApiKeys/JOIN_GROUP
(JoinGroupRequest$Builder. "group-test"
10000
""
"consumer"
[
(JoinGroupRequest$ProtocolMetadata. "range"
(ByteBuffer/wrap
(.getBytes "")))
(JoinGroupRequest$ProtocolMetadata. "roundrobin"
(ByteBuffer/wrap
(.getBytes "")))]))
{noformat}
was (Author: nachomdo):
I was playing with the Kafka protocol from Clojure leveraging the AdminClient
send method. I'm happy to open a PR to fix this if it's considered an issue.
Please see below a handcrafted JoinGroupRequest that triggers the exception
described:
{noformat}
(send-to admin-client (Node. 1 "localhost" 9092) ApiKeys/JOIN_GROUP
(JoinGroupRequest$Builder. "group-test"
10000
""
"consumer"
[
(JoinGroupRequest$ProtocolMetadata. "range"
(ByteBuffer/wrap
(.getBytes "")))
(JoinGroupRequest$ProtocolMetadata. "roundrobin"
(ByteBuffer/wrap
(.getBytes "")))]))
{noformat}
> Handcrafted JoinGroupRequest makes consumers on the same group to exit
> abruptly
> -------------------------------------------------------------------------------
>
> Key: KAFKA-4979
> URL: https://issues.apache.org/jira/browse/KAFKA-4979
> Project: Kafka
> Issue Type: Bug
> Components: consumer
> Affects Versions: 0.10.2.0
> Reporter: Nacho Munoz
> Priority: Minor
>
> Given an application with a consumer group X it is possible to make the
> application crash by creating a bad JoinGroupRequest for the same group. I
> don't know if that is the expected behaviour but I think that, maybe, this
> could return an error and leave the application unaffected by the rogue
> request. Please let me know if you need any further information or if I can
> do anything to help.
> You can see the exception below:
> [2017-03-28 16:57:00,819] ERROR Error processing message, terminating
> consumer process: (kafka.tools.ConsoleConsumer$)
> org.apache.kafka.common.protocol.types.SchemaException: Error reading field
> 'version': java.nio.BufferUnderflowException
> at org.apache.kafka.common.protocol.types.Schema.read(Schema.java:72)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerProtocol.deserializeSubscription(ConsumerProtocol.java:94)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.performAssignment(ConsumerCoordinator.java:320)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.onJoinLeader(AbstractCoordinator.java:488)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.access$1100(AbstractCoordinator.java:89)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator$JoinGroupResponseHandler.handle(AbstractCoordinator.java:438)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator$JoinGroupResponseHandler.handle(AbstractCoordinator.java:420)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:764)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:745)
> at
> org.apache.kafka.clients.consumer.internals.RequestFuture$1.onSuccess(RequestFuture.java:186)
> at
> org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess(RequestFuture.java:149)
> at
> org.apache.kafka.clients.consumer.internals.RequestFuture.complete(RequestFuture.java:116)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.fireCompletion(ConsumerNetworkClient.java:493)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.firePendingCompletedRequests(ConsumerNetworkClient.java:322)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:253)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:172)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:334)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:303)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:286)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1030)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:995)
> at kafka.consumer.NewShinyConsumer.receive(BaseConsumer.scala:88)
> at kafka.tools.ConsoleConsumer$.process(ConsoleConsumer.scala:120)
> at kafka.tools.ConsoleConsumer$.run(ConsoleConsumer.scala:75)
> at kafka.tools.ConsoleConsumer$.main(ConsoleConsumer.scala:50)
> at kafka.tools.ConsoleConsumer.main(ConsoleConsumer.scala)
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
