You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2020/08/29 20:14:45 UTC

[GitHub] [kafka] ijuma opened a new pull request #9229: MINOR: Reduce allocations in requests via buffer caching

ijuma opened a new pull request #9229:
URL: https://github.com/apache/kafka/pull/9229


   Use a caching BufferSupplier per request handler thread so that
   decompression buffers are cached if supported by the underlying
   CompressionType. This reduces allocations significantly for LZ4 when the
   number of partitions is high. The decompression buffer size is typically
   64 KB, so a produce request with 1000 partitions results in 64 MB of
   allocations even if each produce batch is small (likely, when there are
   so many partitions).
   
   I did a quick producer perf local test with 5000 partitions, 1 KB record
   size,
   1 broker, lz4 and ~0.5 for the producer compression rate metric:
   
   Before this change:
   > 20000000 records sent, 346314.349535 records/sec (330.27 MB/sec),
   148.33 ms avg latency, 2267.00 ms max latency, 115 ms 50th, 383 ms 95th,
   777 ms 99th, 1514 ms 99.9th.
   
   After this change:
   > 20000000 records sent, 431956.113259 records/sec (411.95 MB/sec),
   117.79 ms avg latency, 1219.00 ms max latency, 99 ms 50th, 295 ms 95th,
   440 ms 99th, 662 ms 99.9th.
   
   That's a 25% throughput improvement and p999 latency was reduced to
   under half (in this test).
   
   TODO: Remove default arguments
   
   ### Committer Checklist (excluded from commit message)
   - [ ] Verify design and implementation 
   - [ ] Verify test coverage and CI build status
   - [ ] Verify documentation (including upgrade notes)
   


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] ijuma commented on a change in pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#discussion_r607129364



##########
File path: core/src/main/scala/kafka/server/RequestLocal.scala
##########
@@ -0,0 +1,24 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.server
+
+import org.apache.kafka.common.utils.BufferSupplier
+
+case class RequestLocal(bufferSupplier: BufferSupplier) {

Review comment:
       Add documentation.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] ijuma edited a comment on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
