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<----------------------------------------
