I think I found where the problem is, how to solve and why, still not sure.
it related to disk (maybe flushing?). I did a single machine, single node, single topic and single partition setup. producer pub as 2000 message/s, 10K size message size. and single key. when I save kafka log to the memory based partition, I don't see a latency over 100ms. top around 70ms. when I save to a ssd hard drive. I do see latency spike, sometime over 1s. adjust the log.flush.inteval.message / log.flush.intefval.ms has impact, but only to make thing worse... need suggestion. I think log flushing is totally async and done by OS in the default setting. does kafka has to wait when flushing data to disk? Thanks, Nan On Wed, Aug 22, 2018 at 11:37 PM Guozhang Wang <wangg...@gmail.com> wrote: > Given your application code: > > ---------------------------- > > final KStream<String, NodeMutation> localDeltaStream = builder.stream( > > localDeltaTopic, > > Consumed.with( > > new Serdes.StringSerde(), > > new NodeMutationSerde<>() > > ) > > ); > > KStream<String, NodeState> localHistStream = localDeltaStream.mapValues( > > (mutation) -> NodeState > > .newBuilder() > > .setMeta( > > mutation.getMetaMutation().getMeta() > > ) > > .setValue( > > mutation.getValueMutation().getValue() > > ) > > .build() > > ); > > localHistStream.to( > > localHistTopic, > > Produced.with(new Serdes.StringSerde(), new NodeStateSerde<>()) > > ); > > ---------------------------- > > which is pure stateless, committing will not touch on an state directory at > all. Hence committing only involves committing offsets to Kafka. > > > Guozhang > > > On Wed, Aug 22, 2018 at 8:11 PM, Nan Xu <nanxu1...@gmail.com> wrote: > > > I was suspecting that too, but I also noticed the spike is not spaced > > around 10s. to further prove it. I put kafka data directory in a memory > > based directory. it still has such latency spikes. I am going to test > it > > on a single broker, single partition env. will report back soon. > > > > On Wed, Aug 22, 2018 at 5:14 PM Guozhang Wang <wangg...@gmail.com> > wrote: > > > > > Hello Nan, > > > > > > Thanks for the detailed information you shared. When Kafka Streams is > > > normally running, no rebalances should be triggered unless some of the > > > instances (in your case, docker containers) have soft failures. > > > > > > I suspect the latency spike is due to the commit intervals: streams > will > > > try to commit its offset at a regular paces, which may increase > latency. > > It > > > is controlled by the "commit.interval.ms" config value. I saw that in > > your > > > original email you've set it to 10 * 1000 (10 seconds). Is that aligned > > > with the frequency you observe latency spikes? > > > > > > > > > Guozhang > > > > > > > > > On Sun, Aug 19, 2018 at 10:41 PM, Nan Xu <nanxu1...@gmail.com> wrote: > > > > > > > did more test and and make the test case simple. > > > > all the setup now is a single physical machine. running 3 docker > > > instance. > > > > a1, a2, a3 > > > > > > > > kafka + zookeeper running on all of those docker containers. > > > > producer running on a1, send a single key, update speed 2000 > > message/s, > > > > each message is 10K size. > > > > 3 consumer(different group) are running. one on each docker. > > > > all topics are pre-created. > > > > in startup, I do see some latency greater than 100ms, which is fine. > > and > > > > then everything is good. latency is low and consumer don't see > anything > > > > over 100ms for a while. > > > > then I see a few messages have latency over 100ms. then back to > normal, > > > > then happen again..... do seems like gc problem. but I check the gc > > > log. I > > > > don't think it can cause over 100ms. (both are G1 collector) > > > > > > > > after the stream stable running( exclude the startup), the first > > message > > > > over 100ms take 179ms and the gc ( it has a 30ms pause, but should > not > > > > cause a 179ms end to end). > > > > > > > > FROM APP > > > > > > > > 2018-08-20T00:28:47.118-0500: 406.838: [GC (Allocation Failure) > > > > 3184739K->84018K(5947904K), 0.0093730 secs] > > > > > > > > 2018-08-20T00:28:56.094-0500: 415.814: [GC (Allocation Failure) > > > > 3184690K->84280K(6053888K), 0.0087473 secs] > > > > > > > > 2018-08-20T00:29:05.069-0500: 424.789: [GC (Allocation Failure) > > > > 3301176K->84342K(6061056K), 0.0127339 secs] > > > > > > > > 2018-08-20T00:29:14.234-0500: 433.954: [GC (Allocation Failure) > > > > 3301238K->84624K(6143488K), 0.0140844 secs] > > > > > > > > 2018-08-20T00:29:24.523-0500: 444.243: [GC (Allocation Failure) > > > > 3386000K->89949K(6144000K), 0.0108118 secs] > > > > > > > > > > > > > > > > kafka a1 > > > > > > > > 2018-08-20T00:29:14.306-0500: 7982.657: [GC pause (G1 Evacuation > Pause) > > > > (young), 0.0214200 secs] > > > > > > > > [Parallel Time: 17.2 ms, GC Workers: 8] > > > > > > > > [GC Worker Start (ms): Min: 7982657.5, Avg: 7982666.9, Max: > > > > 7982673.8, Diff: 16.3] > > > > > > > > [Ext Root Scanning (ms): Min: 0.0, Avg: 0.2, Max: 1.5, Diff: > 1.5, > > > > Sum: 1.5] > > > > > > > > [Update RS (ms): Min: 0.0, Avg: 1.0, Max: 6.5, Diff: 6.5, Sum: > > 8.4] > > > > > > > > [Processed Buffers: Min: 0, Avg: 4.6, Max: 13, Diff: 13, > Sum: > > > 37] > > > > > > > > [Scan RS (ms): Min: 0.0, Avg: 0.9, Max: 2.0, Diff: 2.0, Sum: > 7.1] > > > > > > > > [Code Root Scanning (ms): Min: 0.0, Avg: 0.0, Max: 0.0, Diff: > > 0.0, > > > > Sum: 0.0] > > > > > > > > [Object Copy (ms): Min: 0.0, Avg: 4.6, Max: 6.5, Diff: 6.5, > Sum: > > > > 36.5] > > > > > > > > [Termination (ms): Min: 0.0, Avg: 0.4, Max: 0.9, Diff: 0.9, > Sum: > > > 2.9] > > > > > > > > [Termination Attempts: Min: 1, Avg: 10.4, Max: 25, Diff: 24, > > > Sum: > > > > 83] > > > > > > > > [GC Worker Other (ms): Min: 0.0, Avg: 0.0, Max: 0.0, Diff: 0.0, > > > Sum: > > > > 0.1] > > > > > > > > [GC Worker Total (ms): Min: 0.2, Avg: 7.1, Max: 16.4, Diff: > 16.2, > > > > Sum: 56.5] > > > > > > > > [GC Worker End (ms): Min: 7982673.9, Avg: 7982674.0, Max: > > > 7982674.5, > > > > Diff: 0.6] > > > > > > > > [Code Root Fixup: 0.0 ms] > > > > > > > > [Code Root Purge: 0.0 ms] > > > > > > > > [Clear CT: 1.0 ms] > > > > > > > > [Other: 3.2 ms] > > > > > > > > [Choose CSet: 0.0 ms] > > > > > > > > [Ref Proc: 1.9 ms] > > > > > > > > [Ref Enq: 0.0 ms] > > > > > > > > [Redirty Cards: 0.8 ms] > > > > > > > > [Humongous Register: 0.1 ms] > > > > > > > > [Humongous Reclaim: 0.0 ms] > > > > > > > > [Free CSet: 0.2 ms] > > > > > > > > [Eden: 48.0M(48.0M)->0.0B(48.0M) Survivors: 3072.0K->3072.0K Heap: > > > > 265.5M(1024.0M)->217.9M(1024.0M)] > > > > > > > > [Times: user=0.05 sys=0.00, real=0.03 secs] > > > > > > > > 2018-08-20T00:29:16.074-0500: 7984.426: [GC pause (G1 Evacuation > Pause) > > > > (young), 0.0310004 secs] > > > > > > > > [Parallel Time: 24.4 ms, GC Workers: 8] > > > > > > > > [GC Worker Start (ms): Min: 7984426.1, Avg: 7984436.8, Max: > > > > 7984444.7, Diff: 18.6] > > > > > > > > [Ext Root Scanning (ms): Min: 0.0, Avg: 0.3, Max: 1.9, Diff: > 1.9, > > > > Sum: 2.0] > > > > > > > > [Update RS (ms): Min: 0.0, Avg: 4.1, Max: 11.8, Diff: 11.8, > Sum: > > > > 32.9] > > > > > > > > [Processed Buffers: Min: 0, Avg: 5.4, Max: 25, Diff: 25, > Sum: > > > 43] > > > > > > > > [Scan RS (ms): Min: 0.1, Avg: 3.2, Max: 11.3, Diff: 11.2, Sum: > > > 25.5] > > > > > > > > [Code Root Scanning (ms): Min: 0.0, Avg: 0.0, Max: 0.0, Diff: > > 0.0, > > > > Sum: 0.0] > > > > > > > > [Object Copy (ms): Min: 0.0, Avg: 4.1, Max: 6.9, Diff: 6.9, > Sum: > > > > 32.7] > > > > > > > > [Termination (ms): Min: 0.0, Avg: 0.8, Max: 1.6, Diff: 1.6, > Sum: > > > 6.8] > > > > > > > > [Termination Attempts: Min: 1, Avg: 5.4, Max: 11, Diff: 10, > > Sum: > > > > 43] > > > > > > > > [GC Worker Other (ms): Min: 0.0, Avg: 0.0, Max: 0.0, Diff: 0.0, > > > Sum: > > > > 0.1] > > > > > > > > [GC Worker Total (ms): Min: 4.4, Avg: 12.5, Max: 23.6, Diff: > > 19.1, > > > > Sum: 100.1] > > > > > > > > [GC Worker End (ms): Min: 7984449.1, Avg: 7984449.3, Max: > > > 7984449.9, > > > > Diff: 0.8] > > > > > > > > [Code Root Fixup: 0.0 ms] > > > > > > > > [Code Root Purge: 0.0 ms] > > > > > > > > [Clear CT: 1.1 ms] > > > > > > > > [Other: 5.5 ms] > > > > > > > > [Choose CSet: 0.0 ms] > > > > > > > > [Ref Proc: 2.2 ms] > > > > > > > > [Ref Enq: 0.0 ms] > > > > > > > > [Redirty Cards: 2.8 ms] > > > > > > > > [Humongous Register: 0.1 ms] > > > > > > > > [Humongous Reclaim: 0.0 ms] > > > > > > > > [Free CSet: 0.1 ms] > > > > > > > > [Eden: 48.0M(48.0M)->0.0B(48.0M) Survivors: 3072.0K->3072.0K Heap: > > > > 265.9M(1024.0M)->218.4M(1024.0M)] > > > > > > > > [Times: user=0.05 sys=0.00, real=0.03 secs] > > > > > > > > > > > > so when kafka stream running, is there any trying to rebalance? > either > > > > broker rebalance or client rebalance? > > > > any kind of test to see what cause the trouble? > > > > > > > > Thanks, > > > > Nan > > > > > > > > > > > > > > > > > > > > > > > > > > > > On Sun, Aug 19, 2018 at 10:34 PM Guozhang Wang <wangg...@gmail.com> > > > wrote: > > > > > > > > > Okay, so you're measuring end-to-end time from producer -> broker > -> > > > > > streams' consumer client, there are multiple phases that can > > contribute > > > > to > > > > > the 100ms latency, and I cannot tell if stream's consumer phase is > > the > > > > > major contributor. For example, if the topic was not created > before, > > > then > > > > > when the broker first received a produce request it may need to > > create > > > > the > > > > > topic, which involves multiple steps including writes to ZK which > > could > > > > > take time. > > > > > > > > > > There are some confusions from your description: you mentioned > "Kafka > > > > > cluster is already up and running", but I think you are referring > to > > > > "Kafka > > > > > Streams application instances are already up and running", right? > > Since > > > > > only the latter has rebalance process, while the Kafak brokers do > not > > > > > really have "rebalances" except balancing load by migrating > > partitions. > > > > > > > > > > Guozhang > > > > > > > > > > > > > > > > > > > > On Sun, Aug 19, 2018 at 7:47 PM, Nan Xu <nanxu1...@gmail.com> > wrote: > > > > > > > > > > > right, so my kafka cluster is already up and running for a while, > > > and I > > > > > can > > > > > > see from the log all broker instance already change from > rebalance > > to > > > > > > running. > > > > > > > > > > > > I did a another test. > > > > > > from producer, right before the message get send to the broker, I > > > put a > > > > > > timestamp in the message. and from the consumer side which is > after > > > > > stream > > > > > > processing, I compare this timestamp with current time. I can see > > > some > > > > > > message processing time is above 100ms on some real powerful > > > hardware. > > > > > and > > > > > > from my application gc, all the gc time is below 1ms, kafka gc > only > > > > > happen > > > > > > once and below 1ms too. > > > > > > > > > > > > very puzzled. is there any communication to zookeeper, if not get > > > > > response, > > > > > > will cause the broker to pause? I don't think that's the case but > > at > > > > this > > > > > > time don't know what else can be suspected. > > > > > > > > > > > > Nan > > > > > > > > > > > > On Sun, Aug 19, 2018 at 1:08 PM Guozhang Wang < > wangg...@gmail.com> > > > > > wrote: > > > > > > > > > > > > > Hello Nan, > > > > > > > > > > > > > > Note that Streams may need some time to rebalance and assign > > tasks > > > > even > > > > > > if > > > > > > > you only starts with one instance. > > > > > > > > > > > > > > I'd suggest you register your state listener in Kafka Streams > via > > > > > > > KafkaStreams#setStateListener, and your customized > StateListener > > > > should > > > > > > > record when the state transits from REBALANCING to RUNNING > since > > > only > > > > > > after > > > > > > > that the streams client will start to process the first record. > > > > > > > > > > > > > > > > > > > > > Guozhang > > > > > > > > > > > > > > > > > > > > > On Sat, Aug 18, 2018 at 8:52 PM, Nan Xu <nanxu1...@gmail.com> > > > wrote: > > > > > > > > > > > > > > > thanks, which JMX properties indicate "processing latency > > > > spikes" / > > > > > > > > "throughput" > > > > > > > > > > > > > > > > > > > > > > > > On Sat, Aug 18, 2018 at 5:45 PM Matthias J. Sax < > > > > > matth...@confluent.io > > > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > I cannot spot any obvious reasons. > > > > > > > > > > > > > > > > > > As you consume from the result topic for verification, we > > > should > > > > > > verify > > > > > > > > > that the latency spikes original on write and not on read: > > you > > > > > might > > > > > > > > > want to have a look into Kafka Streams JMX metric to see if > > > > > > processing > > > > > > > > > latency spikes or throughput drops. > > > > > > > > > > > > > > > > > > Also watch for GC pauses in the JVM. > > > > > > > > > > > > > > > > > > Hope this helps. > > > > > > > > > > > > > > > > > > > > > > > > > > > -Matthias > > > > > > > > > > > > > > > > > > On 8/17/18 12:13 PM, Nan Xu wrote: > > > > > > > > > > btw, I am using version 0.10.2.0 > > > > > > > > > > > > > > > > > > > > On Fri, Aug 17, 2018 at 2:04 PM Nan Xu < > > nanxu1...@gmail.com> > > > > > > wrote: > > > > > > > > > > > > > > > > > > > >> I am working on a kafka stream app, and see huge latency > > > > > variance, > > > > > > > > > >> wondering what can cause this? > > > > > > > > > >> > > > > > > > > > >> the processing is very simple and don't have state, > > > linger.ms > > > > > > > already > > > > > > > > > >> change to 5ms. the message size is around 10K byes and > > > > published > > > > > > as > > > > > > > > 2000 > > > > > > > > > >> messages/s, network is 10G. using a regular consumer > > watch > > > > the > > > > > > > > > >> localHistTopic topic and just every 2000 message print > > out > > > a > > > > > > > counter, > > > > > > > > > it > > > > > > > > > >> usually every second I get a count 2000 as the publish > > > speed, > > > > > but > > > > > > > > > sometime > > > > > > > > > >> I see it stall for 3 or more seconds and then print out > a > > > few > > > > > > count. > > > > > > > > > like > > > > > > > > > >> cpu is paused during that time or message being > > cache/batch > > > > then > > > > > > > > > processed. > > > > > > > > > >> any suggestion? > > > > > > > > > >> > > > > > > > > > >> final Properties streamsConfiguration = new > > Properties(); > > > > > > > > > >> > > > > > > > > > >> > > > > > > > streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, > > > > > > > > > >> applicationId); > > > > > > > > > >> > > > > > > > > > >> > streamsConfiguration.put(StreamsConfig.CLIENT_ID_ > > > > CONFIG, > > > > > > > > > clientId); > > > > > > > > > >> > > > > > > > > > >> > streamsConfiguration.put(StreamsConfig.BOOTSTRAP_ > > > > > > > > SERVERS_CONFIG, > > > > > > > > > >> bootstrapServers); > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > > >> streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_ > > > > > > > > SERDE_CLASS_CONFIG, > > > > > > > > > >> Serdes.String() > > > > > > > > > >> > > > > > > > > > >> .getClass().getName()); > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > > streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_ > > > > MS_CONFIG, > > > > > > > > > >> 10 * 1000); > > > > > > > > > >> > > > > > > > > > >> // > > > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > > > > > streamsConfiguration.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, > > > > > > > > 0); > > > > > > > > > >> > > > > > > > > > >> streamsConfiguration.put( > > > > StreamsConfig.PRODUCER_PREFIX > > > > > > > > > >> > > > > > > > > > >> + ProducerConfig.BATCH_SIZE_CONFIG,163840); > > > > > > > > > >> > > > > > > > > > >> streamsConfiguration.put( > > > > StreamsConfig.PRODUCER_PREFIX > > > > > > > > > >> > > > > > > > > > >> + ProducerConfig.BUFFER_MEMORY_CONFIG, > > > > 335544320); > > > > > > > > > >> > > > > > > > > > >> streamsConfiguration.put( > > > > StreamsConfig.PRODUCER_PREFIX > > > > > > > > > >> > > > > > > > > > >> + ProducerConfig.MAX_IN_FLIGHT_ > > > > > > REQUESTS_PER_CONNECTION, > > > > > > > > 30); > > > > > > > > > >> > > > > > > > > > >> streamsConfiguration.put( > > > > StreamsConfig.PRODUCER_PREFIX > > > > > > > > > >> > > > > > > > > > >> + ProducerConfig.LINGER_MS_CONFIG,"5"); > > > > > > > > > >> > > > > > > > > > >> streamsConfiguration.put( > > > > StreamsConfig.consumerPrefix( > > > > > > > > > >> ConsumerConfig.MAX_PARTITION_ > > > > FETCH_BYTES_CONFIG),20 > > > > > * > > > > > > > > 1024 * > > > > > > > > > >> 1024);MS_CONFIG, 10 * 1000); > > > > > > > > > >> > > > > > > > > > >> // > > > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > > > > > streamsConfiguration.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, > > > > > > > > 0); > > > > > > > > > >> > > > > > > > > > >> streamsConfiguration.put( > > > > StreamsConfig.PRODUCER_PREFIX > > > > > > > > > >> > > > > > > > > > >> + ProducerConfig.BATCH_SIZE_CONFIG,163840); > > > > > > > > > >> > > > > > > > > > >> streamsConfiguration.put( > > > > StreamsConfig.PRODUCER_PREFIX > > > > > > > > > >> > > > > > > > > > >> + ProducerConfig.BUFFER_MEMORY_CONFIG, > > > > 335544320); > > > > > > > > > >> > > > > > > > > > >> streamsConfiguration.put( > > > > StreamsConfig.PRODUCER_PREFIX > > > > > > > > > >> > > > > > > > > > >> + ProducerConfig.MAX_IN_FLIGHT_ > > > > > > REQUESTS_PER_CONNECTION, > > > > > > > > 30); > > > > > > > > > >> > > > > > > > > > >> streamsConfiguration.put( > > > > StreamsConfig.PRODUCER_PREFIX > > > > > > > > > >> > > > > > > > > > >> + ProducerConfig.LINGER_MS_CONFIG,"5"); > > > > > > > > > >> > > > > > > > > > >> streamsConfiguration.put( > > > > StreamsConfig.consumerPrefix( > > > > > > > > > >> > > > > > > > > > >> ConsumerConfig. MAX_PARTITION_FETCH_BYTES_ > > CONFIG > > > > , > > > > > > 20 * > > > > > > > > > 1024 * > > > > > > > > > >> 1024); > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > > >> final StreamsBuilder builder = new StreamsBuilder(); > > > > > > > > > >> > > > > > > > > > >> final KStream<String, NodeMutation> localDeltaStream = > > > > > > > > builder.stream( > > > > > > > > > >> > > > > > > > > > >> localDeltaTopic, > > > > > > > > > >> > > > > > > > > > >> Consumed.with( > > > > > > > > > >> > > > > > > > > > >> new Serdes.StringSerde(), > > > > > > > > > >> > > > > > > > > > >> new NodeMutationSerde<>() > > > > > > > > > >> > > > > > > > > > >> ) > > > > > > > > > >> > > > > > > > > > >> ); > > > > > > > > > >> > > > > > > > > > >> KStream<String, NodeState> localHistStream = > > > > > > > > > localDeltaStream.mapValues( > > > > > > > > > >> > > > > > > > > > >> (mutation) -> NodeState > > > > > > > > > >> > > > > > > > > > >> .newBuilder() > > > > > > > > > >> > > > > > > > > > >> .setMeta( > > > > > > > > > >> > > > > > > > > > >> mutation.getMetaMutation().getMeta() > > > > > > > > > >> > > > > > > > > > >> ) > > > > > > > > > >> > > > > > > > > > >> .setValue( > > > > > > > > > >> > > > > > > > > > >> mutation.getValueMutation(). > > getValue() > > > > > > > > > >> > > > > > > > > > >> ) > > > > > > > > > >> > > > > > > > > > >> .build() > > > > > > > > > >> > > > > > > > > > >> ); > > > > > > > > > >> > > > > > > > > > >> localHistStream.to( > > > > > > > > > >> > > > > > > > > > >> localHistTopic, > > > > > > > > > >> > > > > > > > > > >> Produced.with(new Serdes.StringSerde(), new > > > > > > > > > NodeStateSerde<>()) > > > > > > > > > >> > > > > > > > > > >> ); > > > > > > > > > >> > > > > > > > > > >> streams = new KafkaStreams(builder.build(), > > > > > > streamsConfiguration); > > > > > > > > > >> > > > > > > > > > >> streams.cleanUp(); > > > > > > > > > >> > > > > > > > > > >> streams.start(); > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > -- > > > > > > > -- Guozhang > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > -- > > > > > -- Guozhang > > > > > > > > > > > > > > > > > > > > > -- > > > -- Guozhang > > > > > > > > > -- > -- Guozhang >