ijuma edited a comment on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-814090052


   > Could we unify the interface?
   
   @chia7712 Yes, it's worth exploring. I think `MemoryPool` is intended to be a thread-safe cache, so it's not trivial, but it may be possible. Are you interested in looking into that?


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] ijuma commented on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-683352094


   In my opinion, thread locals are most useful when one doesn't control the code. For cases like this, being explicit makes it easier to reason about and also test. Even if it's a bit more work initially.


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] chia7712 commented on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-683400436


   @ijuma I have closed #9220 and assign https://issues.apache.org/jira/browse/KAFKA-10433 to you.


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] ijuma edited a comment on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
ijuma edited a comment on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-850747419


   This PR:
   
   ```
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_consumer_throughput.security_protocol=PLAINTEXT.compression_type=lz4
   status:     PASS
   run time:   1 minute 16.243 seconds
   {"records_per_sec": 2044571.6622, "mb_per_sec": 194.9855}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_consumer_throughput.security_protocol=PLAINTEXT.compression_type=zstd
   status:     PASS
   run time:   1 minute 19.227 seconds
   {"records_per_sec": 1779992.88, "mb_per_sec": 169.7533}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_and_consumer.security_protocol=PLAINTEXT.compression_type=lz4
   status:     PASS
   run time:   1 minute 13.064 seconds
   {"producer": {"records_per_sec": 402868.423173, "mb_per_sec": 38.42}, "consumer": {"records_per_sec": 408363.28, "mb_per_sec": 38.9446}}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_and_consumer.security_protocol=PLAINTEXT.compression_type=zstd
   status:     PASS
   run time:   1 minute 12.112 seconds
   {"producer": {"records_per_sec": 347886.588972, "mb_per_sec": 33.18}, "consumer": {"records_per_sec": 352534.7247, "mb_per_sec": 33.6203}}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_end_to_end_latency.security_protocol=PLAINTEXT.compression_type=lz4
   status:     PASS
   run time:   51.120 seconds
   {"latency_50th_ms": 0.0, "latency_99th_ms": 3.0, "latency_999th_ms": 8.0}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_end_to_end_latency.security_protocol=PLAINTEXT.compression_type=zstd
   status:     PASS
   run time:   45.992 seconds
   {"latency_50th_ms": 0.0, "latency_99th_ms": 3.0, "latency_999th_ms": 9.0}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_long_term_producer_throughput.security_protocol=PLAINTEXT.compression_type=lz4
   status:     PASS
   run time:   1 minute 11.957 seconds
   {"0": {"records_per_sec": 400994.466276, "mb_per_sec": 38.24}}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_long_term_producer_throughput.security_protocol=PLAINTEXT.compression_type=zstd
   status:     PASS
   run time:   1 minute 12.859 seconds
   {"0": {"records_per_sec": 366716.784627, "mb_per_sec": 34.97}}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=lz4.message_size=10
   status:     PASS
   run time:   55.828 seconds
   {"records_per_sec": 1101318.782309, "mb_per_sec": 10.5}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=lz4.message_size=100
   status:     PASS
   run time:   44.917 seconds
   {"records_per_sec": 373345.479833, "mb_per_sec": 35.6}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=lz4.message_size=1000
   status:     PASS
   run time:   45.609 seconds
   {"records_per_sec": 63912.857143, "mb_per_sec": 60.95}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=lz4.message_size=10000
   status:     PASS
   run time:   45.665 seconds
   {"records_per_sec": 8099.57755, "mb_per_sec": 77.24}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=lz4.message_size=100000
   status:     PASS
   run time:   42.768 seconds
   {"records_per_sec": 1127.731092, "mb_per_sec": 107.55}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=zstd.message_size=10
   status:     PASS
   run time:   55.771 seconds
   {"records_per_sec": 1051286.284953, "mb_per_sec": 10.03}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=zstd.message_size=100
   status:     PASS
   run time:   48.832 seconds
   {"records_per_sec": 331319.921007, "mb_per_sec": 31.6}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=zstd.message_size=1000
   status:     PASS
   run time:   46.853 seconds
   {"records_per_sec": 67615.617128, "mb_per_sec": 64.48}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=zstd.message_size=10000
   status:     PASS
   run time:   43.870 seconds
   {"records_per_sec": 7647.293447, "mb_per_sec": 72.93}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=zstd.message_size=100000
   status:     PASS
   run time:   43.045 seconds
   {"records_per_sec": 1222.222222, "mb_per_sec": 116.56}
   --------------------------------------------------------------------------------
   ```
   
   trunk:
   
   ```
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_consumer_throughput.security_protocol=PLAINTEXT.compression_type=lz4
   status:     PASS
   run time:   1 minute 20.214 seconds
   {"records_per_sec": 1927153.5941, "mb_per_sec": 183.7877}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_consumer_throughput.security_protocol=PLAINTEXT.compression_type=zstd
   status:     PASS
   run time:   1 minute 16.091 seconds
   {"records_per_sec": 1754693.8059, "mb_per_sec": 167.3406}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_and_consumer.security_protocol=PLAINTEXT.compression_type=lz4
   status:     PASS
   run time:   1 minute 3.464 seconds
   {"producer": {"records_per_sec": 472009.817804, "mb_per_sec": 45.01}, "consumer": {"records_per_sec": 480676.7929, "mb_per_sec": 45.8409}}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_and_consumer.security_protocol=PLAINTEXT.compression_type=zstd
   status:     PASS
   run time:   1 minute 12.971 seconds
   {"producer": {"records_per_sec": 346392.323946, "mb_per_sec": 33.03}, "consumer": {"records_per_sec": 350741.8189, "mb_per_sec": 33.4493}}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_end_to_end_latency.security_protocol=PLAINTEXT.compression_type=lz4
   status:     PASS
   run time:   48.883 seconds
   {"latency_999th_ms": 9.0, "latency_99th_ms": 3.0, "latency_50th_ms": 0.0}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_end_to_end_latency.security_protocol=PLAINTEXT.compression_type=zstd
   status:     PASS
   run time:   47.810 seconds
   {"latency_999th_ms": 9.0, "latency_99th_ms": 3.0, "latency_50th_ms": 0.0}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_long_term_producer_throughput.security_protocol=PLAINTEXT.compression_type=lz4
   status:     PASS
   run time:   1 minute 6.919 seconds
   {"0": {"records_per_sec": 426894.34365, "mb_per_sec": 40.71}}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_long_term_producer_throughput.security_protocol=PLAINTEXT.compression_type=zstd
   status:     PASS
   run time:   1 minute 8.106 seconds
   {"0": {"records_per_sec": 369617.445943, "mb_per_sec": 35.25}}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=lz4.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=10
   status:     PASS
   run time:   54.631 seconds
   {"records_per_sec": 1138306.504961, "mb_per_sec": 10.86}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=lz4.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=100
   status:     PASS
   run time:   49.852 seconds
   {"records_per_sec": 429909.352979, "mb_per_sec": 41.0}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=lz4.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=1000
   status:     PASS
   run time:   46.940 seconds
   {"records_per_sec": 67276.691729, "mb_per_sec": 64.16}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=lz4.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=10000
   status:     PASS
   run time:   44.910 seconds
   {"records_per_sec": 8114.26844, "mb_per_sec": 77.38}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=lz4.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=100000
   status:     PASS
   run time:   45.013 seconds
   {"records_per_sec": 1166.956522, "mb_per_sec": 111.29}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=zstd.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=10
   status:     PASS
   run time:   55.821 seconds
   {"records_per_sec": 1123630.975303, "mb_per_sec": 10.72}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=zstd.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=100
   status:     PASS
   run time:   46.809 seconds
   {"records_per_sec": 342043.068298, "mb_per_sec": 32.62}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=zstd.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=1000
   status:     PASS
   run time:   44.890 seconds
   {"records_per_sec": 63012.676056, "mb_per_sec": 60.09}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=zstd.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=10000
   status:     PASS
   run time:   48.846 seconds
   {"records_per_sec": 7501.9564, "mb_per_sec": 71.54}
   --------------------------------------------------------------------------------
   test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=zstd.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=100000
   status:     PASS
   run time:   46.991 seconds
   {"records_per_sec": 1142.12766, "mb_per_sec": 108.92}
   --------------------------------------------------------------------------------
   ```
   
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] chia7712 commented on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-683343524


   This patch makes each request (handler) thread have a ```BufferSupplier``` to simplify concurrency handling (by contrast, #9220 offers a thread-safe BufferSupplier). 
   
   This idea is good to me :)


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] chia7712 commented on a change in pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#discussion_r608321139



