KAFKA-12410 KafkaAPis ought to group fetch data before generating fet…#10269
KAFKA-12410 KafkaAPis ought to group fetch data before generating fet…#10269chia7712 wants to merge 10 commits intoapache:trunkfrom
Conversation
| // Iterator that goes over the given partition map and selects partitions that need to be included in the response. | ||
| // If updateFetchContextAndRemoveUnselected is set to true, the fetch context will be updated for the selected | ||
| // partitions and also remove unselected ones as they are encountered. | ||
| private class PartitionIterator(val iter: FetchSession.RESP_MAP_ITER, |
There was a problem hiding this comment.
The iterator is unnecessary since we have to generate list collection in order to calculate message size.
|
|
||
| def partitionsToLogString(partitions: util.Collection[TopicPartition]): String = | ||
| FetchSession.partitionsToLogString(partitions, isTraceEnabled) | ||
| def partitionsToLogString(topics: FetchSession.RESP_MAP): String = { |
There was a problem hiding this comment.
this method is used to log (DEBUG level) so it should be fine to iterate through whole collection.
| // the callback for process a fetch response, invoked before throttling | ||
| def processResponseCallback(responsePartitionData: Seq[(TopicPartition, FetchPartitionData)]): Unit = { | ||
| val partitions = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData] | ||
| val topicResponses = new util.ArrayList[FetchResponseData.FetchableTopicResponse]() |
There was a problem hiding this comment.
This is the main purpose of this PR. KafkaApis keeps grouped data.
|
Nice PR! I will take a look at it on Monday. |
dajac
left a comment
dajac
left a comment
There was a problem hiding this comment.
I've left a few comments. I couldn't ready all the PR yet.
|
Thanks for the PR. Can you check the perf impact of these changes? |
sure. will add benchmark results tomorrow. |
|
@ijuma The results of performance tests are attached. It does not show obvious performance regression. Will run more tests tomorrow. |
|
@chia7712 I was thinking about the jmh microbenchmarks that stress fetch, fetch session and so on. |
will copy that. |
|
@ijuma the JMH result of fetch session is attached. I tried to have a JMH for stress fetch. However, |
|
@chia7712 is there any work in progress for a KafkaApis.handleFetchRequest test? I suspect it would be similar but maybe a bit harder than what I did for the LeaderAndIsr version #10071 (trading replicamanager for fetchmanager, etc). This benchmark would be helpful for #9944 as you could probably guess :) |
this PR is blocked by #9944. This PR (and other related issues) aim to remove all extra collection creation by using auto-generated data. In #9944 we have to create a lot of collections to handle the topic id in fetch request. Hence, I need to rethink the value (and approach) of this PR :) |
|
@chia7712 I'm rewriting #9944 to use the autogenerated structures based on this PR. Just pushed a version that simplifies the unresolved topic ID handling. I tried to make it easier to build the fetch response using the data object. Going to try to build the response using the data object in most places today and I can push that version as soon as I can. |
sounds good. If the new approach is very different from #9944, please open a new PR in order to compare them :) |
@chia7712 I've updated the code. I think this is the direction we want so I didn't open a PR. The idea is that FetchSession can now generate a list of the unresolvedTopics' But let me know if it's hard to read. I can open a new one and revert the changes on the old. |
4 participants
The fetch data generated by
KafkaApisis re-grouped when it is converted toFetchResponse. That is unnecessary sinceKafkaApiscan keep a grouped collection for fetch data. The other main changes are shown below.FetchResponse#ofFetchResponsePartitionIteratorJMH Tests
IncrementalFetchContextBenchmarktrunk (#10291)
patch (603da4d)
Performance Tests
benchmark_test.py::Benchmark.test_consumer_throughputcase 0: +2.760595128 %
{ "compression_type": "none", "security_protocol": "PLAINTEXT", "interbroker_security_protocol": "PLAINTEXT" }case 1: -0.6257537776 %
{ "compression_type": "snappy", "security_protocol": "PLAINTEXT", "interbroker_security_protocol": "PLAINTEXT" }Committer Checklist (excluded from commit message)