Re: java.nio.channels.ClosedChannelException in console-consumer.sh
Some googling indicates that there are issues on AWS / EC2 when using the
private IP, and its recommended to use the public ip as the advertised
hostname instead.
I have zookeeper and Kafka both running on EC2, and both are in the same
availability zone, so both should be able to talk to each other using the
private IPs.
Shouldn't that be enough? I don't want to expose kafka publicly.
On Fri, Sep 16, 2016 at 10:48 PM, Ali Akhtar wrote:
> I've created a 3 broker kafka cluster, changing only the config values for
> broker id, log.dirs, and zookeeper connect. I left the remaining fields as
> default.
>
> The broker ids are 1, 2, 3. I opened the port 9092 on AWS.
>
> I then created a topic 'test' with replication factor of 2, and 3
> partitions.
>
> When I describe the topic using kafka-topics.sh --describe , it shows:
>
> Topic:test PartitionCount:3 ReplicationFactor:2 Configs:
> Topic: test Partition: 0 Leader: 1 Replicas: 1,3 Isr: 1,3
> Topic: test Partition: 1 Leader: 2 Replicas: 2,1 Isr: 2,1
> Topic: test Partition: 2 Leader: 3 Replicas: 3,2 Isr: 3,2
>
> So it looks like the 3 brokers have successfully connected to each other.
>
> I then tried running
>
> bin/kafka-console-consumer.sh --zookeeper $KAFKA_ZOOKEEPER_CONNECT --topic
> test --from-beginning
>
> But it began to give me a lot of the following exceptions:
>
> WARN Fetctestng topic metadata with correlation id 1 for topics
> [Set(test)] from broker [BrokerEndPoint(1,kafka1-1876849043-91zso,9092)]
> failed (kafka.client.ClientUtils$)
> java.nio.channels.ClosedChannelException
> at kafka.network.BlockingChannel.send(BlockingChannel.scala:110)
> at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:80)
> at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$
> doSend(SyncProducer.scala:79)
> at kafka.producer.SyncProducer.send(SyncProducer.scala:124)
> at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59)
> at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:94)
> at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(
> ConsumerFetcherManager.scala:66)
> at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
> [2016-09-16 17:22:11,323] WARN Fetctestng topic metadata with correlation
> id 2 for topics [Set(test)] from broker
> [BrokerEndPoint(3,kafka3-2571399577-96he4,9092)]
> failed (kafka.client.ClientUtils$)
> java.nio.channels.ClosedChannelException
> at kafka.network.BlockingChannel.send(BlockingChannel.scala:110)
> at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:80)
> at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$
> doSend(SyncProducer.scala:79)
> at kafka.producer.SyncProducer.send(SyncProducer.scala:124)
> at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59)
> at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:94)
> at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(
> ConsumerFetcherManager.scala:66)
> at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
>
> Likewise when I run kafka-console-producer.sh , I see errors like:
>
> [2016-09-16 17:24:42,901] WARN Error while fetching metadata with
> correlation id 1343 : {test=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.
> NetworkClient)
> [2016-09-16 17:24:43,012] WARN Error while fetching metadata with
> correlation id 1344 : {test=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.
> NetworkClient)
> [2016-09-16 17:24:43,127] WARN Error while fetching metadata with
> correlation id 1345 : {test=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.
> NetworkClient)
>
> Any ideas what the problem here is?
>
> I'm using kafka_2.11-0.10.0.1
>
>
>
>
>
Re: java.nio.channels.ClosedChannelException
I usually got that exception if
a) there is a firewall issue
b) you are trying to consume messages from another node and have not set
your advertised host name to the public ip address in the kafka server
settings.
On Tue, Feb 24, 2015 at 12:07 PM, alexey yakubovich <
alexe...@yahoo.com.invalid> wrote:
> Hi, I am learning kafka, ... hard way
> I try to run the example given here:
> https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+SimpleConsumer+Example
> Whatever I do i I have
> java.nio.channels.ClosedChannelException when try to send message
> Exception in thread "main" java.nio.channels.ClosedChannelExceptionat
> 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)
> And, sometimes it's
> 2015-02-24 13:46:51 INFO SimpleConsumer:68 - Reconnect due to socket
> error: java.io.EOFException: Received -1 when reading from channel,
> socket has likely been closed.
> (in all cases it fails in findLeader())
>
>
> At the same some other consumers work, as e.g. the command line
> console-consumer
> Now, I put a very small code snippet (below), that seems should do the
> job, but it's not, something missing. Please help.
> public class Test1 {public static void main(String[] args) {String
> host = "localhost"; int port = 2181; String topic = "test"; int
> soTimeout = -1; int bufferSize = 10; String clientId = "me-aly";
> SimpleConsumer consumer = new SimpleConsumer(host, port, soTimeout,
> bufferSize, clientId);List topics =
> Collections.singletonList(topic);
> TopicMetadataRequest req = new TopicMetadataRequest(topics);
> kafka.javaapi.TopicMetadataResponse resp = consumer.send(req);
> System.out.println(resp); }}
>
>
Re: java.nio.channels.ClosedChannelException...Firewall Issue?
Hi Jaikiran, Thanks for the reply! 1) I started Kafka server on instance A by simply downloading Kafka_2.10-0.8.2-beta.tgz from the kafka website, and using the scripts mentioned here: http://kafka.apache.org/documentation.html#introduction. This is the same way I downloaded Kafka on B, except I only started the producer, not the ZK or Server. 2) How do I check to see if the services are bound to non localhost IP? 3) I believe it is firewall because I started 2 instances (C and D) using the AWS Quick Start ( https://s3.amazonaws.com/quickstart-reference/cloudera/hadoop/latest/doc/Cloudera_EDH_on_AWS.pdf). I was able to download Kafka on C and D and pass messages using Kafka Producer/Consumer from D to C using the private/public ip address of C (which was the instance I was running the consumer/ZK/Server on). Since I used AWS Quick Start both of these were in a private subnet. Edit: Woops, I saw I said B was an ec2-instance, B is a gateway, and I downloaded Kafka same way I downloaded it on EC2 and I started a producer on the gateway hoping to send it to my EC2 instance. I can ping A (in a private subnet) from my gateway fine. I have pretty much changed all security settings on A so that it essentially has none, but still in a private subnet ( so can't ssh into it still from putty, can only ssh via the cluster launcher). I can't lessen the security anymore on instance A. Hope that wasn't too confusing, really appreciate the help. Thank you! Best, Su On Mon, Jan 19, 2015 at 7:58 PM, Jaikiran Pai wrote: > Hi Su, > > How exactly did you start the Kafka server on instance "A"? Are you sure > the services on it are bound to non localhost IP? What does the following > command result from instance B: > > telnet public.ip.of.A 9092 > > > -Jaikiran > > On Tuesday 20 January 2015 07:16 AM, Su She wrote: > >> Hello Everyone, >> >> Thank you for the help! >> >> Preface: I've created producers/consumers before and they have worked. I >> have also made consumers/producers using java programs, but they have all >> been locally. >> >> 1) I have a Zookeeper/Kafka Server running on an EC2 instance called "A" >> >> 2) I started the Zookeeper/Kafka Server on A and created a topic "test" >> like it says on the kafka documentation >> >> 3) I then started a console consumer on A like the documentation states >> for >> test. >> >> 4) I then downloaded Kafka on EC2 instance called "B" and created a >> console >> producer with broker with broker-list public.ip.of.A:9092 and topic test. >> >> 5) Once I start publishing messages this is the message I get (what's >> worse >> is that I can't cntrl C to stop and have to exit putty): >> >> WARN Fetching topic metadata with correlation id 16 for topics [Set(test)] >> from broker [id:0,host:54.183.40.224,port:9092] failed >> (kafka.client.ClientUtils$) >> >> java.nio.channels.ClosedChannelException >> >> at kafka.network.BlockingChannel.send(BlockingChannel.scala:97) >> >> at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer. >> scala:74) >> >> at >> kafka.producer.SyncProducer.kafka$producer$SyncProducer$$ >> doSend(SyncProducer.scala:73) >> >> at kafka.producer.SyncProducer.send(SyncProducer.scala:114) >> >> at >> kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:58) >> >> at >> kafka.producer.BrokerPartitionInfo.updateInfo( >> BrokerPartitionInfo.scala:82) >> >> at >> kafka.producer.async.DefaultEventHandler$$anonfun$handle$2.apply$mcV$sp( >> DefaultEventHandler.scala:78) >> >> at kafka.utils.Utils$.swallow(Utils.scala:172) >> >> at kafka.utils.Logging$class.swallowError(Logging.scala:106) >> >> at kafka.utils.Utils$.swallowError(Utils.scala:45) >> >> at >> kafka.producer.async.DefaultEventHandler.handle( >> DefaultEventHandler.scala:78) >> >> at >> kafka.producer.async.ProducerSendThread.tryToHandle( >> ProducerSendThread.scala:104) >> >> at >> kafka.producer.async.ProducerSendThread.processEvents( >> ProducerSendThread.scala:93) >> >> at >> kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:44) >> >> [2015-01-20 01:29:28,289] ERROR fetching topic metadata for topics >> [Set(test)] from broker [ArrayBuffer(id:0,host: >> >> Meanwhile in ZK...not sure if this happened as soon as I started >> publishing >> messages: >> >> Error:KeeperErrorCode = NoNode for /consumers/console-consumer- >> 2615/offsets >> (org.apache.zookeeper.server.PrepRequestProcessor) >> >> Any tips/suggestions are greatly appreciated! >> >> >
Re: java.nio.channels.ClosedChannelException...Firewall Issue?
Hi Su, How exactly did you start the Kafka server on instance "A"? Are you sure the services on it are bound to non localhost IP? What does the following command result from instance B: telnet public.ip.of.A 9092 -Jaikiran On Tuesday 20 January 2015 07:16 AM, Su She wrote: Hello Everyone, Thank you for the help! Preface: I've created producers/consumers before and they have worked. I have also made consumers/producers using java programs, but they have all been locally. 1) I have a Zookeeper/Kafka Server running on an EC2 instance called "A" 2) I started the Zookeeper/Kafka Server on A and created a topic "test" like it says on the kafka documentation 3) I then started a console consumer on A like the documentation states for test. 4) I then downloaded Kafka on EC2 instance called "B" and created a console producer with broker with broker-list public.ip.of.A:9092 and topic test. 5) Once I start publishing messages this is the message I get (what's worse is that I can't cntrl C to stop and have to exit putty): WARN Fetching topic metadata with correlation id 16 for topics [Set(test)] from broker [id:0,host:54.183.40.224,port:9092] failed (kafka.client.ClientUtils$) java.nio.channels.ClosedChannelException at kafka.network.BlockingChannel.send(BlockingChannel.scala:97) at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:74) at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:73) at kafka.producer.SyncProducer.send(SyncProducer.scala:114) at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:58) at kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:82) at kafka.producer.async.DefaultEventHandler$$anonfun$handle$2.apply$mcV$sp(DefaultEventHandler.scala:78) at kafka.utils.Utils$.swallow(Utils.scala:172) at kafka.utils.Logging$class.swallowError(Logging.scala:106) at kafka.utils.Utils$.swallowError(Utils.scala:45) at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:78) at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:104) at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:93) at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:44) [2015-01-20 01:29:28,289] ERROR fetching topic metadata for topics [Set(test)] from broker [ArrayBuffer(id:0,host: Meanwhile in ZK...not sure if this happened as soon as I started publishing messages: Error:KeeperErrorCode = NoNode for /consumers/console-consumer-2615/offsets (org.apache.zookeeper.server.PrepRequestProcessor) Any tips/suggestions are greatly appreciated!
