{"action":"created","comment":{"url":"https://api.github.com/repos/apache/kafka/pulls/comments/180587492","pull_request_review_id":111030096,"id":180587492,"diff_hunk":"@@ -266,521 +282,339 @@ private Properties setProduceConsumeProperties(final String clientId) {\n         return props;\n     }\n \n-    private boolean maybeSetupPhase(final String topic, final String clientId,\n-                                    final boolean skipIfAllTests) {\n-        resetStats();\n-        // initialize topics\n-        if (loadPhase) {\n-            if (skipIfAllTests) {\n-                // if we run all tests, the produce test will have already loaded the data\n-                if (testName.equals(ALL_TESTS)) {\n-                    // Skipping loading phase since previously loaded\n-                    return true;\n-                }\n-            }\n-            System.out.println(\"Initializing topic \" + topic);\n-            // WARNING: The keys must be sequential, i.e., unique, otherwise the logic for when this test\n-            // stops will not work (in createCountStreams)\n-            produce(topic, VALUE_SIZE, clientId, numRecords, true, numRecords, false);\n-            return true;\n-        }\n-        return false;\n-    }\n-\n     void resetStats() {\n-        processedRecords.set(0);\n-        processedBytes = 0;\n-    }\n-\n-\n-    private KafkaStreams createCountStreams(Properties streamConfig, String topic, final CountDownLatch latch) {\n-        final StreamsBuilder builder = new StreamsBuilder();\n-        final KStream<Integer, byte[]> input = builder.stream(topic);\n-\n-        input.groupByKey()\n-            .count(\"tmpStoreName\").foreach(new CountDownAction(latch));\n-\n-        return new KafkaStreams(builder.build(), streamConfig);\n-    }\n-\n-\n-    private void yahooBenchmark(final String campaignsTopic, final String eventsTopic) {\n-        YahooBenchmark benchmark = new YahooBenchmark(this, campaignsTopic, eventsTopic);\n-\n-        benchmark.run();\n-    }\n-\n-    /**\n-     * Measure the performance of a simple aggregate like count.\n-     * Counts the occurrence of numbers (note that normally people count words, this\n-     * example counts numbers)\n-     * @param countTopic Topic where numbers are stored\n-     * @throws Exception\n-     */\n-    public void count(String countTopic) {\n-        if (maybeSetupPhase(countTopic, \"simple-benchmark-produce-count\", false)) {\n-            return;\n-        }\n-\n-        CountDownLatch latch = new CountDownLatch(1);\n-        setStreamProperties(\"simple-benchmark-count\");\n-        final KafkaStreams streams = createCountStreams(props, countTopic, latch);\n-        runGenericBenchmark(streams, \"Streams Count Performance [records/latency/rec-sec/MB-sec counted]: \", latch);\n-    }\n-\n-    /**\n-     * Measure the performance of a KStream-KTable left join. The setup is such that each\n-     * KStream record joins to exactly one element in the KTable\n-     */\n-    public void kStreamKTableJoin(String kStreamTopic, String kTableTopic) {\n-        if (maybeSetupPhase(kStreamTopic, \"simple-benchmark-produce-kstream\", false)) {\n-            maybeSetupPhase(kTableTopic, \"simple-benchmark-produce-ktable\", false);\n-            return;\n-        }\n-\n-        CountDownLatch latch = new CountDownLatch(1);\n-\n-        // setup join\n-        setStreamProperties(\"simple-benchmark-kstream-ktable-join\");\n-        final KafkaStreams streams = createKafkaStreamsKStreamKTableJoin(props, kStreamTopic, kTableTopic, latch);\n-\n-        // run benchmark\n-        runGenericBenchmark(streams, \"Streams KStreamKTable LeftJoin Performance [records/latency/rec-sec/MB-sec joined]: \", latch);\n+        processedRecords = 0;\n+        processedBytes = 0L;\n     }\n \n     /**\n-     * Measure the performance of a KStream-KStream left join. The setup is such that each\n-     * KStream record joins to exactly one element in the other KStream\n+     * Produce values to a topic\n+     * @param clientId String specifying client ID\n+     * @param topic Topic to produce to\n+     * @param numRecords Number of records to produce\n+     * @param keySkew Key zipf distribution skewness\n+     * @param valueSize Size of value in bytes\n      */\n-    public void kStreamKStreamJoin(String kStreamTopic1, String kStreamTopic2) {\n-        if (maybeSetupPhase(kStreamTopic1, \"simple-benchmark-produce-kstream-topic1\", false)) {\n-            maybeSetupPhase(kStreamTopic2, \"simple-benchmark-produce-kstream-topic2\", false);\n-            return;\n-        }\n-\n-        CountDownLatch latch = new CountDownLatch(1);\n-\n-        // setup join\n-        setStreamProperties(\"simple-benchmark-kstream-kstream-join\");\n-        final KafkaStreams streams = createKafkaStreamsKStreamKStreamJoin(props, kStreamTopic1, kStreamTopic2, latch);\n-\n-        // run benchmark\n-        runGenericBenchmark(streams, \"Streams KStreamKStream LeftJoin Performance [records/latency/rec-sec/MB-sec  joined]: \", latch);\n-    }\n+    private void produce(final String clientId,\n+                         final String topic,\n+                         final int numRecords,\n+                         final double keySkew,\n+                         final int valueSize) {\n+        final Properties props = setProduceConsumeProperties(clientId);\n+        final ZipfGenerator keyGen = new ZipfGenerator(KEY_SPACE_SIZE, keySkew);\n+        final KafkaProducer<Integer, byte[]> producer = new KafkaProducer<>(props);\n+\n+        final byte[] value = new byte[valueSize];\n+        // put some random values to increase entropy. Some devices\n+        // like SSDs do compression and if the array is all zeros\n+        // the performance will be too good.\n+        new Random().nextBytes(value);\n \n-    /**\n-     * Measure the performance of a KTable-KTable left join. The setup is such that each\n-     * KTable record joins to exactly one element in the other KTable\n-     */\n-    public void kTableKTableJoin(String kTableTopic1, String kTableTopic2) {\n-        if (maybeSetupPhase(kTableTopic1, \"simple-benchmark-produce-ktable-topic1\", false)) {\n-            maybeSetupPhase(kTableTopic2, \"simple-benchmark-produce-ktable-topic2\", false);\n-            return;\n+        for (int i = 0; i < numRecords; i++) {\n+            producer.send(new ProducerRecord<>(topic, keyGen.next(), value));\n         }\n-        CountDownLatch latch = new CountDownLatch(1);\n-\n-        // setup join\n-        setStreamProperties(\"simple-benchmark-ktable-ktable-join\");\n-        final KafkaStreams streams = createKafkaStreamsKTableKTableJoin(props, kTableTopic1, kTableTopic2, latch);\n \n-        // run benchmark\n-        runGenericBenchmark(streams, \"Streams KTableKTable LeftJoin Performance [records/latency/rec-sec/MB-sec joined]: \", latch);\n-    }\n-\n-    void printResults(final String nameOfBenchmark, final long latency) {\n-        System.out.println(nameOfBenchmark +\n-            processedRecords.get() + \"/\" +\n-            latency + \"/\" +\n-            recordsPerSec(latency, processedRecords.get()) + \"/\" +\n-            megabytesPerSec(latency, processedBytes));\n+        producer.close();\n     }\n \n-    void runGenericBenchmark(final KafkaStreams streams, final String nameOfBenchmark, final CountDownLatch latch) {\n-        streams.start();\n+    private void consumeAndProduce(final String topic) {\n+        final Properties consumerProps = setProduceConsumeProperties(\"simple-benchmark-consumer\");\n+        final Properties producerProps = setProduceConsumeProperties(\"simple-benchmark-producer\");\n \n-        long startTime = System.currentTimeMillis();\n+        final KafkaConsumer<Integer, byte[]> consumer = new KafkaConsumer<>(consumerProps);\n+        final KafkaProducer<Integer, byte[]> producer = new KafkaProducer<>(producerProps);\n+        final List<TopicPartition> partitions = getAllPartitions(consumer, topic);\n \n-        while (latch.getCount() > 0) {\n-            try {\n-                latch.await();\n-            } catch (InterruptedException ex) {\n-                Thread.currentThread().interrupt();\n-            }\n-        }\n-        long endTime = System.currentTimeMillis();\n-        printResults(nameOfBenchmark, endTime - startTime);\n-\n-        streams.close();\n-    }\n-\n-    private long startStreamsThread(final KafkaStreams streams, final CountDownLatch latch) {\n-        Thread thread = new Thread() {\n-            public void run() {\n-                streams.start();\n-            }\n-        };\n-        thread.start();\n+        consumer.assign(partitions);\n+        consumer.seekToBeginning(partitions);\n \n         long startTime = System.currentTimeMillis();\n \n-        while (latch.getCount() > 0) {\n-            try {\n-                latch.await();\n-            } catch (InterruptedException ex) {\n-                Thread.interrupted();\n+        boolean keepProcessing = true;\n+        while (keepProcessing) {\n+            ConsumerRecords<Integer, byte[]> records = consumer.poll(POLL_MS);\n+            if (records.isEmpty()) {\n+                if (processedRecords == numRecords)\n+                    keepProcessing = false;\n+            } else {\n+                for (ConsumerRecord<Integer, byte[]> record : records) {\n+                    producer.send(new ProducerRecord<>(SINK_TOPIC, record.key(), record.value()));\n+                    processedRecords++;\n+                    processedBytes += record.value().length + Integer.SIZE;\n+                }\n             }\n+            if (processedRecords == numRecords)\n+                keepProcessing = false;\n         }\n \n         long endTime = System.currentTimeMillis();\n \n-        streams.close();\n-        try {\n-            thread.join();\n-        } catch (Exception ex) {\n-            // ignore\n-        }\n-\n-        return endTime - startTime;\n-    }\n-\n-    public void processStream(final String topic) {\n-        if (maybeSetupPhase(topic, \"simple-benchmark-process-stream-load\", true)) {\n-            return;\n-        }\n-\n-        CountDownLatch latch = new CountDownLatch(1);\n-\n-        final KafkaStreams streams = createKafkaStreams(topic, latch);\n-        long latency = startStreamsThread(streams, latch);\n-\n-        printResults(\"Streams Performance [records/latency/rec-sec/MB-sec source]: \", latency);\n-    }\n-\n-    public void processStreamWithSink(String topic) {\n-        if (maybeSetupPhase(topic, \"simple-benchmark-process-stream-with-sink-load\", true)) {\n-            return;\n-        }\n-\n-        CountDownLatch latch = new CountDownLatch(1);\n-        final KafkaStreams streams = createKafkaStreamsWithSink(topic, latch);\n-        long latency = startStreamsThread(streams, latch);\n-\n-        printResults(\"Streams Performance [records/latency/rec-sec/MB-sec source+sink]: \", latency);\n-\n-    }\n-\n-    public void processStreamWithStateStore(String topic) {\n-        if (maybeSetupPhase(topic, \"simple-benchmark-process-stream-with-state-store-load\", true)) {\n-            return;\n-        }\n-\n-        CountDownLatch latch = new CountDownLatch(1);\n-        final KafkaStreams streams = createKafkaStreamsWithStateStore(topic, latch, false);\n-        long latency = startStreamsThread(streams, latch);\n-        printResults(\"Streams Performance [records/latency/rec-sec/MB-sec source+store]: \", latency);\n-\n-    }\n-\n-    public void processStreamWithCachedStateStore(String topic) {\n-        if (maybeSetupPhase(topic, \"simple-benchmark-process-stream-with-cached-state-store-load\", true)) {\n-            return;\n-        }\n-\n-        CountDownLatch latch = new CountDownLatch(1);\n-        final KafkaStreams streams = createKafkaStreamsWithStateStore(topic, latch, true);\n-        long latency = startStreamsThread(streams, latch);\n-        printResults(\"Streams Performance [records/latency/rec-sec/MB-sec source+cache+store]: \", latency);\n-    }\n-\n-    public void produce(String topic) {\n-        // loading phase does not make sense for producer\n-        if (loadPhase) {\n-            resetStats();\n-            return;\n-        }\n-        produce(topic, VALUE_SIZE, \"simple-benchmark-produce\", numRecords, true, numRecords, true);\n-    }\n-    /**\n-     * Produce values to a topic\n-     * @param topic Topic to produce to\n-     * @param valueSizeBytes Size of value in bytes\n-     * @param clientId String specifying client ID\n-     * @param numRecords Number of records to produce\n-     * @param sequential if True, then keys are produced sequentially from 0 to upperRange. In this case upperRange must be >= numRecords.\n-     *                   if False, then keys are produced randomly in range [0, upperRange)\n-     * @param printStats if True, print stats on how long producing took. If False, don't print stats. False can be used\n-     *                   when this produce step is part of another benchmark that produces its own stats\n-     */\n-    private void produce(String topic, int valueSizeBytes, String clientId, int numRecords, boolean sequential,\n-                         int upperRange, boolean printStats) {\n-\n-\n-        if (sequential) {\n-            if (upperRange < numRecords) throw new IllegalArgumentException(\"UpperRange must be >= numRecords\");\n-        }\n-        if (!sequential) {\n-            System.out.println(\"WARNING: You are using non-sequential keys. If your tests' exit logic expects to see a final key, random keys may not work.\");\n-        }\n-        Properties props = setProduceConsumeProperties(clientId);\n-\n-        int key = 0;\n-        Random rand = new Random();\n-        KafkaProducer<Integer, byte[]> producer = new KafkaProducer<>(props);\n-\n-        byte[] value = new byte[valueSizeBytes];\n-        // put some random values to increase entropy. Some devices\n-        // like SSDs do compression and if the array is all zeros\n-        // the performance will be too good.\n-        new Random().nextBytes(value);\n-        long startTime = System.currentTimeMillis();\n-\n-        if (sequential) key = 0;\n-        else key = rand.nextInt(upperRange);\n-        for (int i = 0; i < numRecords; i++) {\n-            producer.send(new ProducerRecord<>(topic, key, value));\n-            if (sequential) key++;\n-            else key = rand.nextInt(upperRange);\n-            processedRecords.getAndIncrement();\n-            processedBytes += value.length + Integer.SIZE;\n-        }\n+        consumer.close();\n         producer.close();\n \n-        long endTime = System.currentTimeMillis();\n-\n-        if (printStats) {\n-            printResults(\"Producer Performance [records/latency/rec-sec/MB-sec write]: \", endTime - startTime);\n-        }\n+        printResults(\"ConsumerProducer Performance [records/latency/rec-sec/MB-sec read]: \", endTime - startTime);\n     }\n \n-    public void consume(String topic) {\n-        if (maybeSetupPhase(topic, \"simple-benchmark-consumer-load\", true)) {\n-            return;\n-        }\n+    private void consume(final String topic) {\n+        final Properties consumerProps = setProduceConsumeProperties(\"simple-benchmark-consumer\");\n+        final KafkaConsumer<Integer, byte[]> consumer = new KafkaConsumer<>(consumerProps);\n+        final List<TopicPartition> partitions = getAllPartitions(consumer, topic);\n \n-        Properties props = setProduceConsumeProperties(\"simple-benchmark-consumer\");\n-\n-        KafkaConsumer<Integer, byte[]> consumer = new KafkaConsumer<>(props);\n-\n-        List<TopicPartition> partitions = getAllPartitions(consumer, topic);\n         consumer.assign(partitions);\n         consumer.seekToBeginning(partitions);\n \n-        Integer key = null;\n-\n         long startTime = System.currentTimeMillis();\n \n         while (true) {\n             ConsumerRecords<Integer, byte[]> records = consumer.poll(POLL_MS);\n             if (records.isEmpty()) {\n-                if (processedRecords.get() == numRecords)\n+                if (processedRecords == numRecords)\n                     break;\n             } else {\n                 for (ConsumerRecord<Integer, byte[]> record : records) {\n-                    processedRecords.getAndIncrement();\n+                    processedRecords++;\n                     processedBytes += record.value().length + Integer.SIZE;\n-                    Integer recKey = record.key();\n-                    if (key == null || key < recKey)\n-                        key = recKey;\n-                    if (processedRecords.get() == numRecords)\n+                    if (processedRecords == numRecords)\n                         break;\n                 }\n             }\n-            if (processedRecords.get() == numRecords)\n+            if (processedRecords == numRecords)\n                 break;\n         }\n \n         long endTime = System.currentTimeMillis();\n \n         consumer.close();\n+\n         printResults(\"Consumer Performance [records/latency/rec-sec/MB-sec read]: \", endTime - startTime);\n     }\n \n-    private KafkaStreams createKafkaStreams(String topic, final CountDownLatch latch) {\n-        setStreamProperties(\"simple-benchmark-streams\");\n+    private void processStream(final String topic) {\n+        final CountDownLatch latch = new CountDownLatch(1);\n+\n+        setStreamProperties(\"simple-benchmark-streams-source\");\n \n         StreamsBuilder builder = new StreamsBuilder();\n \n         KStream<Integer, byte[]> source = builder.stream(topic, Consumed.with(INTEGER_SERDE, BYTE_SERDE));\n \n-        source.process(new ProcessorSupplier<Integer, byte[]>() {\n-            @Override\n-            public Processor<Integer, byte[]> get() {\n-                return new AbstractProcessor<Integer, byte[]>() {\n+        source.peek(new CountDownAction(latch));\n \n-                    @Override\n-                    public void init(ProcessorContext context) {\n-                    }\n+        final KafkaStreams streams = createKafkaStreamsWithExceptionHandler(builder, props);\n+        runGenericBenchmark(streams, \"Streams Source Performance [records/latency/rec-sec/MB-sec joined]: \", latch);\n+    }\n \n-                    @Override\n-                    public void process(Integer key, byte[] value) {\n-                        processedRecords.getAndIncrement();\n-                        processedBytes += value.length + Integer.SIZE;\n-                        if (processedRecords.get() == numRecords) {\n-                            latch.countDown();\n-                        }\n-                    }\n+    private void processStreamWithSink(final String topic) {\n+        final CountDownLatch latch = new CountDownLatch(1);\n \n-                    @Override\n-                    public void punctuate(long timestamp) {\n-                    }\n+        setStreamProperties(\"simple-benchmark-streams-source-sink\");\n \n-                    @Override\n-                    public void close() {\n-                    }\n-                };\n-            }\n-        });\n+        StreamsBuilder builder = new StreamsBuilder();\n \n-        return createKafkaStreamsWithExceptionHandler(builder, props);\n+        KStream<Integer, byte[]> source = builder.stream(topic);\n+\n+        source.peek(new CountDownAction(latch)).to(SINK_TOPIC);\n+\n+        final KafkaStreams streams = createKafkaStreamsWithExceptionHandler(builder, props);\n+        runGenericBenchmark(streams, \"Streams SourceSink Performance [records/latency/rec-sec/MB-sec joined]: \", latch);\n     }\n \n-    private KafkaStreams createKafkaStreamsWithSink(String topic, final CountDownLatch latch) {\n-        setStreamProperties(\"simple-benchmark-streams-with-sink\");\n+    private void processStreamWithStateStore(String topic) {\n+        final CountDownLatch latch = new CountDownLatch(1);\n+\n+        setStreamProperties(\"simple-benchmark-streams-with-store\");\n \n         StreamsBuilder builder = new StreamsBuilder();","path":"streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java","position":780,"original_position":780,"commit_id":"6a1bdd02a086e991643a070738d3a33ce50213d0","original_commit_id":"6a1bdd02a086e991643a070738d3a33ce50213d0","user":{"login":"mjsax","id":8959638,"avatar_url":"https://avatars1.githubusercontent.com/u/8959638?v=4","gravatar_id":"","url":"https://api.github.com/users/mjsax","html_url":"https://github.com/mjsax","followers_url":"https://api.github.com/users/mjsax/followers","following_url":"https://api.github.com/users/mjsax/following{/other_user}","gists_url":"https://api.github.com/users/mjsax/gists{/gist_id}","starred_url":"https://api.github.com/users/mjsax/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mjsax/subscriptions","organizations_url":"https://api.github.com/users/mjsax/orgs","repos_url":"https://api.github.com/users/mjsax/repos","events_url":"https://api.github.com/users/mjsax/events{/privacy}","received_events_url":"https://api.github.com/users/mjsax/received_events","type":"User","site_admin":false},"body":"nit: add `final`","created_at":"2018-04-10T22:35:12Z","updated_at":"2018-04-10T22:51:56Z","html_url":"https://github.com/apache/kafka/pull/4744#discussion_r180587492","pull_request_url":"https://api.github.com/repos/apache/kafka/pulls/4744","author_association":"MEMBER","_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments/180587492"},"html":{"href":"https://github.com/apache/kafka/pull/4744#discussion_r180587492"},"pull_request":{"href":"https://api.github.com/repos/apache/kafka/pulls/4744"}}},"pull_request":{"url":"https://api.github.com/repos/apache/kafka/pulls/4744","id":176355980,"html_url":"https://github.com/apache/kafka/pull/4744","diff_url":"https://github.com/apache/kafka/pull/4744.diff","patch_url":"https://github.com/apache/kafka/pull/4744.patch","issue_url":"https://api.github.com/repos/apache/kafka/issues/4744","number":4744,"state":"open","locked":false,"title":"KAFKA-6611, PART II: Improve Streams SimpleBenchmark","user":{"login":"guozhangwang","id":10729597,"avatar_url":"https://avatars0.githubusercontent.com/u/10729597?v=4","gravatar_id":"","url":"https://api.github.com/users/guozhangwang","html_url":"https://github.com/guozhangwang","followers_url":"https://api.github.com/users/guozhangwang/followers","following_url":"https://api.github.com/users/guozhangwang/following{/other_user}","gists_url":"https://api.github.com/users/guozhangwang/gists{/gist_id}","starred_url":"https://api.github.com/users/guozhangwang/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/guozhangwang/subscriptions","organizations_url":"https://api.github.com/users/guozhangwang/orgs","repos_url":"https://api.github.com/users/guozhangwang/repos","events_url":"https://api.github.com/users/guozhangwang/events{/privacy}","received_events_url":"https://api.github.com/users/guozhangwang/received_events","type":"User","site_admin":false},"body":"1. SimpleBenchmark:\r\n\r\n    1.a Do not rely on manual num.records / bytes collection on atomic integers.\r\n    1.b Rely on config files for num.threads, bootstrap.servers, etc.\r\n    1.c Add parameters for key skewness and value size.\r\n    1.d Refactor the tests for loading phase, adding tumbling-windowed count.\r\n    1.e For consumer / consumeproduce, collect metrics on consumer instead.\r\n    1.f Force stop the test after 3 minutes, this is based on empirical numbers of 10M records.\r\n\r\n2. Other tests: use config for kafka bootstrap servers.\r\n\r\n3. streams_simple_benchmark.py: only use scale 1 for system test, remove yahoo from benchmark tests.\r\n\r\nNote that the JMX based metrics is more accurate than the manually collected metrics. For example:\r\n\r\n```\r\nModule: kafkatest.benchmarks.streams.streams_simple_benchmark_test\r\nClass:  StreamsSimpleBenchmarkTest\r\nMethod: test_simple_benchmark\r\nArguments:\r\n{\r\n  \"scale\": 3,\r\n  \"test\": \"streamcount\"\r\n}\r\n----\r\n\r\n{\r\n  \"Streams Count Performance [records/latency/rec-sec/MB-sec counted]0\": \" 3691724/180042/20504.79332600171/20.64996886468678\\n\",\r\n  \"Streams Count Performance [records/latency/rec-sec/MB-sec counted]1\": \" 3337273/180037/18536.595255419717/18.667835797999594\\n\",\r\n  \"Streams Count Performance [records/latency/rec-sec/MB-sec counted]2\": \" 2971003/180029/16502.913419504635/16.61975533580484\\n\",\r\n  \"jmx-avg0\": {\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:commit-latency-avg\": 473.9851851851854,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:commit-rate\": 0.03802973633453574,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:poll-latency-avg\": 1.6337178935425791,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:poll-rate\": 41.49027662118575,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:process-latency-avg\": 0.0021335594038200878,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:process-rate\": 21042.80038005481\r\n  },\r\n  \"jmx-avg1\": {\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:commit-latency-avg\": 354.0805555555556,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:commit-rate\": 0.030462783608341253,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:poll-latency-avg\": 2.2233941132541286,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:poll-rate\": 35.82136735561879,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:process-latency-avg\": 0.002402570337120884,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:process-rate\": 19100.53692641491\r\n  },\r\n  \"jmx-avg2\": {\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:commit-latency-avg\": 365.85555555555555,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:commit-rate\": 0.03052173041795845,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:poll-latency-avg\": 1.1700956516025365,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:poll-rate\": 28.06739395694291,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:process-latency-avg\": 0.0016847557478484937,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:process-rate\": 15594.917035789254\r\n  },\r\n  \"jmx-max0\": {\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:commit-latency-avg\": 547.0,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:commit-rate\": 0.06657567990413102,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:poll-latency-avg\": 13.04,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:poll-rate\": 196.4955857339561,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:process-latency-avg\": 0.04672288495817908,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:process-rate\": 99520.53241178217\r\n  },\r\n  \"jmx-max1\": {\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:commit-latency-avg\": 653.0,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:commit-rate\": 0.065242211710977,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:poll-latency-avg\": 13.909090909090908,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:poll-rate\": 156.47243937161267,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:process-latency-avg\": 0.08381330685203575,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:process-rate\": 83263.21836209696\r\n  },\r\n  \"jmx-max2\": {\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:commit-latency-avg\": 632.0,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:commit-rate\": 0.06459531038046638,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:poll-latency-avg\": 8.25,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:poll-rate\": 173.80159786950733,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:process-latency-avg\": 0.07433900760593988,\r\n    \"kafka.streams:type=stream-metrics,client-id=simple-benchmark-StreamThread-1:process-rate\": 97828.82822902796\r\n  }\r\n}\r\n\r\n```\r\n\r\n\r\n### Committer Checklist (excluded from commit message)\r\n- [ ] Verify design and implementation \r\n- [ ] Verify test coverage and CI build status\r\n- [ ] Verify documentation (including upgrade notes)\r\n","created_at":"2018-03-21T00:11:40Z","updated_at":"2018-04-10T22:51:57Z","closed_at":null,"merged_at":null,"merge_commit_sha":"852522c9af5c376dfea22021c7a7d52f86605b02","assignee":null,"assignees":[],"requested_reviewers":[],"requested_teams":[],"labels":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/kafka/pulls/4744/commits","review_comments_url":"https://api.github.com/repos/apache/kafka/pulls/4744/comments","review_comment_url":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/kafka/issues/4744/comments","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/6a1bdd02a086e991643a070738d3a33ce50213d0","head":{"label":"guozhangwang:KMinor-Simplebenchmark-improvement","ref":"KMinor-Simplebenchmark-improvement","sha":"6a1bdd02a086e991643a070738d3a33ce50213d0","user":{"login":"guozhangwang","id":10729597,"avatar_url":"https://avatars0.githubusercontent.com/u/10729597?v=4","gravatar_id":"","url":"https://api.github.com/users/guozhangwang","html_url":"https://github.com/guozhangwang","followers_url":"https://api.github.com/users/guozhangwang/followers","following_url":"https://api.github.com/users/guozhangwang/following{/other_user}","gists_url":"https://api.github.com/users/guozhangwang/gists{/gist_id}","starred_url":"https://api.github.com/users/guozhangwang/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/guozhangwang/subscriptions","organizations_url":"https://api.github.com/users/guozhangwang/orgs","repos_url":"https://api.github.com/users/guozhangwang/repos","events_url":"https://api.github.com/users/guozhangwang/events{/privacy}","received_events_url":"https://api.github.com/users/guozhangwang/received_events","type":"User","site_admin":false},"repo":{"id":39164613,"name":"kafka","full_name":"guozhangwang/kafka","owner":{"login":"guozhangwang","id":10729597,"avatar_url":"https://avatars0.githubusercontent.com/u/10729597?v=4","gravatar_id":"","url":"https://api.github.com/users/guozhangwang","html_url":"https://github.com/guozhangwang","followers_url":"https://api.github.com/users/guozhangwang/followers","following_url":"https://api.github.com/users/guozhangwang/following{/other_user}","gists_url":"https://api.github.com/users/guozhangwang/gists{/gist_id}","starred_url":"https://api.github.com/users/guozhangwang/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/guozhangwang/subscriptions","organizations_url":"https://api.github.com/users/guozhangwang/orgs","repos_url":"https://api.github.com/users/guozhangwang/repos","events_url":"https://api.github.com/users/guozhangwang/events{/privacy}","received_events_url":"https://api.github.com/users/guozhangwang/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/guozhangwang/kafka","description":"Mirror of Apache Kafka","fork":true,"url":"https://api.github.com/repos/guozhangwang/kafka","forks_url":"https://api.github.com/repos/guozhangwang/kafka/forks","keys_url":"https://api.github.com/repos/guozhangwang/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/guozhangwang/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/guozhangwang/kafka/teams","hooks_url":"https://api.github.com/repos/guozhangwang/kafka/hooks","issue_events_url":"https://api.github.com/repos/guozhangwang/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/guozhangwang/kafka/events","assignees_url":"https://api.github.com/repos/guozhangwang/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/guozhangwang/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/guozhangwang/kafka/tags","blobs_url":"https://api.github.com/repos/guozhangwang/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/guozhangwang/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/guozhangwang/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/guozhangwang/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/guozhangwang/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/guozhangwang/kafka/languages","stargazers_url":"https://api.github.com/repos/guozhangwang/kafka/stargazers","contributors_url":"https://api.github.com/repos/guozhangwang/kafka/contributors","subscribers_url":"https://api.github.com/repos/guozhangwang/kafka/subscribers","subscription_url":"https://api.github.com/repos/guozhangwang/kafka/subscription","commits_url":"https://api.github.com/repos/guozhangwang/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/guozhangwang/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/guozhangwang/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/guozhangwang/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/guozhangwang/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/guozhangwang/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/guozhangwang/kafka/merges","archive_url":"https://api.github.com/repos/guozhangwang/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/guozhangwang/kafka/downloads","issues_url":"https://api.github.com/repos/guozhangwang/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/guozhangwang/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/guozhangwang/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/guozhangwang/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/guozhangwang/kafka/labels{/name}","releases_url":"https://api.github.com/repos/guozhangwang/kafka/releases{/id}","deployments_url":"https://api.github.com/repos/guozhangwang/kafka/deployments","created_at":"2015-07-15T22:40:38Z","updated_at":"2018-04-06T21:11:36Z","pushed_at":"2018-04-10T22:48:59Z","git_url":"git://github.com/guozhangwang/kafka.git","ssh_url":"git@github.com:guozhangwang/kafka.git","clone_url":"https://github.com/guozhangwang/kafka.git","svn_url":"https://github.com/guozhangwang/kafka","homepage":null,"size":64087,"stargazers_count":1,"watchers_count":1,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":1,"mirror_url":null,"archived":false,"open_issues_count":1,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0"},"forks":1,"open_issues":1,"watchers":1,"default_branch":"trunk"}},"base":{"label":"apache:trunk","ref":"trunk","sha":"0a8f35b68415bb3f79b0ec61df6cb0ab0db937c4","user":{"login":"apache","id":47359,"avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":2211243,"name":"kafka","full_name":"apache/kafka","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/kafka","description":"Mirror of Apache Kafka","fork":false,"url":"https://api.github.com/repos/apache/kafka","forks_url":"https://api.github.com/repos/apache/kafka/forks","keys_url":"https://api.github.com/repos/apache/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/kafka/teams","hooks_url":"https://api.github.com/repos/apache/kafka/hooks","issue_events_url":"https://api.github.com/repos/apache/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/apache/kafka/events","assignees_url":"https://api.github.com/repos/apache/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/apache/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/apache/kafka/tags","blobs_url":"https://api.github.com/repos/apache/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/kafka/languages","stargazers_url":"https://api.github.com/repos/apache/kafka/stargazers","contributors_url":"https://api.github.com/repos/apache/kafka/contributors","subscribers_url":"https://api.github.com/repos/apache/kafka/subscribers","subscription_url":"https://api.github.com/repos/apache/kafka/subscription","commits_url":"https://api.github.com/repos/apache/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/apache/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/kafka/merges","archive_url":"https://api.github.com/repos/apache/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/kafka/downloads","issues_url":"https://api.github.com/repos/apache/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/apache/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/kafka/labels{/name}","releases_url":"https://api.github.com/repos/apache/kafka/releases{/id}","deployments_url":"https://api.github.com/repos/apache/kafka/deployments","created_at":"2011-08-15T18:06:16Z","updated_at":"2018-04-10T19:50:11Z","pushed_at":"2018-04-10T21:38:41Z","git_url":"git://github.com/apache/kafka.git","ssh_url":"git@github.com:apache/kafka.git","clone_url":"https://github.com/apache/kafka.git","svn_url":"https://github.com/apache/kafka","homepage":null,"size":64272,"stargazers_count":7944,"watchers_count":7944,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":4693,"mirror_url":null,"archived":false,"open_issues_count":443,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0"},"forks":4693,"open_issues":443,"watchers":7944,"default_branch":"trunk"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/4744"},"html":{"href":"https://github.com/apache/kafka/pull/4744"},"issue":{"href":"https://api.github.com/repos/apache/kafka/issues/4744"},"comments":{"href":"https://api.github.com/repos/apache/kafka/issues/4744/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/kafka/pulls/4744/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/kafka/pulls/4744/commits"},"statuses":{"href":"https://api.github.com/repos/apache/kafka/statuses/6a1bdd02a086e991643a070738d3a33ce50213d0"}},"author_association":"CONTRIBUTOR"}}
{"actor":{"display_login":"mjsax"}}