Hello Nan, What you described seems to be a broker-side latency spike, not a client (either producer, or consumer, or streams)-side latency spike. There's a number of possible reasons for such spikes: disk flushing (though async, it can also cause the processing thread to halt), GC, page faults (in case the thread needs to access a cold page), etc. It is hard to tell which one is the actual root cause.
For example, you can take a look at this slides (starting at 14), for a concrete example of such an investigation: https://www.slideshare.net/kawamuray/multitenancy-kafka-clusters-for-everyone-at-line My point is that it is not really easy via email discussion and by looking at your experiment code to tell exactly what is the root cause: the community can share with your some past experience and a few quick hinters, but most likely the issue varies case by case and hence can only be fully understandable by yourself. Guozhang On Sat, Aug 25, 2018 at 6:58 PM, Nan Xu <nanxu1...@gmail.com> wrote: > maybe easier to use github. > > https://github.com/angelfox123/kperf > > > On Sat, Aug 25, 2018 at 8:43 PM Nan Xu <nanxu1...@gmail.com> wrote: > > > so I did upgrade to 2.0.0 and still seeing the same result. below is the > > program I am using. I am running everything on a single server. (centos > 7, > > 24 core, 32G ram , 1 broker, 1 zookeeper, single harddrive), I understand > > the single hard drive is less ideal. but still don't expect it can over 3 > > seconds. > > > > > > case 1. > > I create 1 parittions for input and 1 partition for output. message size > > 10K > > producer give parameter (3600, 1000, 2) // 2 message per 1000 micro > > second for 3600 seconds, that translate to 2,000 message/s, I still see > > latency, sometime can reach to 3 seconds. > > > > case 2 > > 50 partitions for input, and 50 partitions for output. message size 10K > > producer give parameter (3600, 1000, 20) // 20 message per 1000 micro > > second for 3600 seconds, that translate to 20,000 message/s,latency not > > only high, and happen more often. > > > > > > Any suggestion is appreciated. target is per partition handle 1,000 -- > > 2,000 message/s and all latency lower than 100ms. > > > > ====build.gradle====== > > plugins { > > id 'application' > > id 'java' > > } > > group 'com.bofa' > > version '1.0-SNAPSHOT' > > sourceCompatibility = 1.8 > > mainClassName="main.StreamApp" > > > > repositories { > > mavenCentral() > > } > > > > dependencies { > > compile group: 'org.apache.kafka', name: 'kafka-clients', version: > > '2.0.0' > > compile group: "org.apache.kafka", name: "kafka-streams", version: > > "2.0.0" > > compile group: 'io.dropwizard.metrics', name: 'metrics-core', > > version:'3.2.6' > > testCompile group: 'junit', name: 'junit', version: '4.12' > > } > > > > ========producer======== > > package main; > > > > import java.util.Properties; > > import java.util.concurrent.atomic.AtomicInteger; > > > > import Util.BusyTimer; > > import org.apache.kafka.clients.producer.KafkaProducer; > > import org.apache.kafka.clients.producer.ProducerConfig; > > import org.apache.kafka.clients.producer.ProducerRecord; > > import org.apache.kafka.common.serialization.Serde; > > import org.apache.kafka.common.serialization.Serdes; > > import org.apache.kafka.common.serialization.StringSerializer; > > > > public class SimpleProducer { > > public static void main(String[] args) { > > final int time =Integer.valueOf(args[0]); > > final long interval = Integer.valueOf(args[1]); > > final int batch =Integer.valueOf(args[2]); > > Properties props = new Properties(); > > props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, > > "localhost:9092"); > > props.put(ProducerConfig.CLIENT_ID_CONFIG, > > "kafka-perf-test-producer"); > > props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, > > StringSerializer.class); > > props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, > > StringSerializer.class); > > > > KafkaProducer<String,String> kafkaProducer = new > > KafkaProducer(props); > > > > StringBuffer buffer = new StringBuffer(); > > for(int i=0; i<10240; i++) { > > buffer.append('a'); > > } > > String value = buffer.toString(); > > > > final long speed = 1000000/interval; > > Runnable task = new Runnable() { > > int sendNum=0; > > @Override > > public void run() { > > > > for(int i=0; i<batch; i++) { > > ProducerRecord<String, String> record = new > > ProducerRecord<>("input", System.nanoTime() + "-" + value); > > kafkaProducer.send(record); > > sendNum++; > > } > > > > if(sendNum % (speed * batch) == 0){ > > System.out.println(System.currentTimeMillis() + " : > " > > + sendNum); > > } > > } > > }; > > > > BusyTimer timer = new BusyTimer(interval,time, task); > > timer.spaceMessageWithInterval(); > > } > > } > > > > > > ============kafka stream============= > > package main; > > > > import java.util.Properties; > > > > import org.apache.kafka.clients.producer.ProducerConfig; > > import org.apache.kafka.common.serialization.Serdes; > > import org.apache.kafka.streams.Consumed; > > import org.apache.kafka.streams.KafkaStreams; > > import org.apache.kafka.streams.StreamsBuilder; > > import org.apache.kafka.streams.StreamsConfig; > > import org.apache.kafka.streams.kstream.KStream; > > import org.apache.kafka.streams.kstream.Produced; > > > > public class StreamApp { > > public static void main(String[] args) { > > final Properties streamsConfiguration = new Properties(); > > streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, > > "simple-stream"); > > streamsConfiguration.put(StreamsConfig.CLIENT_ID_CONFIG, > > "simple_stream_1"); > > streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, > > "localhost:9092"); > > > > streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, > > Serdes.String() > > .getClass().getName()); > > 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.NUM_STREAM_THREADS_CONFIG,2); > > > > > > StreamsBuilder builder = new StreamsBuilder(); > > final KStream<String, String> inputStream = builder.stream( > > "input", > > Consumed.with( > > new Serdes.StringSerde(), > > new Serdes.StringSerde() > > ) > > ); > > > > inputStream.to( > > "output", > > Produced.with(new Serdes.StringSerde(), new > > Serdes.StringSerde()) > > ); > > > > KafkaStreams streams = new KafkaStreams(builder.build(), > > streamsConfiguration); > > streams.start(); > > } > > } > > > > =============consumer============================ > > package main; > > > > import java.util.Collections; > > import java.util.Properties; > > > > import com.codahale.metrics.Reservoir; > > import com.codahale.metrics.UniformReservoir; > > import org.apache.kafka.clients.consumer.ConsumerConfig; > > import org.apache.kafka.clients.consumer.ConsumerRecord; > > import org.apache.kafka.clients.consumer.ConsumerRecords; > > import org.apache.kafka.clients.consumer.KafkaConsumer; > > import org.apache.kafka.common.serialization.StringDeserializer; > > > > public class SimpleConsumer { > > public static void main(String[] args) { > > int expectedSpeed = args[0]; > > Properties props = new Properties(); > > props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, > > "localhost:9092"); > > props.put(ConsumerConfig.GROUP_ID_CONFIG, > > "kafka-perf-consumer-group"); > > props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, > > StringDeserializer.class.getName()); > > props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); > > props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true); > > > > props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, > > StringDeserializer.class.getName()); > > > > KafkaConsumer consumer = new KafkaConsumer<String, > String>(props); > > consumer.subscribe(Collections.singletonList("output")); > > > > consumer.poll(0); > > int recNum=0; > > > > Reservoir totalRes = new UniformReservoir(); > > > > while (true) { > > ConsumerRecords<String, String> records = consumer.poll(10); > > for(ConsumerRecord<String,String> record : records){ > > long sendTime = Long.valueOf(record.value(). > split("-")[0]); > > long takeTime = System.nanoTime() - sendTime; > > if(recNum> 20000) { > > totalRes.update(takeTime); > > } > > recNum++; > > > > if(recNum % expectedSpeed == 0){ > > System.out.println("==============="+ recNum + > > "============"); > > System.out.println(" mean: " + > > totalRes.getSnapshot().getMean()/1000000); > > System.out.println(" 75%: " + > > totalRes.getSnapshot().get75thPercentile()/1000000); > > System.out.println(" 99%: " + > > totalRes.getSnapshot().get99thPercentile()/1000000); > > System.out.println(" 99.9%: " + > > totalRes.getSnapshot().get999thPercentile()/1000000); > > System.out.println(" Max: " + > > totalRes.getSnapshot().getMax()/1000000); > > > > System.out.println("========================================"); > > totalRes = new UniformReservoir(); > > } > > }; > > } > > } > > } > > > > ==========busy timer===================== > > //idea is space the message at a fixed time.(as thread.sleep, but sleep > is > > less accurate) > > package Util; > > > > import java.util.ArrayList; > > import java.util.concurrent.ExecutorService; > > import java.util.concurrent.Executors; > > import java.util.concurrent.atomic.AtomicInteger; > > import java.util.concurrent.atomic.AtomicLong; > > > > public class BusyTimer { > > long interval; > > long duration; > > ArrayList<Long> pubTime; > > ExecutorService ex = Executors.newSingleThreadExecutor(); > > Runnable task; > > > > > > public BusyTimer(long microInterval, long exDurationInSeconds, > > Runnable task){ > > pubTime = new ArrayList<Long>((int)(exDurationInSeconds * 1000 * > > 1000 / microInterval+1)); > > > > this.interval = microInterval * 1000; > > this.duration = exDurationInSeconds * 1000000000; > > this.task = task; > > > > } > > > > private void busywaitUntil(long nano){ > > while(System.nanoTime() < nano){ > > > > } > > } > > > > public void spaceMessageWithInterval(){ > > int i =0 ; > > long baseTime = System.nanoTime(); > > long doneTime = baseTime + duration; > > while(true) { > > task.run(); > > pubTime.add(System.nanoTime()); > > long targetTime = System.nanoTime() + interval; > > if(System.nanoTime() > doneTime ){ > > break; > > } > > busywaitUntil(targetTime); > > } > > } > > } > > > > > > > > On Fri, Aug 24, 2018 at 3:37 PM Nan Xu <nanxu1...@gmail.com> wrote: > > > >> Looks really promising but after upgrade, still show the same result. I > >> will post the program soon. Maybe you can see where the problem could > be. > >> > >> Nan > >> > >> On Thu, Aug 23, 2018, 7:34 PM Guozhang Wang <wangg...@gmail.com> wrote: > >> > >>> Hello Nan, > >>> > >>> Kafka does not tie up the processing thread to do disk flushing. > However, > >>> since you are on an older version of Kafka I suspect you're bumping > into > >>> some old issues that have been resolved in later versions. e.g. > >>> > >>> https://issues.apache.org/jira/browse/KAFKA-4614 > >>> > >>> I'd suggest you upgrading to latest version (2.0.0) and try again to > see > >>> if > >>> you observe the same pattern. > >>> > >>> Guozhang > >>> > >>> On Thu, Aug 23, 2018 at 3:22 PM, Sudhir Babu Pothineni < > >>> sbpothin...@gmail.com> wrote: > >>> > >>> > I will wait for the expert’s opinion: > >>> > > >>> > Did the Transparent Huge Pages(THP) disabled on the broker machine? > >>> it’s a > >>> > Linux kernel parameter. > >>> > > >>> > -Sudhir > >>> > > >>> > > On Aug 23, 2018, at 4:46 PM, Nan Xu <nanxu1...@gmail.com> wrote: > >>> > > > >>> > > 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 > >>> > >> > >>> > > >>> > >>> > >>> > >>> -- > >>> -- Guozhang > >>> > >> > -- -- Guozhang
