[FEATURE] Support kafka LogValidator validate inner records and compression codec when handle producer request with entryFormat=kafka

[wenbingshen]

fixes #687

Motivation

Previously, in order to support low version kafka clients, such as 0.10 version clients, we verified the data when processing fetch requests. Because the lower message format would maintain the internal message set, according to the production situation I encountered, in a certain sometimes there is an error in the offset of the internal message set, such as the bug in the low version of the sarama client.

The problem is that for the low version message format, no matter whether there is a problem with the internal message set, we will regenerate the message records, which is unnecessary in some cases.

Modifications

Support message set verification in production like kafka, and only need to verify when entryFormat=kafka.
Because when entryFormat=pulsar, we still cannot avoid doing message conversion during consumption.

[wenbingshen]

If we can use this pr. Then we can use direct memory to rebuild kafka records like #673.
I can work for this after this pr has been merged.

[wenbingshen]

@BewareMyPower PTAL if you have free time. Thanks. :)

[BewareMyPower]

Please rebase to master since it has conflicts with your previous PR.

[wenbingshen]

Please rebase to master since it has conflicts with your previous PR.

Thanks for the reminder, I happen to be doing this job and the conflict has been resolved. :)

[wenbingshen]

The result of my local performance test.

[wenbingshen]

The failed test does not seem to be related to this pr.

[BewareMyPower]

Here's my test result for encoding before and after this PR. Before (Kafka) represents KafkaEntryFormatter before this PR.

	Before (Kafka)	Before (Pulsar)	After (Kafka)	After (Pulsar)
Fixed records for 100 times	12	1246	48	1267
Random records for 100 times	10	854	23	961
Fixed records for 1000 times	78	7630	257	7230
Random records for 1000 times	88	7376	170	6682

We can see the encoding time of KafkaEntryFormatter has increased 2 to 3 times after this PR because the time of PulsarEntryFormatter didn't change significantly.

I think we cannot accept this performance lost just for compatibility with some special cases for Kafka 0.10 client.

If you insisted on this change, please add a new EntryFormatter implementation.

[wenbingshen]

Here's my test result for encoding before and after this PR. Before (Kafka) represents KafkaEntryFormatter before this PR.

Before (Kafka) Before (Pulsar) After (Kafka) After (Pulsar)
Fixed records for 100 times 12 1246 48 1267
Random records for 100 times 10 854 23 961
Fixed records for 1000 times 78 7630 257 7230
Random records for 1000 times 88 7376 170 6682
We can see the encoding time of KafkaEntryFormatter has increased 2 to 3 times after this PR because the time of PulsarEntryFormatter didn't change significantly.

I think we cannot accept this performance lost just for compatibility with some special cases for Kafka 0.10 client.

If you insisted on this change, please add a new EntryFormatter implementation.

Okay, if we add a new entryFormatter, does it mean that the previous KafkaEntryFormatter no longer needs to support 0.10 clients and only supports magic=2?

[wenbingshen]

@BewareMyPower
I think the current failed test does not seem to be related to this pr, because I also saw the same error on October 10, 2021.

https://github.com/wenbingshen/Fail-tests/blob/master/io.streamnative.pulsar.handlers.kop.KafkaRequestTypeTest.txt

[wenbingshen]

This is a known flaky test #809
https://github.com/streamnative/kop/pull/791/checks?check_run_id=3902214148

[BewareMyPower]

@BewareMyPower And can you help add docs? Thanks very much.

There's a doc-required label for this PR. I'll add related docs after this PR is merged.

[wenbingshen]

@BewareMyPower And can you help add docs? Thanks very much.

There's a doc-required label for this PR. I'll add related docs after this PR is merged.

Good. Thanks. :)

[BewareMyPower]

Left some comments. In addition, I cannot see any different tests between KafkaV1EntryFormatter and KafkaMixedEntryFormatter. I think the newly added tests for mixed_kafka are not necessary, except tests under io.streamnative.pulsar.handlers.kop.compatibility.

We already have Kafka 0.10 client for tests. Does it mean only some older version Golang sarama client can make a difference between kafka and mixed_kafka?

Or could you do some simple tests just for EntryFormatter#encode method? Like creating a V0 message set, and show the different behaviors between kafka and mixed_kafka?

[BewareMyPower]

If the newer sarama client has fixed the problem. Could you give the version?

[wenbingshen]

If the newer sarama client has fixed the problem. Could you give the version?

For reproducing the problem I described, please take a look at this pr: sarama-1002 sarama go low version client jump After allocating the internal message offset for compressed data, this will be redistributed by the Kafka server in Kafka, otherwise the message with the wrong attribute will be a corrupted message.

here sarama-1002 v1.16 @BewareMyPower

[wenbingshen]

Left some comments. In addition, I cannot see any different tests between KafkaV1EntryFormatter and KafkaMixedEntryFormatter. I think the newly added tests for mixed_kafka are not necessary, except tests under io.streamnative.pulsar.handlers.kop.compatibility.

We already have Kafka 0.10 client for tests. Does it mean only some older version Golang sarama client can make a difference between kafka and mixed_kafka?

Or could you do some simple tests just for EntryFormatter#encode method? Like creating a V0 message set, and show the different behaviors between kafka and mixed_kafka?

All comments done. PTAL io.streamnative.pulsar.handlers.kop.format.EntryFormatterTest for the test to show the different behaviors between kafka and mixed_kafka.

[BewareMyPower]

I will review it soon.

[BewareMyPower]

Overall LGTM. Please check the comments.