In case the attachments don't work out here is an imgur link -
http://imgur.com/NslGpT3,Uw6HFow#0
On Mon, Dec 29, 2014 at 3:13 PM, Rajiv Kurian <ra...@signalfuse.com> wrote:
> Never mind about (2). I see these stats are already being output by the
> kafka producer. I've attached a couple of screenshots (couldn't copy paste
> from jvisualvm ). Do any of these things strike as odd? The
> bufferpool-wait-ratio sadly shows up as a NaN.
>
> I still don't know how to figure out (1).
>
> Thanks!
>
>
> On Mon, Dec 29, 2014 at 3:02 PM, Rajiv Kurian <ra...@signalfuse.com>
> wrote:
>
>> Hi Jay,
>>
>> Re (1) - I am not sure how to do this? Actually I am not sure what this
>> means. Is this the time every write/fetch request is received on the
>> broker? Do I need to enable some specific log level for this to show up? It
>> doesn't show up in the usual log. Is this information also available via
>> jmx somehow?
>> Re (2) - Are you saying that I should instrument the "percentage of time
>> waiting for buffer space" stat myself? If so how do I do this. Or are these
>> stats already output to jmx by the kafka producer code. This seems like
>> it's in the internals of the kafka producer client code.
>>
>> Thanks again!
>>
>>
>> On Mon, Dec 29, 2014 at 10:22 AM, Rajiv Kurian <ra...@signalfuse.com>
>> wrote:
>>
>>> 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!
>>>> > > > >>>
>>>> > > > >>>
>>>> > > >
>>>> > >
>>>> >
>>>>
>>>
>>>
>>
>
