kafka-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Helin Xiang <xkee...@gmail.com>
Subject Re: Possibly leaking socket on ReplicaFetcherThread
Date Wed, 10 Dec 2014 03:14:32 GMT
Sorry for me not replying in the thread. ignore last email.


Hi, Jun

We experienced a network device problem. and cause all brokers crashed.
After investigation, we found server log throw similar exceptions.

this:

java.nio.channels.UnresolvedAddressException
        at sun.nio.ch.Net.checkAddress(Net.java:29)
        at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:512)
        at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57)
        at kafka.consumer.SimpleConsumer.connect(SimpleConsumer.scala:44)
        at kafka.consumer.SimpleConsumer.reconnect(SimpleConsumer.scala:57)
        at
kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.scala:79)
        at
kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:71)
        at
kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SimpleConsumer.scala:109)


and this:

2014-12-04 17:12:19,223 [kafka-scheduler-1] FATAL
kafka.server.ReplicaManager  - [Replica Manager on Broker 1]: Error writing
to highwatermark file:
java.io.FileNotFoundException:
/data3/kafka-logs/replication-offset-checkpoint.tmp (Too many open files)
        at java.io.FileOutputStream.open(Native Method)
        at java.io.FileOutputStream.<init>(FileOutputStream.java:194)
        at java.io.FileOutputStream.<init>(FileOutputStream.java:145)
        at java.io.FileWriter.<init>(FileWriter.java:73)
        at kafka.server.OffsetCheckpoint.write(OffsetCheckpoint.scala:37)
        at
kafka.server.ReplicaManager$$anonfun$checkpointHighWatermarks$2.apply(ReplicaManager.scala:447)
        at
kafka.server.ReplicaManager$$anonfun$checkpointHighWatermarks$2.apply(ReplicaManager.scala:444)
        at
scala.collection.immutable.HashMap$HashMap1.foreach(HashMap.scala:178)
        at
scala.collection.immutable.HashMap$HashTrieMap.foreach(HashMap.scala:347)


we count the number of java.nio.channels.UnresolvedAddressException  and
found it is around 63000, since a healthy kafka would open 2k fd in our
environment, we believe opened fd hit the our system's limit 65535.

so, it seems the bug is not fixed.

after checking the code, we believe it still would leak socket fd.
===============================================
our guess:

in simpleconsumer.scala:

  private def disconnect() = {
    if(blockingChannel.isConnected) {
      debug("Disconnecting from " + host + ":" + port)
      blockingChannel.disconnect()
    }
  }

