We allocate a new BoundedByteBufferReceive for every fetch request. Are you
using SimpleConsumer directly? It seems it's started by the high level
consumer through the FetchFetcher thread.
Thanks,
Jun
On Mon, Sep 22, 2014 at 11:41 AM, Jagbir Hooda <jho...@gmail.com> wrote:
> Note: Re-posting the older message from another account due to
> formatting issues.
>
>
> Folks,
>
> Recently in one of our SimpleConsumer based client applications (0.8.1.1),
> we spotted a very busy CPU with almost no traffic in/out from the client
> and Kafka broker (1broker+1zookeeper) (the stack trace is attached at the
> end).
>
> The busy thread was invoked in a while loop anchored at the readFrom
> function
>
> ---scala/kafka/network/Transmission.scala:55-59-----
> .....
> while(!complete) {
> val read = readFrom(channel)
> trace(read + " bytes read.")
> totalRead += read
> }
> ....
> ----------------------------------------------------------------------
>
> The readFrom funtion and the associated thread was RUNNABLE at
> "read += Utils.read(channel, sizeBuffer)" (see below)
>
> ---scala/kafka/network/BoundedByteBufferReceive.scala:49-95--------
> ....
> def readFrom(channel: ReadableByteChannel): Int = {
> expectIncomplete()
> var read = 0
> // have we read the request size yet?
> if(sizeBuffer.remaining > 0)
> read += Utils.read(channel, sizeBuffer)
> // have we allocated the request buffer yet?
> if(contentBuffer == null && !sizeBuffer.hasRemaining) {
> sizeBuffer.rewind()
> val size = sizeBuffer.getInt()
> if(size <= 0)
> throw new InvalidRequestException("%d is not a valid request
> size.".format(size))
> if(size > maxSize)
> throw new InvalidRequestException("Request of length %d is not
> valid, it is larger than the maximum size of %d bytes.".format(size,
> maxSize))
> contentBuffer = byteBufferAllocate(size)
> }
> // if we have a buffer read some stuff into it
> if(contentBuffer != null) {
> read = Utils.read(channel, contentBuffer)
> // did we get everything?
> if(!contentBuffer.hasRemaining) {
> contentBuffer.rewind()
> complete = true
> }
> }
> read
> }
> .....
>
> ------------------------------------------------------------------------------
>
> It looks like contentBuffer size is initialized only once in
> SimpleConsumer life-cycle (we keep
> SimpleConsumer alive until the app is restarted).
>
> Wondering what's the communication pattern between the client and broker.
>
> Is our assumption that contentBuffer 'size' is only negotiated when
> SimpleConsumer
> is created true?
>
> If above is true then why the thread is BUSY at "read +=
> Utils.read(channel, sizeBuffer)"?
> (Our application was running fine for over three days and on the third
> day we noticed
> the busy CPU and this behavior.)
>
> If contentBuffer size is negotiated every message (or more frequently)
> then contentBuffer
> variable must be set to "null" somewhere to allow it to reconfigured
> (as per existing codebase).
>
> Another question is about the broker configuration setting which
> controls the contentBuffer size.
>
> I'll really appreciate if someone can help us in figuring it out.
>
> Thanks,
> jaguar
>
> ----------------BUSY THREAD--------------------------------
> "ConsumerFetcherThread-aaaa_ubuntu-1410940574038-305417a6-0-0" prio=10
> tid=0x00007f7e4c04e800 nid=0x738c runnable [0x00007f7e4a3e1000]
> java.lang.Thread.State: RUNNABLE
> at sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)
> at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
> at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:79)
> at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:87)
> - locked <0x00000000e583e670> (a sun.nio.ch.Util$2)
> - locked <0x00000000e583e660> (a java.util.Collections$UnmodifiableSet)
> - locked <0x00000000e583e1d8> (a sun.nio.ch.EPollSelectorImpl)
> at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:98)
> at sun.nio.ch.SocketAdaptor$SocketInputStream.read(SocketAdaptor.java:221)
> - locked <0x00000000e59b69a8> (a java.lang.Object)
> at sun.nio.ch.ChannelInputStream.read(ChannelInputStream.java:103)
> - locked <0x00000000e59b6b30> (a
> sun.nio.ch.SocketAdaptor$SocketInputStream)
> at
> java.nio.channels.Channels$ReadableByteChannelImpl.read(Channels.java:385)
> - locked <0x00000000e59cc4f0> (a java.lang.Object)
> at kafka.utils.Utils$.read(Utils.scala:375)
> at
> kafka.network.BoundedByteBufferReceive.readFrom(BoundedByteBufferReceive.scala:54)
> at kafka.network.Receive$class.readCompletely(Transmission.scala:56)
> at
> kafka.network.BoundedByteBufferReceive.readCompletely(BoundedByteBufferReceive.scala:29)
> at kafka.network.BlockingChannel.receive(BlockingChannel.scala:100)
> at kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.scala:73)
> at
> kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:71)
> - locked <0x00000000e591cb68> (a java.lang.Object)
> at
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SimpleConsumer.scala:109)
> at
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:109)
> at
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:109)
> at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> at
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply$mcV$sp(SimpleConsumer.scala:108)
> at
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:108)
> at
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:108)
> at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> at kafka.consumer.SimpleConsumer.fetch(SimpleConsumer.scala:107)
> at
> kafka.server.AbstractFetcherThread.processFetchRequest(AbstractFetcherThread.scala:96)
> at
> kafka.server.AbstractFetcherThread.doWork(AbstractFetcherThread.scala:88)
> at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:51)
>
> ------------------------------------8<----------------------------------------
>
