You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@kafka.apache.org by Priya Matpadi <pr...@ecofactor.com> on 2014/02/13 03:28:19 UTC
Re: Possibly leaking socket on ReplicaFetcherThread
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 <ju...@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 <ah...@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 <ju...@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 <ah...@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 <ju...@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 <ah...@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
> > > > > >
> > > > >
> > > >
> > >
> >
>
Re: Possibly leaking socket on ReplicaFetcherThread
Posted by Helin Xiang <xk...@gmail.com>.
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 <ju...@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
> <pr...@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 <ju...@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 <ah...@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 <ju...@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 <ah...@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 <ju...@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*
Re: Possibly leaking socket on ReplicaFetcherThread
Posted by Jun Rao <ju...@gmail.com>.
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
<pr...@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 <ju...@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 <ah...@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 <ju...@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 <ah...@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 <ju...@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
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
>
Re: Possibly leaking socket on ReplicaFetcherThread
Posted by Priya Matpadi <pr...@ecofactor.com>.
Further more, the problem is not just restricted to ReplicaFetcherThread.
Kafka consumer server also leaks sockets due to SendThread using same code
. See below stack trace:
2014-01-23 06:48:09,699 INFO [org.apache.zookeeper.ClientCnxn]
(OurKafkaMessageFetcher-blah1-SendThread(pkafka3.our.com:2181)) Opening
socket connection to server pkafka2.our.com/10.58.0.191:2181
2014-01-23 06:48:10,124 INFO [org.apache.zookeeper.ClientCnxn]
(OurKafkaMessageFetcher-blah2-SendThread(pkafka1.our.com:2181)) Opening
socket connection to server pkafka2.our.com/10.58.0.191:2181
2014-01-23 06:48:10,272 INFO [org.apache.zookeeper.ClientCnxn]
(OurKafkaMessageFetcher-blah3-SendThread(pkafka2.our.com:2181)) Opening
socket connection to server 10.58.0.190/10.58.0.190:2181
2014-01-23 06:48:11,033 INFO [org.apache.zookeeper.ClientCnxn]
(OurKafkaMessageFetcher-blah1-SendThread(pkafka2.our.com:2181)) Client
session timed out, have not heard from server in 3257ms for sessionid
0x3436ae2b16a0071, closing socket connection and attempting reconnect
2014-01-23 06:48:11,070 WARN [kafka.consumer.ConsumerFetcherThread]
(ConsumerFetcherThread-blah1KafkaGroup_pjb1.our.com-1390441665650-635d2718-0-2)
[ConsumerFetcherThread-blah1KafkaGroup_pjb1.our.com-1390441665650-635d2718-0-2],
Error in fetch Name: FetchRequest; Version: 0; CorrelationId: 176434;
ClientId:
blah1KafkaGroup-ConsumerFetcherThread-blah1KafkaGroup_pjb1.our.com-1390441665650-635d2718-0-2;
ReplicaId: -1; MaxWait: 100
ms; MinBytes: 1 bytes; RequestInfo: [blah,0] ->
PartitionFetchInfo(0,1048576),[blah,4] -> PartitionFetchInfo(0,1048576)
java.nio.channels.UnresolvedAddressException
at sun.nio.ch.Net.checkAddress(Net.java:30)
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)