KAFKA-10684; Avoid additional envelope copies during network transmission

[hachikuji]

This patch creates a new SendBuilder class which allows us to avoid copying "bytes" types when transmitting an api message over the network. This is used in EnvelopeRequest and EnvelopeResponse to avoid copying the embedded data.

The patch also contains a few minor cleanups such as moving envelope parsing logic into RequestContext.

Committer Checklist (excluded from commit message)

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

[chia7712]

Does this improvement work for other requests?

[hachikuji]

@chia7712 I think it could. I was actually thinking about #9401 when I was working on it. As far as I can tell, the client currently copies the whole produce buffer when the request reaches NetworkClient. One remaining thing to do here I think is converge SendBuilder and RecordsWritable, which are using similar tricks.

[hachikuji]

I got inspired to try and extend this to apply to all message types. I've updated the patch to remove the custom response logic in FetchResponse in favor of a general pattern using SendBuilder. We had an existing benchmark FetchResponseBenchmark.testSerializeFetchResponse, so I tried it out.

Here is the results before the patch (10 topics and 500 topics):

Result "org.apache.kafka.jmh.common.FetchResponseBenchmark.testSerializeFetchResponse":
  6322.861 ±(99.9%) 310.785 ns/op [Average]
  (min, avg, max) = (6057.758, 6322.861, 7090.658), stdev = 290.708
  CI (99.9%): [6012.076, 6633.646] (assumes normal distribution)

Result "org.apache.kafka.jmh.common.FetchResponseBenchmark.testSerializeFetchResponse":
  323310.283 ±(99.9%) 25947.515 ns/op [Average]
  (min, avg, max) = (301370.273, 323310.283, 383716.556), stdev = 24271.322
  CI (99.9%): [297362.768, 349257.799] (assumes normal distribution)

Here is the new benchmark (10 topics and 500 topics):

Result "org.apache.kafka.jmh.common.FetchResponseBenchmark.testSerializeFetchResponse":
  5701.378 ±(99.9%) 100.848 ns/op [Average]
  (min, avg, max) = (5601.838, 5701.378, 5925.943), stdev = 94.333
  CI (99.9%): [5600.530, 5802.225] (assumes normal distribution)

Result "org.apache.kafka.jmh.common.FetchResponseBenchmark.testSerializeFetchResponse":
  298221.825 ±(99.9%) 8173.945 ns/op [Average]
  (min, avg, max) = (287615.891, 298221.825, 321499.618), stdev = 7645.913
  CI (99.9%): [290047.880, 306395.770] (assumes normal distribution)

So looks like a modest overall improvement.

Note I still need to polish up a few things in the PR.

[chia7712]

Does it need null check (maybe no-op)?

[hachikuji]

I was concerned about this also, but the generated code adds its own null check.

[hachikuji]

@chia7712 Thanks for the reviews. I pushed an update to address your comments. I was a little concerned about the garbage created from the MessageSize objects, so I changed the generated code to use an accumulator pattern instead. Not sure it was really worth it, but let me know what you think.

[hachikuji]

Here are updated results for FetchResponseBenchmark.

Trunk:

Benchmark                                          (partitionCount)  (topicCount)  Mode  Cnt        Score       Error  Units
FetchResponseBenchmark.testSerializeFetchResponse                 3            10  avgt   15     6037.058 ±    41.647  ns/op
FetchResponseBenchmark.testSerializeFetchResponse                 3           500  avgt   15   296913.061 ±  2548.748  ns/op
FetchResponseBenchmark.testSerializeFetchResponse                 3          1000  avgt   15   607466.969 ±  5149.176  ns/op
FetchResponseBenchmark.testSerializeFetchResponse                10            10  avgt   15    17971.239 ±   141.934  ns/op
FetchResponseBenchmark.testSerializeFetchResponse                10           500  avgt   15   888464.188 ±  6341.169  ns/op
FetchResponseBenchmark.testSerializeFetchResponse                10          1000  avgt   15  1938108.325 ± 32056.839  ns/op
FetchResponseBenchmark.testSerializeFetchResponse                20            10  avgt   15    34217.206 ±   234.166  ns/op
FetchResponseBenchmark.testSerializeFetchResponse                20           500  avgt   15  1830593.478 ± 22601.172  ns/op
FetchResponseBenchmark.testSerializeFetchResponse                20          1000  avgt   15  4983809.392 ± 38217.505  ns/op

This patch:

Benchmark                                          (partitionCount)  (topicCount)  Mode  Cnt        Score       Error  Units
FetchResponseBenchmark.testSerializeFetchResponse                 3            10  avgt   15     5686.644 ±    92.892  ns/op
FetchResponseBenchmark.testSerializeFetchResponse                 3           500  avgt   15   294136.644 ±  6709.506  ns/op
FetchResponseBenchmark.testSerializeFetchResponse                 3          1000  avgt   15   599678.709 ±  5837.803  ns/op
FetchResponseBenchmark.testSerializeFetchResponse                10            10  avgt   15    17065.100 ±   241.715  ns/op
FetchResponseBenchmark.testSerializeFetchResponse                10           500  avgt   15   907448.421 ± 14203.990  ns/op
FetchResponseBenchmark.testSerializeFetchResponse                10          1000  avgt   15  1906411.458 ± 26450.041  ns/op
FetchResponseBenchmark.testSerializeFetchResponse                20            10  avgt   15    33147.044 ±  1181.493  ns/op
FetchResponseBenchmark.testSerializeFetchResponse                20           500  avgt   15  1861946.674 ± 25145.868  ns/op
FetchResponseBenchmark.testSerializeFetchResponse                20          1000  avgt   15  4339003.253 ± 69576.240  ns/op

So generally better, but not groundbreaking. That is fine since the main improvement is simplifying the generation of Send objects for all request types.

[chia7712]

@hachikuji This improvement LGTM overall.

It seems to me the serialization mechanism get more complicated since we are trying to add different implementation for IO-heavy requests get better memory usage. For more readable and consistent code, is there a follow-up to apply this improvement to all requests (if we complete all auto-generated protocol migration)?

[chia7712]

If all requests are using auto-generated data, should this be default implementation of AbstractRequest?

[hachikuji]

Yeah, I think so. And looks like we're almost there. After your patch for Produce, the only remaining unconverted API that I see is OffsetsForLeaderEpoch.

[hachikuji]

#7409 might be a good opportunity to complete this. What do you think @ijuma ?

[chia7712]

I'm thinking about how to simplify this process.

Could we reuse the method void write(Writable writable, ObjectSerializationCache cache, short version) ? Maybe we can create a Writable instance but it does not write data to any output. Instead, it calculate the size of buffer according to input data.

[chia7712]

if (tagged) {
  buffer.printf("int _sizeBeforeArray = _size.totalSize();%n");
}

[chia7712]

Is this method equal to https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java#L727 ?

[hachikuji]

If it's ok with you, I'd like to address this in a separate patch. The main difference is the presence of the correlation validation logic in NetworkClient, which has been tailored to a subtle case in SaslClientAuthenticator. I think the envelope parsing logic should also be checking the correlationId, but probably not with the same quirky behavior.

[chia7712]

sure. Open a jira as follow-up :)

[chia7712]

@hachikuji this is a great improvement. +1

[hachikuji]

@chia7712 Thanks for reviews, merging to trunk. I will follow up with the suggestion about AbstractResponse.parseResponse as well as a couple other improvements.