[jira] [Commented] (KAFKA-2022) simpleconsumer.fetch(req) throws a java.nio.channels.ClosedChannelException: null exception when the original leader fails instead of being trapped in the fetchResponse api while consuming messages

["Jinder Aujla (JIRA)" <ji...@apache.org>]

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

Jinder Aujla commented on KAFKA-2022:
-------------------------------------

Hi

I also noticed this, here is part of the stack trace is there a work around or something I can do prevent this from happening?

java.nio.channels.ClosedChannelException
        at kafka.network.BlockingChannel.send(BlockingChannel.scala:100)
        at kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.scala:78)
        at kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:68)
        at kafka.consumer.SimpleConsumer.fetchOffsets(SimpleConsumer.scala:147)
        at kafka.javaapi.consumer.SimpleConsumer.fetchOffsets(SimpleConsumer.scala:99)


thanks

> simpleconsumer.fetch(req) throws a java.nio.channels.ClosedChannelException: null exception when the original leader fails instead of being trapped in the fetchResponse api while consuming messages
> -----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
>                 Key: KAFKA-2022
>                 URL: https://issues.apache.org/jira/browse/KAFKA-2022
>             Project: Kafka
>          Issue Type: Bug
>          Components: consumer
>    Affects Versions: 0.8.2.1
>         Environment: 3 linux nodes with both zookeepr & brokers running under respective users on each..
>            Reporter: Muqeet Mohammed Ali
>            Assignee: Neha Narkhede
>
> simpleconsumer.fetch(req) throws a java.nio.channels.ClosedChannelException: null exception when the original leader fails, instead of being trapped in the fetchResponse api while consuming messages. My understanding was that any fetch failures can be found via fetchResponse.hasError() call and then be handled to fetch new leader in this case. Below is the relevant code snippet from the simple consumer with comments marking the line causing exception..can you please comment on this?
> if (simpleconsumer == null) {
> 					simpleconsumer = new SimpleConsumer(leaderAddress.getHostName(), leaderAddress.getPort(), consumerTimeout,
> 							consumerBufferSize, consumerId);
> }
> FetchRequest req = new FetchRequestBuilder().clientId(getConsumerId())
> 						.addFetch(topic, partition, offsetManager.getTempOffset(), consumerBufferSize)
> 		// Note: the fetchSize might need to be increased
> 					// if large batches are written to Kafka
> 						.build();
> // exception is throw at the below line
> FetchResponse fetchResponse = simpleconsumer.fetch(req);
> if (fetchResponse.hasError()) {
> 			numErrors++;
> etc...



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)