Thanks Jay. Will check (1) and (2) and get back to you. The test is not
stand-alone now. It might be a bit of work to extract it to a stand-alone
executable. It might take me a bit of time to get that going.
On Mon, Dec 29, 2014 at 9:45 AM, Jay Kreps <j...@confluent.io> wrote:
> Hey Rajiv,
>
> This sounds like a bug. The more info you can help us get the easier to
> fix. Things that would help:
> 1. Can you check if the the request log on the servers shows latency spikes
> (in which case it is a server problem)?
> 2. It would be worth also getting the jmx stats on the producer as they
> will show things like what percentage of time it is waiting for buffer
> space etc.
>
> If your test is reasonably stand-alone it would be great to file a JIRA and
> attach the test code and the findings you already have so someone can dig
> into what is going on.
>
> -Jay
>
> On Sun, Dec 28, 2014 at 7:15 PM, Rajiv Kurian <ra...@signalfuse.com>
> wrote:
>
> > Hi all,
> >
> > Bumping this up, in case some one has any ideas. I did yet another
> > experiment where I create 4 producers and stripe the send requests across
> > them in a manner such that any one producer only sees 256 partitions
> > instead of the entire 1024. This seems to have helped a bit, and though I
> > still see crazy high 99th (25-30 seconds), the median, mean, 75th and
> 95th
> > percentile have all gone down.
> >
> > Thanks!
> >
> > On Sun, Dec 21, 2014 at 12:27 PM, Thunder Stumpges <tstump...@ntent.com>
> > wrote:
> >
> > > Ah I thought it was restarting the broker that made things better :)
> > >
> > > Yeah I have no experience with the Java client so can't really help
> > there.
> > >
> > > Good luck!
> > >
> > > -----Original Message-----
> > > From: Rajiv Kurian [ra...@signalfuse.com]
> > > Received: Sunday, 21 Dec 2014, 12:25PM
> > > To: users@kafka.apache.org [users@kafka.apache.org]
> > > Subject: Re: Trying to figure out kafka latency issues
> > >
> > > I'll take a look at the GC profile of the brokers Right now I keep a
> tab
> > on
> > > the CPU, Messages in, Bytes in, Bytes out, free memory (on the machine
> > not
> > > JVM heap) free disk space on the broker. I'll need to take a look at
> the
> > > JVM metrics too. What seemed strange is that going from 8 -> 512
> > partitions
> > > increases the latency, but going fro 512-> 8 does not decrease it. I
> have
> > > to restart the producer (but not the broker) for the end to end latency
> > to
> > > go down That made it seem that the fault was probably with the
> producer
> > > and not the broker. Only restarting the producer made things better.
> I'll
> > > do more extensive measurement on the broker.
> > >
> > > On Sun, Dec 21, 2014 at 9:08 AM, Thunder Stumpges <tstump...@ntent.com
> >
> > > wrote:
> > > >
> > > > Did you see my response and have you checked the server logs
> especially
> > > > the GC logs? It still sounds like you are running out of memory on
> the
> > > > broker. What is your max heap memory and are you thrashing once you
> > start
> > > > writing to all those partitions?
> > > >
> > > > You have measured very thoroughly from an external point of view, i
> > think
> > > > now you'll have to start measuring the internal metrics. Maybe
> someone
> > > else
> > > > will have ideas on what jmx values to watch.
> > > >
> > > > Best,
> > > > Thunder
> > > >
> > > >
> > > > -----Original Message-----
> > > > From: Rajiv Kurian [ra...@signalfuse.com]
> > > > Received: Saturday, 20 Dec 2014, 10:24PM
> > > > To: users@kafka.apache.org [users@kafka.apache.org]
> > > > Subject: Re: Trying to figure out kafka latency issues
> > > >
> > > > Some more work tells me that the end to end latency numbers vary with
> > the
> > > > number of partitions I am writing to. I did an experiment, where
> based
> > > on a
> > > > run time flag I would dynamically select how many of the *1024
> > > partitions*
> > > > I write to. So say I decide I'll write to at most 256 partitions I
> mod
> > > > whatever partition I would actually write to by 256. Basically the
> > number
> > > > of partitions for this topic on the broker remains the same at *1024*
> > > > partitions but the number of partitions my producers write to changes
> > > > dynamically based on a run time flag. So something like this:
> > > >
> > > > int partition = getPartitionForMessage(message);
> > > > int maxPartitionsToWriteTo = maxPartitionsFlag.get(); // This flag
> > can
> > > be
> > > > updated without bringing the application down - just a volatile read
> of
> > > > some number set externally.
> > > > int moddedPartition = partition % maxPartitionsToWrite.
> > > > // Send a message to this Kafka partition.
> > > >
> > > > Here are some interesting things I've noticed:
> > > >
> > > > i) When I start my client and it *never writes* to more than *8
> > > > partitions *(same
> > > > data rate but fewer partitions) - the end to end *99th latency is
> > 300-350
> > > > ms*. Quite a bit of this (numbers in my previous emails) is the
> latency
> > > > from producer -> broker and the latency from broker -> consumer.
> Still
> > > > nowhere as poor as the *20 - 30* seconds I was seeing.
> > > >
> > > > ii) When I increase the maximum number of partitions, end to end
> > latency
> > > > increases dramatically. At *256 partitions* the end to end *99th
> > latency
> > > is
> > > > still 390 - 418 ms.* Worse than the latency figures for *8
> *partitions,
> > > but
> > > > not by much. When I increase this number to *512 partitions *the end
> > > > to end *99th
> > > > latency *becomes an intolerable *19-24 seconds*. At *1024* partitions
> > the
> > > > *99th
> > > > latency is at 25 - 30 seconds*.
> > > > A table of the numbers:
> > > >
> > > > Max number of partitions written to (out of 1024)
> > > >
> > > > End to end latency
> > > >
> > > > 8
> > > >
> > > > 300 - 350 ms
> > > >
> > > > 256
> > > >
> > > > 390 - 418 ms
> > > >
> > > > 512
> > > >
> > > > 19 - 24 seconds
> > > >
> > > > 1024
> > > >
> > > > 25 - 30 seconds
> > > >
> > > >
> > > > iii) Once I make the max number of partitions high enough, reducing
> it
> > > > doesn't help. For eg: If I go up from *8* to *512 *partitions, the
> > > latency
> > > > goes up. But while the producer is running if I go down from *512*
> to
> > > > *8 *partitions,
> > > > it doesn't reduce the latency numbers. My guess is that the producer
> is
> > > > creating some state lazily per partition and this state is causing
> the
> > > > latency. Once this state is created, writing to fewer partitions
> > doesn't
> > > > seem to help. Only a restart of the producer calms things down.
> > > >
> > > > So my current plan is to reduce the number of partitions on the
> topic,
> > > but
> > > > there seems to be something deeper going on for the latency numbers
> to
> > be
> > > > so poor to begin with and then suffer so much more (non linearly)
> with
> > > > additional partitions.
> > > >
> > > > Thanks!
> > > >
> > > > On Sat, Dec 20, 2014 at 6:03 PM, Rajiv Kurian <ra...@signalfuse.com>
> > > > wrote:
> > > > >
> > > > > I've done some more measurements. I've also started measuring the
> > > latency
> > > > > between when I ask my producer to send a message and when I get an
> > > > > acknowledgement via the callback. Here is my code:
> > > > >
> > > > > // This function is called on every producer once every 30 seconds.
> > > > >
> > > > > public void addLagMarkers(final Histogram enqueueLag) {
> > > > >
> > > > > final int numberOfPartitions = 1024;
> > > > >
> > > > > final long timeOfEnqueue = System.currentTimeMillis();
> > > > >
> > > > > final Callback callback = new Callback() {
> > > > >
> > > > > @Override
> > > > >
> > > > > public void onCompletion(RecordMetadata metadata,
> > Exception
> > > > ex)
> > > > > {
> > > > >
> > > > > if (metadata != null) {
> > > > >
> > > > > // The difference between ack time from broker
> > and
> > > > > enqueue time.
> > > > >
> > > > > final long timeOfAck =
> > System.currentTimeMillis();
> > > > >
> > > > > final long lag = timeOfAck - timeOfEnqueue;
> > > > >
> > > > > enqueueLag.update(lag);
> > > > >
> > > > > }
> > > > >
> > > > > }
> > > > >
> > > > > };
> > > > >
> > > > > for (int i = 0; i < numberOfPartitions; i++) {
> > > > >
> > > > > try {
> > > > >
> > > > > byte[] value = LagMarker.serialize(timeOfEnqueue);
> > //
> > > 10
> > > > > bytes -> short version + long timestamp.
> > > > >
> > > > > // This message is later used by the consumers to
> > > measure
> > > > > lag.
> > > > >
> > > > > ProducerRecord record = new
> ProducerRecord(MY_TOPIC,
> > i,
> > > > > null, value);
> > > > >
> > > > > kafkaProducer.send(record, callback);
> > > > >
> > > > > } catch (Exception e) {
> > > > >
> > > > > // We just dropped a lag marker.
> > > > >
> > > > > }
> > > > >
> > > > > }
> > > > >
> > > > > }
> > > > >
> > > > > The* 99th* on this lag is about* 350 - 400* ms. It's not stellar,
> but
> > > > > doesn't account for the *20-30 second 99th* I see on the end to end
> > > lag.
> > > > > I am consuming in a tight loop on the Consumers (using the
> > > > SimpleConsumer)
> > > > > with minimal processing with a *99th fetch time *of *130-140* ms,
> so
> > I
> > > > > don't think that should be a problem either. Completely baffled.
> > > > >
> > > > >
> > > > > Thanks!
> > > > >
> > > > >
> > > > >
> > > > > On Sat, Dec 20, 2014 at 5:51 PM, Rajiv Kurian <
> ra...@signalfuse.com>
> > > > > wrote:
> > > > >>
> > > > >>
> > > > >>
> > > > >> On Sat, Dec 20, 2014 at 3:49 PM, Rajiv Kurian <
> ra...@signalfuse.com
> > >
> > > > >> wrote:
> > > > >>>
> > > > >>> I am trying to replace a Thrift peer to peer API with kafka for a
> > > > >>> particular work flow. I am finding the 99th percentile latency to
> > be
> > > > >>> unacceptable at this time. This entire work load runs in an
> Amazon
> > > > VPC. I'd
> > > > >>> greatly appreciate it if some one has any insights on why I am
> > seeing
> > > > such
> > > > >>> poor numbers. Here are some details and measurements taken:
> > > > >>>
> > > > >>> i) I have a single topic with 1024 partitions that I am writing
> to
> > > from
> > > > >>> six clients using the kafka 0.8.2 beta kafka producer.
> > > > >>>
> > > > >>> ii) I have 3 brokers, each on a c3 2x machine on ec2. Each of
> > those
> > > > >>> machines has 8 virtual cpus, 15 GB memory and 2 * 80 GB SSDs. The
> > > > broker ->
> > > > >>> partitions mapping was decided by Kafka when I created the topic.
> > > > >>>
> > > > >>> iii) I write about 22 thousand messages per second from across
> the
> > 6
> > > > >>> clients. This number was calculated using distributed counters. I
> > > just
> > > > >>> increment a distributed counter in the callback from my enqueue
> job
> > > if
> > > > the
> > > > >>> metadata returned is not null. I also increment a number o
> dropped
> > > > messages
> > > > >>> counter if the callback has a non-null exception or if there was
> an
> > > > >>> exception in the synchronous send call. The number of dropped
> > > messages
> > > > is
> > > > >>> pretty much always zero. Out of the 6 clients 3 are responsible
> for
> > > > 95% of
> > > > >>> the traffic. Messages are very tiny and have null keys and 27
> byte
> > > > values
> > > > >>> (2 byte version and 25 byte payload). Again these messages are
> > > written
> > > > >>> using the kafka 0.8.2 client.
> > > > >>>
> > > > >>> iv) I have 3 consumer processes consuming only from this topic.
> > Each
> > > > >>> consumer process is assigned a disjoint set of the 1024
> partitions
> > by
> > > > an
> > > > >>> eternal arbiter. Each consumer process then creates a mapping
> from
> > > > brokers
> > > > >>> -> partitions it has been assigned. It then starts one fetcher
> > thread
> > > > per
> > > > >>> broker. Each thread queries the broker (using the SimpleConsumer)
> > it
> > > > has
> > > > >>> been assigned for partitions such that partitions = (partitions
> on
> > > > broker)
> > > > >>> *∩ *(partitions assigned to the process by the arbiter). So in
> > effect
> > > > >>> there are 9 consumer threads across 3 consumer processes that
> > query a
> > > > >>> disjoint set of partitions for the single topic and amongst
> > > themselves
> > > > they
> > > > >>> manage to consume from every topic. So each thread has a run loop
> > > > where it
> > > > >>> asks for messages, consumes them then asks for more messages.
> When
> > a
> > > > run
> > > > >>> loop starts, it starts by querying for the latest message in a
> > > > partition
> > > > >>> (i.e. discards any previous backup) and then maintains a map of
> > > > partition
> > > > >>> -> nextOffsetToRequest in memory to make sure that it consumes
> > > > messages in
> > > > >>> order.
> > > > >>>
> > > > >> Edit: Mean't external arbiter.
> > > > >>
> > > > >>>
> > > > >>> v) Consumption is really simple. Each message is put on a non
> > > blocking
> > > > >>> efficient ring buffer. If the ring buffer is full the message is
> > > > dropped. I
> > > > >>> measure the mean time between fetches and it is the same as the
> > time
> > > > for
> > > > >>> fetches up to a ms, meaning no matter how many messages are
> > dequeued,
> > > > it
> > > > >>> takes almost no time to process them. At the end of processing a
> > > fetch
> > > > >>> request I increment another distributed counter that counts the
> > > number
> > > > of
> > > > >>> messages processed. This counter tells me that on average I am
> > > > consuming
> > > > >>> the same number of messages/sec that I enqueue on the producers
> > i.e.
> > > > around
> > > > >>> 22 thousand messages/sec.
> > > > >>>
> > > > >>> vi) The 99th percentile of the number of messages fetched per
> fetch
> > > > >>> request is about 500 messages. The 99th percentile of the time
> it
> > > > takes to
> > > > >>> fetch a batch is abut 130 - 140 ms. I played around with the
> buffer
> > > and
> > > > >>> maxWait settings on the SimpleConsumer and attempting to consume
> > more
> > > > >>> messages was leading the 99th percentile of the fetch time to
> > balloon
> > > > up,
> > > > >>> so I am consuming in smaller batches right now.
> > > > >>>
> > > > >>> vii) Every 30 seconds odd each of the producers inserts a trace
> > > message
> > > > >>> into each partition (so 1024 messages per producer every 30
> > seconds).
> > > > Each
> > > > >>> message only contains the System.currentTimeMillis() at the time
> of
> > > > >>> enqueueing. It is about 9 bytes long. The consumer in step (v)
> > always
> > > > >>> checks to see if a message it dequeued is of this trace type
> > (instead
> > > > of
> > > > >>> the regular message type). This is a simple check on the first 2
> > byte
> > > > >>> version that each message buffer contains. If it is a trace
> message
> > > it
> > > > >>> reads it's payload as a long and uses the difference between the
> > > system
> > > > >>> time on the consumer and this payload and updates a histogram
> with
> > > this
> > > > >>> difference value. Ignoring NTP skew (which I've measured to be in
> > the
> > > > order
> > > > >>> of milliseconds) this is the lag between when a message was
> > enqueued
> > > > on the
> > > > >>> producer and when it was read on the consumer.
> > > > >>>
> > > > >> Edit: Trace messages are 10 bytes long - 2byte version + 8 byte
> long
> > > for
> > > > >> timestamp.
> > > > >>
> > > > >> So pseudocode for every consumer thread (one per broker per
> consumer
> > > > >>> process (9 in total across 3 consumers) is:
> > > > >>>
> > > > >>> void run() {
> > > > >>>
> > > > >>> while (running) {
> > > > >>>
> > > > >>> FetchRequest fetchRequest = buildFetchRequest(
> > > > >>> partitionsAssignedToThisProcessThatFallOnThisBroker); // This
> > > > >>> assignment is done by an external arbiter.
> > > > >>>
> > > > >>> measureTimeBetweenFetchRequests(); // The 99th of this is
> > > > 130-140ms
> > > > >>>
> > > > >>> FetchResponse fetchResponse = fetchResponse(fetchRequest);
> //
> > > The
> > > > >>> 99th of this is 130-140ms, which is the same as the time between
> > > > fetches.
> > > > >>>
> > > > >>> processData(fetchResponse); // The 99th on this is 2-3 ms.
> > > > >>> }
> > > > >>> }
> > > > >>>
> > > > >>> void processData(FetchResponse response) {
> > > > >>> try (Timer.Context _ = processWorkerDataTimer.time() { // The
> > 99th
> > > > on
> > > > >>> this is 2-3 ms.
> > > > >>> for (Message message : response) {
> > > > >>> if (typeOf(message) == NORMAL_DATA) {
> > > > >>> enqueueOnNonBlockingRingBuffer(message); // Never
> blocks,
> > > > >>> drops message if ring buffer is full.
> > > > >>> } else if (typeOf(message) == TRACE_DATA) {
> > > > >>> long timeOfEnqueue = geEnqueueTime(message);
> > > > >>> long currentTime = System.currentTimeMillis();
> > > > >>> long difference = currentTime - timeOfEnqueue;
> > > > >>> lagHistogram.update(difference);
> > > > >>> }
> > > > >>> }
> > > > >>> }
> > > > >>> }
> > > > >>>
> > > > >>> viii) So the kicker is that this lag is really really high:
> > > > >>> Median Lag: *200 ms*. This already seems pretty high and I could
> > even
> > > > >>> discount some of it through NTP skew.
> > > > >>> Mean Lag: *2 - 4 seconds*! Also notice how mean is bigger than
> > median
> > > > >>> kind of telling us how the distribution has a big tail.
> > > > >>> 99th Lag: *20 seconds*!!
> > > > >>>
> > > > >>> I can't quite figure out the source of the lag. Here are some
> other
> > > > >>> things of note:
> > > > >>>
> > > > >>> 1) The brokers in general are at about 40-50% CPU but I see
> > > occasional
> > > > >>> spikes to more than 80% - 95%. This could probably be causing
> > issues.
> > > > I am
> > > > >>> wondering if this is down to the high number of partitions I have
> > and
> > > > how
> > > > >>> each partition ends up receiving not too many messages. 22k
> > messages
> > > > spread
> > > > >>> across 1024 partitions = only 21 odd messages/sec/partition. But
> > each
> > > > >>> broker should be receiving about 7500 messages/sec. It could
> also
> > be
> > > > down
> > > > >>> to the nature of these messages being tiny. I imagine that the
> > kafka
> > > > wire
> > > > >>> protocol is probably close to the 25 byte message payload. It
> also
> > > has
> > > > to
> > > > >>> do a few CRC32 calculations etc. But again given that these are
> > > C3.2x
> > > > >>> machines and the number of messages is so tiny I'd expect it to
> do
> > > > better.
> > > > >>> But again reading this article
> > > > >>> <
> > > >
> > >
> >
> https://engineering.linkedin.com/kafka/benchmarking-apache-kafka-2-million-writes-second-three-cheap-machines
> > > >
> > > > about
> > > > >>> 2 million messages/sec across three brokers makes it seem like
> 22k
> > 25
> > > > byte
> > > > >>> messages should be very easy. So my biggest suspicion right now
> is
> > > > that the
> > > > >>> large number of partitions is some how responsible for this
> > latency.
> > > > >>>
> > > > >>> 2) On the consumer given that the 99th time between fetches ==
> 99th
> > > > time
> > > > >>> of a fetch, I don't see how I could do any better. It's pretty
> > close
> > > to
> > > > >>> just getting batches of data, iterating through them and dropping
> > > them
> > > > on
> > > > >>> the floor. My requests on the consumer are built like this:
> > > > >>>
> > > > >>> FetchRequestBuilder builder = new
> > > > >>> FetchRequestBuilder().clientId(clientName)
> > > > >>>
> > > > >>> .maxWait(100).minBytes(1000); // max wait of 100
> > ms
> > > or
> > > > >>> at least 1000 bytes whichever happens first.
> > > > >>>
> > > > >>> // Add a fetch request for every partition.
> > > > >>>
> > > > >>> for (PartitionMetadata partition : partitions) {
> > > > >>>
> > > > >>> int partitionId = partition.partitionId();
> > > > >>>
> > > > >>> long offset = readOffsets.get(partition); // This
> map
> > > > >>> maintains the next partition to be read.
> > > > >>>
> > > > >>> builder.addFetch(MY_TOPIC, partitionId, offset,
> 3000);
> > > //
> > > > >>> Up to 3000 bytes per fetch request.}
> > > > >>> }
> > > > >>> 3) On the producer side I use the kafka beta producer. I've
> played
> > > > >>> around with many settings but none of them seem to have made a
> > > > difference.
> > > > >>> Here are my current settings:
> > > > >>>
> > > > >>> ProducerConfig values:
> > > > >>>
> > > > >>> block.on.buffer.full = false
> > > > >>>
> > > > >>> retry.backoff.ms = 100
> > > > >>>
> > > > >>> buffer.memory = 83886080 // Kind of big - could it be
> > adding
> > > > >>> lag?
> > > > >>>
> > > > >>> batch.size = 40500 // This is also kind of big - could
> it
> > be
> > > > >>> adding lag?
> > > > >>>
> > > > >>> metrics.sample.window.ms = 30000
> > > > >>>
> > > > >>> metadata.max.age.ms = 300000
> > > > >>>
> > > > >>> receive.buffer.bytes = 32768
> > > > >>>
> > > > >>> timeout.ms = 30000
> > > > >>>
> > > > >>> max.in.flight.requests.per.connection = 5
> > > > >>>
> > > > >>> metric.reporters = []
> > > > >>>
> > > > >>> bootstrap.servers = [broker1, broker2, broker3]
> > > > >>>
> > > > >>> client.id =
> > > > >>>
> > > > >>> compression.type = none
> > > > >>>
> > > > >>> retries = 0
> > > > >>>
> > > > >>> max.request.size = 1048576
> > > > >>>
> > > > >>> send.buffer.bytes = 131072
> > > > >>>
> > > > >>> acks = 1
> > > > >>>
> > > > >>> reconnect.backoff.ms = 10
> > > > >>>
> > > > >>> linger.ms = 250 // Based on this config I'd imagine
> that
> > > the
> > > > >>> lag should be bounded to about 250 ms over all.
> > > > >>>
> > > > >>> metrics.num.samples = 2
> > > > >>>
> > > > >>> metadata.fetch.timeout.ms = 60000
> > > > >>>
> > > > >>> I know this is a lot of information. I just wanted to provide as
> > much
> > > > >>> context as possible.
> > > > >>>
> > > > >>> Thanks in advance!
> > > > >>>
> > > > >>>
> > > >
> > >
> >
>