##########
File path: core/src/main/scala/kafka/server/RequestLocal.scala
##########
@@ -0,0 +1,37 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.server
+
+import org.apache.kafka.common.utils.BufferSupplier
+
+object RequestLocal {
+  val NoCaching: RequestLocal = RequestLocal(BufferSupplier.create())

Review comment:
       `NoCaching` should use `BufferSupplier.NO_CACHING`, right?

##########
File path: core/src/main/scala/kafka/server/KafkaRequestHandler.scala
##########
@@ -64,14 +65,14 @@ class KafkaRequestHandler(id: Int,
       req match {
         case RequestChannel.ShutdownRequest =>
           debug(s"Kafka request handler $id on broker $brokerId received shut down command")
-          shutdownComplete.countDown()
+          completeShutdown()
           return
 
         case request: RequestChannel.Request =>
           try {
             request.requestDequeueTimeNanos = endTime
             trace(s"Kafka request handler $id on broker $brokerId handling request $request")
-            apis.handle(request)
+            apis.handle(request, requestLocal)
           } catch {
             case e: FatalExitError =>
               shutdownComplete.countDown()

Review comment:
       Should we call `completeShutdown()` rather than `shutdownComplete.countDown()`?




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] chia7712 commented on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-814579521


   >  Yes, it's worth exploring. I think MemoryPool is intended to be a thread-safe cache, so it's not trivial, but it may be possible. Are you interested in looking into that?
   
   yes, that is an interesting issue to me. I file a JIRA (https://issues.apache.org/jira/browse/KAFKA-12627). That can be a follow-up of this PR.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] ijuma merged pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
ijuma merged pull request #9229:
URL: https://github.com/apache/kafka/pull/9229


   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] ijuma commented on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-814090052


   @chia7712 Yes, it's worth exploring. I think `MemoryPool` is intended to be a thread-safe cache, so it's not trivial, but it may be possible. Are you interested in looking into that?


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] ijuma commented on a change in pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#discussion_r607129364



##########
File path: core/src/main/scala/kafka/server/RequestLocal.scala
##########
@@ -0,0 +1,24 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.server
+
+import org.apache.kafka.common.utils.BufferSupplier
+
+case class RequestLocal(bufferSupplier: BufferSupplier) {

Review comment:
       Add documentation.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] ijuma commented on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-850747419


   This PR:
   
   > ================================================================================
   > SESSION REPORT (ALL TESTS)
   > ducktape version: 0.8.1
   > session_id:       2021-05-28--005
   > run time:         16 minutes 56.024 seconds
   > tests run:        18
   > passed:           18
   > failed:           0
   > ignored:          0
   > ================================================================================
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_consumer_throughput.security_protocol=PLAINTEXT.compression_type=lz4
   > status:     PASS
   > run time:   1 minute 16.243 seconds
   > {"records_per_sec": 2044571.6622, "mb_per_sec": 194.9855}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_consumer_throughput.security_protocol=PLAINTEXT.compression_type=zstd
   > status:     PASS
   > run time:   1 minute 19.227 seconds
   > {"records_per_sec": 1779992.88, "mb_per_sec": 169.7533}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_and_consumer.security_protocol=PLAINTEXT.compression_type=lz4
   > status:     PASS
   > run time:   1 minute 13.064 seconds
   > {"producer": {"records_per_sec": 402868.423173, "mb_per_sec": 38.42}, "consumer": {"records_per_sec": 408363.28, "mb_per_sec": 38.9446}}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_and_consumer.security_protocol=PLAINTEXT.compression_type=zstd
   > status:     PASS
   > run time:   1 minute 12.112 seconds
   > {"producer": {"records_per_sec": 347886.588972, "mb_per_sec": 33.18}, "consumer": {"records_per_sec": 352534.7247, "mb_per_sec": 33.6203}}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_end_to_end_latency.security_protocol=PLAINTEXT.compression_type=lz4
   > status:     PASS
   > run time:   51.120 seconds
   > {"latency_50th_ms": 0.0, "latency_99th_ms": 3.0, "latency_999th_ms": 8.0}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_end_to_end_latency.security_protocol=PLAINTEXT.compression_type=zstd
   > status:     PASS
   > run time:   45.992 seconds
   > {"latency_50th_ms": 0.0, "latency_99th_ms": 3.0, "latency_999th_ms": 9.0}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_long_term_producer_throughput.security_protocol=PLAINTEXT.compression_type=lz4
   > status:     PASS
   > run time:   1 minute 11.957 seconds
   > {"0": {"records_per_sec": 400994.466276, "mb_per_sec": 38.24}}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_long_term_producer_throughput.security_protocol=PLAINTEXT.compression_type=zstd
   > status:     PASS
   > run time:   1 minute 12.859 seconds
   > {"0": {"records_per_sec": 366716.784627, "mb_per_sec": 34.97}}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=lz4.message_size=10
   > status:     PASS
   > run time:   55.828 seconds
   > {"records_per_sec": 1101318.782309, "mb_per_sec": 10.5}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=lz4.message_size=100
   > status:     PASS
   > run time:   44.917 seconds
   > {"records_per_sec": 373345.479833, "mb_per_sec": 35.6}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=lz4.message_size=1000
   > status:     PASS
   > run time:   45.609 seconds
   > {"records_per_sec": 63912.857143, "mb_per_sec": 60.95}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=lz4.message_size=10000
   > status:     PASS
   > run time:   45.665 seconds
   > {"records_per_sec": 8099.57755, "mb_per_sec": 77.24}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=lz4.message_size=100000
   > status:     PASS
   > run time:   42.768 seconds
   > {"records_per_sec": 1127.731092, "mb_per_sec": 107.55}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=zstd.message_size=10
   > status:     PASS
   > run time:   55.771 seconds
   > {"records_per_sec": 1051286.284953, "mb_per_sec": 10.03}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=zstd.message_size=100
   > status:     PASS
   > run time:   48.832 seconds
   > {"records_per_sec": 331319.921007, "mb_per_sec": 31.6}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=zstd.message_size=1000
   > status:     PASS
   > run time:   46.853 seconds
   > {"records_per_sec": 67615.617128, "mb_per_sec": 64.48}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=zstd.message_size=10000
   > status:     PASS
   > run time:   43.870 seconds
   > {"records_per_sec": 7647.293447, "mb_per_sec": 72.93}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=zstd.message_size=100000
   > status:     PASS
   > run time:   43.045 seconds
   > {"records_per_sec": 1222.222222, "mb_per_sec": 116.56}
   > --------------------------------------------------------------------------------
   
   trunk:
   
   > ================================================================================
   > SESSION REPORT (ALL TESTS)
   > ducktape version: 0.8.1
   > session_id:       2021-05-28--007
   > run time:         16 minutes 49.452 seconds
   > tests run:        18
   > passed:           18
   > failed:           0
   > ignored:          0
   > ================================================================================
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_consumer_throughput.security_protocol=PLAINTEXT.compression_type=lz4
   > status:     PASS
   > run time:   1 minute 20.214 seconds
   > {"records_per_sec": 1927153.5941, "mb_per_sec": 183.7877}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_consumer_throughput.security_protocol=PLAINTEXT.compression_type=zstd
   > status:     PASS
   > run time:   1 minute 16.091 seconds
   > {"records_per_sec": 1754693.8059, "mb_per_sec": 167.3406}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_and_consumer.security_protocol=PLAINTEXT.compression_type=lz4
   > status:     PASS
   > run time:   1 minute 3.464 seconds
   > {"producer": {"records_per_sec": 472009.817804, "mb_per_sec": 45.01}, "consumer": {"records_per_sec": 480676.7929, "mb_per_sec": 45.8409}}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_and_consumer.security_protocol=PLAINTEXT.compression_type=zstd
   > status:     PASS
   > run time:   1 minute 12.971 seconds
   > {"producer": {"records_per_sec": 346392.323946, "mb_per_sec": 33.03}, "consumer": {"records_per_sec": 350741.8189, "mb_per_sec": 33.4493}}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_end_to_end_latency.security_protocol=PLAINTEXT.compression_type=lz4
   > status:     PASS
   > run time:   48.883 seconds
   > {"latency_999th_ms": 9.0, "latency_99th_ms": 3.0, "latency_50th_ms": 0.0}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_end_to_end_latency.security_protocol=PLAINTEXT.compression_type=zstd
   > status:     PASS
   > run time:   47.810 seconds
   > {"latency_999th_ms": 9.0, "latency_99th_ms": 3.0, "latency_50th_ms": 0.0}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_long_term_producer_throughput.security_protocol=PLAINTEXT.compression_type=lz4
   > status:     PASS
   > run time:   1 minute 6.919 seconds
   > {"0": {"records_per_sec": 426894.34365, "mb_per_sec": 40.71}}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_long_term_producer_throughput.security_protocol=PLAINTEXT.compression_type=zstd
   > status:     PASS
   > run time:   1 minute 8.106 seconds
   > {"0": {"records_per_sec": 369617.445943, "mb_per_sec": 35.25}}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=lz4.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=10
   > status:     PASS
   > run time:   54.631 seconds
   > {"records_per_sec": 1138306.504961, "mb_per_sec": 10.86}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=lz4.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=100
   > status:     PASS
   > run time:   49.852 seconds
   > {"records_per_sec": 429909.352979, "mb_per_sec": 41.0}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=lz4.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=1000
   > status:     PASS
   > run time:   46.940 seconds
   > {"records_per_sec": 67276.691729, "mb_per_sec": 64.16}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=lz4.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=10000
   > status:     PASS
   > run time:   44.910 seconds
   > {"records_per_sec": 8114.26844, "mb_per_sec": 77.38}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=lz4.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=100000
   > status:     PASS
   > run time:   45.013 seconds
   > {"records_per_sec": 1166.956522, "mb_per_sec": 111.29}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=zstd.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=10
   > status:     PASS
   > run time:   55.821 seconds
   > {"records_per_sec": 1123630.975303, "mb_per_sec": 10.72}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=zstd.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=100
   > status:     PASS
   > run time:   46.809 seconds
   > {"records_per_sec": 342043.068298, "mb_per_sec": 32.62}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=zstd.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=1000
   > status:     PASS
   > run time:   44.890 seconds
   > {"records_per_sec": 63012.676056, "mb_per_sec": 60.09}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=zstd.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=10000
   > status:     PASS
   > run time:   48.846 seconds
   > {"records_per_sec": 7501.9564, "mb_per_sec": 71.54}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=zstd.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=100000
   > status:     PASS
   > run time:   46.991 seconds
   > {"records_per_sec": 1142.12766, "mb_per_sec": 108.92}
   > --------------------------------------------------------------------------------
   
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] ijuma commented on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-683352558


   @chia7712 One option would be for me to introduce a `RequestContext` case class and add the `BufferSupplier` as one of the fields. It would be easy to extend this class with request bound elements like `ActionQueue`. Thoughts?


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] chia7712 commented on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-767675140


   > One option would be for me to introduce a RequestContext case class and add the BufferSupplier
   
   @ijuma Do you lean toward to implement this? It can help us improve "action queue".


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] chia7712 commented on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-695199506


   @ijuma I had filed a PR to zstd-jni to open the door for reusing byte array of zstd compression (https://github.com/luben/zstd-jni/commit/1346fc1cc6bce27a5f7aed2563402ec20d672bd6). Also, there is a ticket (https://issues.apache.org/jira/browse/KAFKA-10470) which will apply the new API of zstd-jni.


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] chia7712 commented on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-767671568


   > I am leaning towards doing it as a separate PR and maybe after the 2.8 branch is cut (to avoid disrupting other work targeting 2.8).
   
   +1


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] ijuma commented on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-813439278


   @chia7712 I introduced `RequestLocal` as discussed. Does this seem reasonable to you? If so, I propose the following next steps:
   
   1. In this PR, provide utility methods in `RequestLocal` for the two common defaults: `ThreadLocalCaching` and `NoCaching`. The latter should be used when the usage is not guaranteed to be within the same thread. In the future, we can consider a `ThreadSafeCaching`/`GlobalCaching` option, if that makes sense.
   
   2. In a separate PR, remove the default arguments. This will result in a lot of test changes, but no change in behavior. So, it probably makes sense to review separately.
   
   Thoughts?


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] chia7712 commented on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-778864878


   > We already have a RequestContext class, so another name would be RequestLocal (similar to ThreadLocal).
   
   ’RequestLocal’ is good to me


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] ijuma commented on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-851047311


   The results are similar for the ducktape benchmarks since the bottleneck is elsewhere. In the PR description, I include the results for a workload that shows significant improvement with these changes. Also, the following allocation profiles show that the the lz4 buffer allocations dominate trunk and are gone in this PR:
   
   trunk:
   ![image](https://user-images.githubusercontent.com/24747/120117066-310dfc80-c140-11eb-8ad4-490e749e2162.png)
   
   this PR:
   ![image](https://user-images.githubusercontent.com/24747/120117071-4b47da80-c140-11eb-9292-8dc586215245.png)
   
   So, I think we can go ahead and merge this.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] ijuma commented on a change in pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#discussion_r608341480



##########
File path: core/src/main/scala/kafka/server/KafkaRequestHandler.scala
##########
@@ -64,14 +65,14 @@ class KafkaRequestHandler(id: Int,
       req match {
         case RequestChannel.ShutdownRequest =>
           debug(s"Kafka request handler $id on broker $brokerId received shut down command")
-          shutdownComplete.countDown()
+          completeShutdown()
           return
 
         case request: RequestChannel.Request =>
           try {
             request.requestDequeueTimeNanos = endTime
             trace(s"Kafka request handler $id on broker $brokerId handling request $request")
-            apis.handle(request)
+            apis.handle(request, requestLocal)
           } catch {
             case e: FatalExitError =>
               shutdownComplete.countDown()

Review comment:
       Yes.

##########
File path: core/src/main/scala/kafka/server/RequestLocal.scala
##########
@@ -0,0 +1,37 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.server
+
+import org.apache.kafka.common.utils.BufferSupplier
+
+object RequestLocal {
+  val NoCaching: RequestLocal = RequestLocal(BufferSupplier.create())

Review comment:
       Indeed.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] ijuma commented on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-814999556


   @chia7712 When you say performance test, you mean the ducktape ones?


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] chia7712 commented on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-813899837


   > I introduced RequestLocal as discussed. Does this seem reasonable to you? If so, I propose the following next steps:
   
   That LGTM. For another, the memory utils used by `KafkaRequestHandler` (BufferSupplier) is different from `Processor` (MemoryPool). Could we unify the interface? It seems to me `RequestLocal` can be applied to `Processor` as well in the future.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] chia7712 commented on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-815002256


   > When you say performance test, you mean the ducktape ones?
   
   yep. maybe `benchmark_test.py` (although I feel the benefit gets more obvious in long-run production)


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] ijuma commented on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-767662119


   @chia7712 One question we have to decide is whether we want to remove the default arguments in this PR or in a separate PR that is purely mechanical (no behavior changes). A lot of tests call the relevant methods, so removing the defaults would cause a lot of test changes. I am leaning towards doing it as a separate PR and maybe after the 2.8 branch is cut (to avoid disrupting other work targeting 2.8). What do you think?


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] chia7712 commented on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-683347067


   Could we use ThreadLocal to keep those thread resources, like BufferSupplier and ActionQueue, to simplify the method arguments? The cost of ThreadLocal is low and it is easy to add new thread local resource in the future (and we don’t need to changes a lot of method arguments)


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] ijuma edited a comment on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
ijuma edited a comment on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-850747419


   This PR:
   
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_consumer_throughput.security_protocol=PLAINTEXT.compression_type=lz4
   > status:     PASS
   > run time:   1 minute 16.243 seconds
   > {"records_per_sec": 2044571.6622, "mb_per_sec": 194.9855}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_consumer_throughput.security_protocol=PLAINTEXT.compression_type=zstd
   > status:     PASS
   > run time:   1 minute 19.227 seconds
   > {"records_per_sec": 1779992.88, "mb_per_sec": 169.7533}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_and_consumer.security_protocol=PLAINTEXT.compression_type=lz4
   > status:     PASS
   > run time:   1 minute 13.064 seconds
   > {"producer": {"records_per_sec": 402868.423173, "mb_per_sec": 38.42}, "consumer": {"records_per_sec": 408363.28, "mb_per_sec": 38.9446}}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_and_consumer.security_protocol=PLAINTEXT.compression_type=zstd
   > status:     PASS
   > run time:   1 minute 12.112 seconds
   > {"producer": {"records_per_sec": 347886.588972, "mb_per_sec": 33.18}, "consumer": {"records_per_sec": 352534.7247, "mb_per_sec": 33.6203}}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_end_to_end_latency.security_protocol=PLAINTEXT.compression_type=lz4
   > status:     PASS
   > run time:   51.120 seconds
   > {"latency_50th_ms": 0.0, "latency_99th_ms": 3.0, "latency_999th_ms": 8.0}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_end_to_end_latency.security_protocol=PLAINTEXT.compression_type=zstd
   > status:     PASS
   > run time:   45.992 seconds
   > {"latency_50th_ms": 0.0, "latency_99th_ms": 3.0, "latency_999th_ms": 9.0}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_long_term_producer_throughput.security_protocol=PLAINTEXT.compression_type=lz4
   > status:     PASS
   > run time:   1 minute 11.957 seconds
   > {"0": {"records_per_sec": 400994.466276, "mb_per_sec": 38.24}}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_long_term_producer_throughput.security_protocol=PLAINTEXT.compression_type=zstd
   > status:     PASS
   > run time:   1 minute 12.859 seconds
   > {"0": {"records_per_sec": 366716.784627, "mb_per_sec": 34.97}}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=lz4.message_size=10
   > status:     PASS
   > run time:   55.828 seconds
   > {"records_per_sec": 1101318.782309, "mb_per_sec": 10.5}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=lz4.message_size=100
   > status:     PASS
   > run time:   44.917 seconds
   > {"records_per_sec": 373345.479833, "mb_per_sec": 35.6}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=lz4.message_size=1000
   > status:     PASS
   > run time:   45.609 seconds
   > {"records_per_sec": 63912.857143, "mb_per_sec": 60.95}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=lz4.message_size=10000
   > status:     PASS
   > run time:   45.665 seconds
   > {"records_per_sec": 8099.57755, "mb_per_sec": 77.24}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=lz4.message_size=100000
   > status:     PASS
   > run time:   42.768 seconds
   > {"records_per_sec": 1127.731092, "mb_per_sec": 107.55}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=zstd.message_size=10
   > status:     PASS
   > run time:   55.771 seconds
   > {"records_per_sec": 1051286.284953, "mb_per_sec": 10.03}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=zstd.message_size=100
   > status:     PASS
   > run time:   48.832 seconds
   > {"records_per_sec": 331319.921007, "mb_per_sec": 31.6}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=zstd.message_size=1000
   > status:     PASS
   > run time:   46.853 seconds
   > {"records_per_sec": 67615.617128, "mb_per_sec": 64.48}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=zstd.message_size=10000
   > status:     PASS
   > run time:   43.870 seconds
   > {"records_per_sec": 7647.293447, "mb_per_sec": 72.93}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.compression_type=zstd.message_size=100000
   > status:     PASS
   > run time:   43.045 seconds
   > {"records_per_sec": 1222.222222, "mb_per_sec": 116.56}
   > --------------------------------------------------------------------------------
   
   trunk:
   
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_consumer_throughput.security_protocol=PLAINTEXT.compression_type=lz4
   > status:     PASS
   > run time:   1 minute 20.214 seconds
   > {"records_per_sec": 1927153.5941, "mb_per_sec": 183.7877}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_consumer_throughput.security_protocol=PLAINTEXT.compression_type=zstd
   > status:     PASS
   > run time:   1 minute 16.091 seconds
   > {"records_per_sec": 1754693.8059, "mb_per_sec": 167.3406}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_and_consumer.security_protocol=PLAINTEXT.compression_type=lz4
   > status:     PASS
   > run time:   1 minute 3.464 seconds
   > {"producer": {"records_per_sec": 472009.817804, "mb_per_sec": 45.01}, "consumer": {"records_per_sec": 480676.7929, "mb_per_sec": 45.8409}}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_and_consumer.security_protocol=PLAINTEXT.compression_type=zstd
   > status:     PASS
   > run time:   1 minute 12.971 seconds
   > {"producer": {"records_per_sec": 346392.323946, "mb_per_sec": 33.03}, "consumer": {"records_per_sec": 350741.8189, "mb_per_sec": 33.4493}}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_end_to_end_latency.security_protocol=PLAINTEXT.compression_type=lz4
   > status:     PASS
   > run time:   48.883 seconds
   > {"latency_999th_ms": 9.0, "latency_99th_ms": 3.0, "latency_50th_ms": 0.0}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_end_to_end_latency.security_protocol=PLAINTEXT.compression_type=zstd
   > status:     PASS
   > run time:   47.810 seconds
   > {"latency_999th_ms": 9.0, "latency_99th_ms": 3.0, "latency_50th_ms": 0.0}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_long_term_producer_throughput.security_protocol=PLAINTEXT.compression_type=lz4
   > status:     PASS
   > run time:   1 minute 6.919 seconds
   > {"0": {"records_per_sec": 426894.34365, "mb_per_sec": 40.71}}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_long_term_producer_throughput.security_protocol=PLAINTEXT.compression_type=zstd
   > status:     PASS
   > run time:   1 minute 8.106 seconds
   > {"0": {"records_per_sec": 369617.445943, "mb_per_sec": 35.25}}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=lz4.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=10
   > status:     PASS
   > run time:   54.631 seconds
   > {"records_per_sec": 1138306.504961, "mb_per_sec": 10.86}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=lz4.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=100
   > status:     PASS
   > run time:   49.852 seconds
   > {"records_per_sec": 429909.352979, "mb_per_sec": 41.0}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=lz4.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=1000
   > status:     PASS
   > run time:   46.940 seconds
   > {"records_per_sec": 67276.691729, "mb_per_sec": 64.16}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=lz4.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=10000
   > status:     PASS
   > run time:   44.910 seconds
   > {"records_per_sec": 8114.26844, "mb_per_sec": 77.38}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=lz4.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=100000
   > status:     PASS
   > run time:   45.013 seconds
   > {"records_per_sec": 1166.956522, "mb_per_sec": 111.29}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=zstd.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=10
   > status:     PASS
   > run time:   55.821 seconds
   > {"records_per_sec": 1123630.975303, "mb_per_sec": 10.72}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=zstd.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=100
   > status:     PASS
   > run time:   46.809 seconds
   > {"records_per_sec": 342043.068298, "mb_per_sec": 32.62}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=zstd.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=1000
   > status:     PASS
   > run time:   44.890 seconds
   > {"records_per_sec": 63012.676056, "mb_per_sec": 60.09}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=zstd.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=10000
   > status:     PASS
   > run time:   48.846 seconds
   > {"records_per_sec": 7501.9564, "mb_per_sec": 71.54}
   > --------------------------------------------------------------------------------
   > test_id:    kafkatest.benchmarks.core.benchmark_test.Benchmark.test_producer_throughput.acks=1.compression_type=zstd.security_protocol=PLAINTEXT.topic=topic-replication-factor-three.message_size=100000
   > status:     PASS
   > run time:   46.991 seconds
   > {"records_per_sec": 1142.12766, "mb_per_sec": 108.92}
   > --------------------------------------------------------------------------------
   
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] ijuma commented on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-778804930


   We already have a `RequestContext` class, so another name would be `RequestLocal` (similar to `ThreadLocal`). Thoughts @chia7712?


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] ijuma commented on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-767687131


   @chia7712 I had forgotten that part of the discussion. :) Let me take a closer look at that.


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] chia7712 commented on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-683369221


   > One option would be for me to introduce a RequestContext case class and add the BufferSupplier as one of the fields. It would be easy to extend this class with request bound elements like ActionQueue. Thoughts?
   
   It is great! +1


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] ijuma commented on pull request #9229: MINOR: Reduce allocations in requests via buffer caching

Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9229:
URL: https://github.com/apache/kafka/pull/9229#issuecomment-814141618


   @chia7712 I updated the PR. Please review when you have a chance.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org