MINOR: Reduce allocations in requests via buffer caching

[ijuma]

Use a caching BufferSupplier per request handler thread so that
decompression buffers are cached if supported by the underlying
CompressionType. This achieves a similar outcome as #9220, but
with less contention.

We introduce a RequestLocal class to make it easier to introduce
new request scoped stateful instances (one example we discussed
previously was an ActionQueue that could be used to avoid
some of the complex group coordinator locking).

This is a small win for zstd (no synchronization or soft references) and
a more significant win for lz4. In particular, it reduces allocations
significantly 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).

Default arguments will be removed in a subsequent PR.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[chia7712]

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 :)

[chia7712]

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)

[ijuma]

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.

[ijuma]

@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?

[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?

It is great! +1

[chia7712]

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

[chia7712]

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

[ijuma]

@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?

[chia7712]

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

[chia7712]

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".

[ijuma]

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

[ijuma]

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

[chia7712]

We already have a RequestContext class, so another name would be RequestLocal (similar to ThreadLocal).

’RequestLocal’ is good to me

[ijuma]

@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?

[chia7712]

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.

[ijuma]

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?

[ijuma]

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

[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?

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.

[chia7712]

@ijuma thanks for this improvement. overall +1. Could you run performance test?

[chia7712]

NoCaching should use BufferSupplier.NO_CACHING, right?

[ijuma]

Indeed.

[chia7712]

Should we call completeShutdown() rather than shutdownComplete.countDown()?

[ijuma]

Yes.

[ijuma]

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

[chia7712]

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)

[ijuma]

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}
--------------------------------------------------------------------------------

[ijuma]

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:

this PR:

So, I think we can go ahead and merge this.