but when the exception happened, blockingChannel.isConnected would be false,
because in blockingchannel.scala:

  def connect() = lock synchronized  {
    if(!connected) {
      channel = SocketChannel.open()
      if(readBufferSize > 0)
        channel.socket.setReceiveBufferSize(readBufferSize)
      if(writeBufferSize > 0)
        channel.socket.setSendBufferSize(writeBufferSize)
      channel.configureBlocking(true)
      channel.socket.setSoTimeout(readTimeoutMs)
      channel.socket.setKeepAlive(true)
      channel.socket.setTcpNoDelay(true)
      channel.connect(new InetSocketAddress(host, port))    <--  exception
happened here

      writeChannel = channel
      readChannel = Channels.newChannel(channel.socket().getInputStream)
      connected = true   <--  connected reset happened here, no chance to
be true
      ... ...


Thanks.

On Thu, Feb 13, 2014 at 1:33 PM, Jun Rao <junrao@gmail.com> wrote:

> This is fixed in https://issues.apache.org/jira/browse/KAFKA-1228 and will
> be included in 0.8.1 release.
>
> Thanks,
>
> Jun
>
>
> On Wed, Feb 12, 2014 at 6:28 PM, Priya Matpadi
> <priya.matpadi@ecofactor.com>wrote:
>
> > Hello,
> > Is there any progress on this issue? We also experience socket leak in
> case
> > of network outage.
> > Thanks,
> > Priya
> >
> >
> > On Fri, Jan 24, 2014 at 7:30 AM, Jun Rao <junrao@gmail.com> wrote:
> >
> > > Thanks for find this out. We probably should disconnect on any
> exception.
> > > Could you file a jira and perhaps attach a patch?
> > >
> > > Thanks,
> > >
> > > Jun
> > >
> > >
> > > On Fri, Jan 24, 2014 at 6:06 AM, Ahmy Yulrizka <ahmy@yulrizka.com>
> > wrote:
> > >
> > > > Hi,
> > > >
> > > > I Think I found out the problem..
> > > >
> > > > this is part of the stack trace. First i think there is connection
> > > problem,
> > > > and when connection restore it get new information from the zookeeper
> > > >
> > > > [2014-01-23 23:24:55,391] INFO Opening socket connection to server
> > > > host2.provider.com/2.2.2.2:2181 (org.apache.zookeeper.ClientCnxn)
> > > > [2014-01-23 23:24:55,391] INFO Socket connection established to
> > > > host2.provider.com/2.2.2.2:2181, initiating session
> > > > (org.apache.zookeeper.ClientCnxn)
> > > > [2014-01-23 23:24:55,391] DEBUG Session establishment request sent on
> > > > host2.provider.com/2.2.2.2:2181 (org.apache.zookeeper.ClientCnxn)
> > > > [2014-01-23 23:24:57,393] INFO Client session timed out, have not
> heard
> > > > from server in 2002ms for sessionid 0x0, closing socket connection
> and
> > > > attempting reconnect (org.apache.zookeeper.ClientCnxn)
> > > > [2014-01-23 23:24:58,047] INFO Opening socket connection to server
> > > > host3.provider.com/3.3.3.3:2181 (org.apache.zookeeper.ClientCnxn)
> > > > [2014-01-23 23:24:59,048] INFO Socket connection established to
> > > > host3.provider.com/3.3.3.3:2181, initiating session
> > > > (org.apache.zookeeper.ClientCnxn)
> > > > [2014-01-23 23:24:59,049] DEBUG Session establishment request sent on
> > > > host3.provider.com/3.3.3.3:2181 (org.apache.zookeeper.ClientCnxn)
> > > > [2014-01-23 23:24:59,539] INFO Session establishment complete on
> server
> > > > host3.provider.com/3.3.3.3:2181, sessionid = 0x343c13436e50001,
> > > negotiated
> > > > timeout = 6000 (org.apache.zookeeper.ClientCnxn)
> > > > [2014-01-23 23:24:59,539] DEBUG Received event: WatchedEvent
> > > > state:SyncConnected type:None path:null
> (org.I0Itec.zkclient.ZkClient)
> > > > [2014-01-23 23:24:59,539] INFO zookeeper state changed
> (SyncConnected)
> > > > (org.I0Itec.zkclient.ZkClient)
> > > > [2014-01-23 23:24:59,539] DEBUG New event: ZkEvent[State changed to
> > > > SyncConnected sent to
> > > > kafka.server.KafkaZooKeeper$SessionExpireListener@163e1f27]
> > > > (org.I0Itec.zkclient.ZkEventThread)
> > > > [2014-01-23 23:24:59,539] DEBUG New event: ZkEvent[State changed to
> > > > SyncConnected sent to
> > > > kafka.controller.KafkaController$SessionExpirationListener@486f44d9]
> > > > (org.I0Itec.zkclient.ZkEventThread)
> > > > [2014-01-23 23:24:59,539] DEBUG Leaving process event
> > > > (org.I0Itec.zkclient.ZkClient)
> > > > [2014-01-23 23:24:59,539] DEBUG State is SyncConnected
> > > > (org.I0Itec.zkclient.ZkClient)
> > > > [2014-01-23 23:24:59,540] DEBUG State is SyncConnected
> > > > (org.I0Itec.zkclient.ZkClient)
> > > >
> > > > Then the ReplicaFetcher Thread tries to reconnect.
> > > > At this point it tries to connect to other brokers. but then it can't
> > > > resolve the IP address and throws :
> > > > "java.nio.channels.UnresolvedAddressException"
> > > >
> > > > [2014-01-23 23:25:01,067] WARN [ReplicaFetcherThread-0-1], Error in
> > fetch
> > > > Name: FetchRequest; Version: 0; CorrelationId: 478411; ClientId:
> > > > ReplicaFetcherThread-0-1; ReplicaId: 2; MaxWait: 500 ms; MinBytes: 1
> > > bytes;
> > > > RequestInfo: [some-topic,0] ->
> > > > PartitionFetchInfo(1247662,1048576),[some-topic,3] ->
> > > > PartitionFetchInfo(839677,1048576)
> (kafka.server.ReplicaFetcherThread)
> > > > java.nio.channels.UnresolvedAddressException
> > > > at sun.nio.ch.Net.checkAddress(Net.java:89)
> > > > at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:510)
> > > >  at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57)
> > > > at kafka.consumer.SimpleConsumer.connect(SimpleConsumer.scala:44)
> > > >  at kafka.consumer.SimpleConsumer.reconnect(SimpleConsumer.scala:57)
> > > > at
> kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.scala:79)
> > > >  at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:71)
> > > > at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SimpleConsumer.scala:110)
> > > >  at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:110)
> > > > at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:110)
> > > >  at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> > > > at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply$mcV$sp(SimpleConsumer.scala:109)
> > > >  at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:109)
> > > > at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:109)
> > > >  at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> > > > at kafka.consumer.SimpleConsumer.fetch(SimpleConsumer.scala:108)
> > > >  at
> > > >
> > > >
> > >
> >
> kafka.server.AbstractFetcherThread.processFetchRequest(AbstractFetcherThread.scala:94)
> > > > at
> > > >
> > kafka.server.AbstractFetcherThread.doWork(AbstractFetcherThread.scala:86)
> > > >  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:51)
> > > > [2014-01-23 23:25:01,077] INFO Reconnect due to socket error: null
> > > > (kafka.consumer.SimpleConsumer)
> > > > [2014-01-23 23:25:01,078] WARN [ReplicaFetcherThread-0-1], Error in
> > fetch
> > > > Name: FetchRequest; Version: 0; CorrelationId: 478412; ClientId:
> > > > ReplicaFetcherThread-0-1; ReplicaId: 2; MaxWait: 500 ms; MinBytes: 1
> > > bytes;
> > > > RequestInfo: [some-topic,0] ->
> > > > PartitionFetchInfo(1247662,1048576),[some-topic,3] ->
> > > > PartitionFetchInfo(839677,1048576)
> (kafka.server.ReplicaFetcherThread)
> > > > java.nio.channels.UnresolvedAddressException
> > > > at sun.nio.ch.Net.checkAddress(Net.java:89)
> > > > at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:510)
> > > >  at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57)
> > > > at kafka.consumer.SimpleConsumer.connect(SimpleConsumer.scala:44)
> > > >  at kafka.consumer.SimpleConsumer.reconnect(SimpleConsumer.scala:57)
> > > > at
> kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.scala:79)
> > > >  at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:71)
> > > > at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SimpleConsumer.scala:110)
> > > >  at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:110)
> > > > at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:110)
> > > >  at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> > > > at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply$mcV$sp(SimpleConsumer.scala:109)
> > > >  at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:109)
> > > > at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:109)
> > > >  at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> > > > at kafka.consumer.SimpleConsumer.fetch(SimpleConsumer.scala:108)
> > > >  at
> > > >
> > > >
> > >
> >
> kafka.server.AbstractFetcherThread.processFetchRequest(AbstractFetcherThread.scala:94)
> > > > at
> > > >
> > kafka.server.AbstractFetcherThread.doWork(AbstractFetcherThread.scala:86)
> > > >  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:51)
> > > > [2014-01-23 23:25:01,079] INFO Reconnect due to socket error: null
> > > > (kafka.consumer.SimpleConsumer)
> > > >
> > > >
> > > > it throws bunch of this error until too many open file...
> > > >
> > > > [2014-01-23 23:25:03,756] INFO Reconnect due to socket error: null
> > > > (kafka.consumer.SimpleConsumer)
> > > > [2014-01-23 23:25:03,756] WARN [ReplicaFetcherThread-0-1], Error in
> > fetch
> > > > Name: FetchRequest; Version: 0; CorrelationId: 482406; ClientId:
> > > > ReplicaFetcherThread-0-1; ReplicaId: 2; MaxWait: 500 ms; MinBytes: 1
> > > bytes;
> > > > RequestInfo: [some-topic,0] ->
> > > > PartitionFetchInfo(1247662,1048576),[some-topic,2] ->
> > > > PartitionFetchInfo(1135106,1048576),[some-topic,3] ->
> > > > PartitionFetchInfo(839677,1048576),[some-topic,1] ->
> > > > PartitionFetchInfo(1273826,1048576)
> (kafka.server.ReplicaFetcherThread)
> > > > java.nio.channels.UnresolvedAddressException
> > > > at sun.nio.ch.Net.checkAddress(Net.java:89)
> > > > at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:510)
> > > >  at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57)
> > > > at kafka.consumer.SimpleConsumer.connect(SimpleConsumer.scala:44)
> > > >  at kafka.consumer.SimpleConsumer.reconnect(SimpleConsumer.scala:57)
> > > > at
> kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.scala:79)
> > > >  at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:71)
> > > > at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SimpleConsumer.scala:110)
> > > >  at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:110)
> > > > at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:110)
> > > >  at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> > > > at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply$mcV$sp(SimpleConsumer.scala:109)
> > > >  at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:109)
> > > > at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:109)
> > > >  at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> > > > at kafka.consumer.SimpleConsumer.fetch(SimpleConsumer.scala:108)
> > > >  at
> > > >
> > > >
> > >
> >
> kafka.server.AbstractFetcherThread.processFetchRequest(AbstractFetcherThread.scala:94)
> > > > at
> > > >
> > kafka.server.AbstractFetcherThread.doWork(AbstractFetcherThread.scala:86)
> > > >  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:51)
> > > > [2014-01-23 23:25:03,757] INFO Reconnect due to socket error: null
> > > > (kafka.consumer.SimpleConsumer)
> > > > [2014-01-23 23:25:03,757] WARN [ReplicaFetcherThread-0-1], Error in
> > fetch
> > > > Name: FetchRequest; Version: 0; CorrelationId: 482407; ClientId:
> > > > ReplicaFetcherThread-0-1; ReplicaId: 2; MaxWait: 500 ms; MinBytes: 1
> > > bytes;
> > > > RequestInfo: [some-topic,0] ->
> > > > PartitionFetchInfo(1247662,1048576),[some-topic,2] ->
> > > > PartitionFetchInfo(1135106,1048576),[some-topic,3] ->
> > > > PartitionFetchInfo(839677,1048576),[some-topic,1] ->
> > > > PartitionFetchInfo(1273826,1048576)
> (kafka.server.ReplicaFetcherThread)
> > > > java.net.SocketException: Too many open files
> > > > at sun.nio.ch.Net.socket0(Native Method)
> > > > at sun.nio.ch.Net.socket(Net.java:156)
> > > >  at sun.nio.ch.SocketChannelImpl.<init>(SocketChannelImpl.java:102)
> > > > at
> > > >
> > > >
> > >
> >
> sun.nio.ch.SelectorProviderImpl.openSocketChannel(SelectorProviderImpl.java:55)
> > > >  at java.nio.channels.SocketChannel.open(SocketChannel.java:122)
> > > > at kafka.network.BlockingChannel.connect(BlockingChannel.scala:48)
> > > >  at kafka.consumer.SimpleConsumer.connect(SimpleConsumer.scala:44)
> > > > at kafka.consumer.SimpleConsumer.reconnect(SimpleConsumer.scala:57)
> > > >  at
> > kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.scala:79)
> > > > at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:71)
> > > >  at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SimpleConsumer.scala:110)
> > > > at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:110)
> > > >  at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:110)
> > > > at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> > > >  at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply$mcV$sp(SimpleConsumer.scala:109)
> > > > at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:109)
> > > >  at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:109)
> > > > at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> > > >  at kafka.consumer.SimpleConsumer.fetch(SimpleConsumer.scala:108)
> > > > at
> > > >
> > > >
> > >
> >
> kafka.server.AbstractFetcherThread.processFetchRequest(AbstractFetcherThread.scala:94)
> > > >  at
> > > >
> > kafka.server.AbstractFetcherThread.doWork(AbstractFetcherThread.scala:86)
> > > > at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:51)
> > > > [2014-01-23 23:25:03,757] INFO Reconnect due to socket error: null
> > > > (kafka.consumer.SimpleConsumer)
> > > > [2014-01-23 23:25:03,757] WARN [ReplicaFetcherThread-0-1], Error in
> > fetch
> > > > Name: FetchRequest; Version: 0; CorrelationId: 482408; ClientId:
> > > > ReplicaFetcherThread-0-1; ReplicaId: 2; MaxWait: 500 ms; MinBytes: 1
> > > bytes;
> > > > RequestInfo: [some-topic,0] ->
> > > > PartitionFetchInfo(1247662,1048576),[some-topic,2] ->
> > > > PartitionFetchInfo(1135106,1048576),[some-topic,3] ->
> > > > PartitionFetchInfo(839677,1048576),[some-topic,1] ->
> > > > PartitionFetchInfo(1273826,1048576)
> (kafka.server.ReplicaFetcherThread)
> > > > java.net.SocketException: Too many open files
> > > > at sun.nio.ch.Net.socket0(Native Method)
> > > > at sun.nio.ch.Net.socket(Net.java:156)
> > > >  at sun.nio.ch.SocketChannelImpl.<init>(SocketChannelImpl.java:102)
> > > > at
> > > >
> > > >
> > >
> >
> sun.nio.ch.SelectorProviderImpl.openSocketChannel(SelectorProviderImpl.java:55)
> > > >  at java.nio.channels.SocketChannel.open(SocketChannel.java:122)
> > > > at kafka.network.BlockingChannel.connect(BlockingChannel.scala:48)
> > > >  at kafka.consumer.SimpleConsumer.connect(SimpleConsumer.scala:44)
> > > > at kafka.consumer.SimpleConsumer.reconnect(SimpleConsumer.scala:57)
> > > >  at
> > kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.scala:79)
> > > > at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:71)
> > > >  at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SimpleConsumer.scala:110)
> > > > at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:110)
> > > >  at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:110)
> > > > at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> > > >  at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply$mcV$sp(SimpleConsumer.scala:109)
> > > > at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:109)
> > > >  at
> > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:109)
> > > > at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> > > >  at kafka.consumer.SimpleConsumer.fetch(SimpleConsumer.scala:108)
> > > > at
> > > >
> > > >
> > >
> >
> kafka.server.AbstractFetcherThread.processFetchRequest(AbstractFetcherThread.scala:94)
> > > >  at
> > > >
> > kafka.server.AbstractFetcherThread.doWork(AbstractFetcherThread.scala:86)
> > > > at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:51)
> > > > [2014-01-23 23:25:03,758] INFO Reconnect due to socket error: null
> > > > (kafka.consumer.SimpleConsumer)
> > > >
> > > >
> > > >
> > > > I look into the source code of
> > > > core/src/main/scala/kafka/consumer/SimpleConsumer.scala line 79,
> > > >
> > > > it tries to reconnect(). but only close the connection when it is
> > > > "java.io.IOException" which did not catch
> > > > "java.nio.channels.UnresolvedAddressException"
> > > >
> > > >   private def sendRequest(request: RequestOrResponse): Receive = {
> > > >     lock synchronized {
> > > >       getOrMakeConnection()
> > > >       var response: Receive = null
> > > >       try {
> > > >         blockingChannel.send(request)
> > > >         response = blockingChannel.receive()
> > > >       } catch {
> > > >         case e : java.io.IOException =>
> > > >           info("Reconnect due to socket error:
> > %s".format(e.getMessage))
> > > >           // retry once
> > > >           try {
> > > >             reconnect()
> > > >             blockingChannel.send(request)
> > > >             response = blockingChannel.receive()
> > > >           } catch {
> > > >             case ioe: java.io.IOException =>
> > > >               disconnect()
> > > >               throw ioe
> > > >           }
> > > >         case e: Throwable => throw e
> > > >       }
> > > >       response
> > > >     }
> > > >   }
> > > >
> > > > This is my production setting
> > > >
> > > > OS: Ubuntu 12.04
> > > > kafka : kafka_2.8.0-0.8.0.jar
> > > >
> > > > java :
> > > > java version "1.6.0_27"
> > > > OpenJDK Runtime Environment (IcedTea6 1.12.6)
> > > > (6b27-1.12.6-1ubuntu0.12.04.4)
> > > > OpenJDK 64-Bit Server VM (build 20.0-b12, mixed mode)
> > > >
> > > > Some interesting fact that if i one of the file descriptor of the
> > process
> > > > using
> > > >
> > > > $ gdb -p KAFKA_PID
> > > > > call close(4567)
> > > >
> > > > it reuse that file description and establish connection to a consumer
> > > >
> > > > java       9708      kafka 4087u     sock                0,7
> > 0t0
> > > > 3258461771 can't identify protocol
> > > > java       9708      kafka 4088u     IPv4         3441430493
> > 0t0
> > > >      TCP host2.provider.com:9092->consumer.host.com:38208
> (ESTABLISHED)
> > > > java       9708      kafka 4089u     sock                0,7
> > 0t0
> > > > 3258461773 can't identify protocol
> > > > java       9708      kafka 4090u     sock                0,7
> > 0t0
> > > > 3258461774 can't identify protocol
> > > >
> > > >
> > > >
> > > >
> > > > --
> > > > Ahmy Yulrizka
> > > > http://ahmy.yulrizka.com
> > > > @yulrizka
> > > >
> > > >
> > > > On Wed, Jan 22, 2014 at 5:41 AM, Jun Rao <junrao@gmail.com> wrote:
> > > >
> > > > > Hmm, without knowing the client ip, it's hard to tell whether those
> > are
> > > > > from replication fetcher threads or not. Are most of those
> > connections
> > > in
> > > > > established mode?
> > > > >
> > > > > Thanks,
> > > > >
> > > > > Jun
> > > > >
> > > > >
> > > > > On Tue, Jan 21, 2014 at 8:06 AM, Ahmy Yulrizka <ahmy@yulrizka.com>
> > > > wrote:
> > > > >
> > > > > > this is the the line i copied on lsof
> > > > > >
> > > > > > ...
> > > > > > java      11818      kafka   98u     sock                0,7
> > > 0t0
> > > > > >  615628183 can't identify protocol
> > > > > > java      11818      kafka   99u     IPv4          615077352
> > > 0t0
> > > > > >    TCP somedomain.com:9092->
> 121-123-123-123.someprovider.net:37547
> > > > > > (CLOSE_WAIT)
> > > > > > java      11818      kafka  100u     IPv4          615077353
> > > 0t0
> > > > > >    TCP somedomain.com:9092->
> 121-123-123-123.someprovider.net:37553
> > > > > > (ESTABLISHED)
> > > > > > java      11818      kafka  101u     sock                0,7
> > > 0t0
> > > > > >  615628184 can't identify protocol
> > > > > > java      11818      kafka  102u     sock                0,7
> > > 0t0
> > > > > >  615628185 can't identify protocol
> > > > > > java      11818      kafka  103u     sock                0,7
> > > 0t0
> > > > > >  615628186 can't identify protocol
> > > > > > ...
> > > > > >
> > > > > > as you can see, from the output, i could see the connection
state
> > on
> > > > some
> > > > > > of the TCP, but the sock only gives information "can't identify
> > > > protocol"
> > > > > > so I could not see where or from this sock is originating
> > > > > >
> > > > > > I could not see the connection also when i run netstat -nat
> > > > > >
> > > > > > --
> > > > > > Ahmy Yulrizka
> > > > > > http://ahmy.yulrizka.com
> > > > > > @yulrizka
> > > > > >
> > > > > >
> > > > > > On Tue, Jan 21, 2014 at 4:42 PM, Jun Rao <junrao@gmail.com>
> wrote:
> > > > > >
> > > > > > > What mode are those sockets in (established, closed, etc)?
> Also,
> > > from
> > > > > the
> > > > > > > ip, could you tell whether those sockets are from the client
or
> > > from
> > > > > the
> > > > > > > replica fetcher in the brokers.
> > > > > > >
> > > > > > > Thanks,
> > > > > > >
> > > > > > > Jun
> > > > > > >
> > > > > > >
> > > > > > > On Tue, Jan 21, 2014 at 3:29 AM, Ahmy Yulrizka <
> > ahmy@yulrizka.com>
> > > > > > wrote:
> > > > > > >
> > > > > > > > We are running 3 kafka nodes, which servers 4 partition.
> > > > > > > > We have been experiencing weird behavior during network
> outage.
> > > > > > > >
> > > > > > > > we had been experiencing twice in the last couple
of days.
> the
> > > > > previous
> > > > > > > one
> > > > > > > > took down all of the cluster.
> > > > > > > > while this one only 2 out of 3 survive. and 1 node
became the
> > > > leader
> > > > > of
> > > > > > > all
> > > > > > > > partition, and other node only in ISR of 1 partition
(out of
> 4)
> > > > > > > >
> > > > > > > > my best guess now is that when the network down, the
broker
> > can't
> > > > > > connect
> > > > > > > > to other broker to do replication and keep opening
the socket
> > > > > > > > without closing it. But I'm not entirely sure about
this.
> > > > > > > >
> > > > > > > > Is there any way to mitigate the problem ? or is there
any
> > > > > > configuration
> > > > > > > > options to stop this from happening again ?
> > > > > > > >
> > > > > > > >
> > > > > > > > The java/kafka process open too many socket file descriptor.
> > > > > > > > running `lsof -a -p 11818` yield thousand of this
line.
> > > > > > > >
> > > > > > > > ...
> > > > > > > > java    11818 kafka 3059u  sock                0,7
      0t0
> > > > > 615637305
> > > > > > > > can't identify protocol
> > > > > > > > java    11818 kafka 3060u  sock                0,7
      0t0
> > > > > 615637306
> > > > > > > > can't identify protocol
> > > > > > > > java    11818 kafka 3061u  sock                0,7
      0t0
> > > > > 615637307
> > > > > > > > can't identify protocol
> > > > > > > > java    11818 kafka 3062u  sock                0,7
      0t0
> > > > > 615637308
> > > > > > > > can't identify protocol
> > > > > > > > java    11818 kafka 3063u  sock                0,7
      0t0
> > > > > 615637309
> > > > > > > > can't identify protocol
> > > > > > > > java    11818 kafka 3064u  sock                0,7
      0t0
> > > > > 615637310
> > > > > > > > can't identify protocol
> > > > > > > > java    11818 kafka 3065u  sock                0,7
      0t0
> > > > > 615637311
> > > > > > > > can't identify protocol
> > > > > > > > ...
> > > > > > > >
> > > > > > > > i verify that the the open socket did not close when
i
> repeated
> > > the
> > > > > > > command
> > > > > > > > after 2 minutes.
> > > > > > > >
> > > > > > > >
> > > > > > > > and the kafka log on the broken node, generate lots
of error
> > like
> > > > > this:
> > > > > > > >
> > > > > > > > [2014-01-21 04:21:48,819]  64573925 [kafka-acceptor]
ERROR
> > > > > > > > kafka.network.Acceptor  - Error in acceptor
> > > > > > > > java.io.IOException: Too many open files
> > > > > > > >         at sun.nio.ch.ServerSocketChannelImpl.accept0(Native
> > > > Method)
> > > > > > > >         at
> > > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> sun.nio.ch.ServerSocketChannelImpl.accept(ServerSocketChannelImpl.java:165)
> > > > > > > >         at
> > kafka.network.Acceptor.accept(SocketServer.scala:200)
> > > > > > > >         at kafka.network.Acceptor.run(SocketServer.scala:154)
> > > > > > > >         at java.lang.Thread.run(Thread.java:701)
> > > > > > > > [2014-01-21 04:21:48,819]  64573925 [kafka-acceptor]
ERROR
> > > > > > > > kafka.network.Acceptor  - Error in acceptor
> > > > > > > > java.io.IOException: Too many open files
> > > > > > > >         at sun.nio.ch.ServerSocketChannelImpl.accept0(Native
> > > > Method)
> > > > > > > >         at
> > > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> sun.nio.ch.ServerSocketChannelImpl.accept(ServerSocketChannelImpl.java:165)
> > > > > > > >         at
> > kafka.network.Acceptor.accept(SocketServer.scala:200)
> > > > > > > >         at kafka.network.Acceptor.run(SocketServer.scala:154)
> > > > > > > >         at java.lang.Thread.run(Thread.java:701)
> > > > > > > > [2014-01-21 04:21:48,811]  64573917
> [ReplicaFetcherThread-0-1]
> > > INFO
> > > > > > > >  kafka.consumer.SimpleConsumer  - Reconnect due to
socket
> > error:
> > > > null
> > > > > > > > [2014-01-21 04:21:48,819]  64573925
> [ReplicaFetcherThread-0-1]
> > > WARN
> > > > > > > >  kafka.server.ReplicaFetcherThread  -
> > [ReplicaFetcherThread-0-1],
> > > > > Error
> > > > > > > in
> > > > > > > > fetch Name: FetchRequest; Version: 0; CorrelationId:
> 74930218;
> > > > > > ClientId:
> > > > > > > > ReplicaFetcherThread-0-1; ReplicaId: 2; MaxWait: 500
ms;
> > > MinBytes:
> > > > 1
> > > > > > > bytes;
> > > > > > > > RequestInfo: [some-topic,0] ->
> > > > > > > > PartitionFetchInfo(959825,1048576),[some-topic,3]
->
> > > > > > > > PartitionFetchInfo(551546,1048576)
> > > > > > > > java.net.SocketException: Too many open files
> > > > > > > >         at sun.nio.ch.Net.socket0(Native Method)
> > > > > > > >         at sun.nio.ch.Net.socket(Net.java:156)
> > > > > > > >         at
> > > > > > > sun.nio.ch.SocketChannelImpl.<init>(SocketChannelImpl.java:102)
> > > > > > > >         at
> > > > > > > >
> > > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> sun.nio.ch.SelectorProviderImpl.openSocketChannel(SelectorProviderImpl.java:55)
> > > > > > > >         at
> > > > > java.nio.channels.SocketChannel.open(SocketChannel.java:122)
> > > > > > > >         at
> > > > > > > kafka.network.BlockingChannel.connect(BlockingChannel.scala:48)
> > > > > > > >         at
> > > > > > kafka.consumer.SimpleConsumer.connect(SimpleConsumer.scala:44)
> > > > > > > >         at
> > > > > > >
> kafka.consumer.SimpleConsumer.reconnect(SimpleConsumer.scala:57)
> > > > > > > >         at
> > > > > > > >
> > > > kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.scala:79)
> > > > > > > >         at
> > > > > > > >
> > > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:71)
> > > > > > > >         at
> > > > > > > >
> > > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SimpleConsumer.scala:110)
> > > > > > > >         at
> > > > > > > >
> > > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:110)
> > > > > > > >         at
> > > > > > > >
> > > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:110)
> > > > > > > >         at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> > > > > > > >         at
> > > > > > > >
> > > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply$mcV$sp(SimpleConsumer.scala:109)
> > > > > > > >         at
> > > > > > > >
> > > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:109)
> > > > > > > >         at
> > > > > > > >
> > > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:109)
> > > > > > > >         at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> > > > > > > >         at
> > > > > > kafka.consumer.SimpleConsumer.fetch(SimpleConsumer.scala:108)
> > > > > > > >         at
> > > > > > > >
> > > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> kafka.server.AbstractFetcherThread.processFetchRequest(AbstractFetcherThread.scala:94)
> > > > > > > >         at
> > > > > > > >
> > > > > >
> > > >
> > kafka.server.AbstractFetcherThread.doWork(AbstractFetcherThread.scala:86)
> > > > > > > >         at
> > > > > > > kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:51)
> > > > > > > >
> > > > > > > >
> > > > > > > > --
> > > > > > > > Ahmy Yulrizka
> > > > > > > > http://ahmy.yulrizka.com
> > > > > > > > @yulrizka
> > > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
>



-- 


*Best RegardsXiang Helin*

Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message