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 2021/01/03 08:09:55 UTC
[GitHub] [kafka] chia7712 opened a new pull request #9818: MINOR: refactor FetchResponse#toMessage to avoid creating a lot of un…
chia7712 opened a new pull request #9818:
URL: https://github.com/apache/kafka/pull/9818
Instead of creating intermediate collections to batch data, we directly check the last topic in auto-generated data to collect partitions having same topic.
### 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 pull request #9818: MINOR: refactor FetchResponse#toMessage to avoid creating a lot of un…
Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9818:
URL: https://github.com/apache/kafka/pull/9818#issuecomment-753653113
Thanks for the PR. Have you checked that we get the same batching this way? The behavior won't be the same unless the iterator order is correct.
----------------------------------------------------------------
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 #9818: MINOR: refactor FetchResponse#toMessage to avoid creating a lot of un…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9818:
URL: https://github.com/apache/kafka/pull/9818#issuecomment-753729083
@ijuma Thanks for your quick response!
> Have you checked that we get the same batching this way?
There is a test to verify the batch order (https://github.com/apache/kafka/blob/trunk/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala#L83).
> The behavior won't be the same unless the iterator order is correct.
There are two rules of batching partition data.
1. the order must be kept. the tp order from input (iterator) must be equal to tp order in ```FetchResponseData```.
2. batch the partitions only if we can keep the their order in ```FetchResponseData```. For example: we should group t0-p0, t0-p1 and t1-p2 to two ```FetchableTopicResponse```s since the t0-p0 and t0-p1 can be merged. By contrast, t0-p0, t1-p1 and t0-p2 ought to generate three ```FetchableTopicResponse```s.
It seems to me the reason of generating the intermediate collections before is that reusing the data in ```Struct/Schema``` is not convenient.
----------------------------------------------------------------
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 merged pull request #9818: MINOR: refactor FetchResponse#toMessage to avoid creating a lot of un…
Posted by GitBox <gi...@apache.org>.
chia7712 merged pull request #9818:
URL: https://github.com/apache/kafka/pull/9818
----------------------------------------------------------------
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 #9818: MINOR: refactor FetchResponse#toMessage to avoid creating a lot of un…
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9818:
URL: https://github.com/apache/kafka/pull/9818#discussion_r551407113
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -331,28 +331,32 @@ public int sessionId() {
private static <T extends BaseRecords> FetchResponseData toMessage(int throttleTimeMs, Errors error,
Iterator<Map.Entry<TopicPartition, PartitionData<T>>> partIterator,
int sessionId) {
- FetchResponseData message = new FetchResponseData();
- message.setThrottleTimeMs(throttleTimeMs);
- message.setErrorCode(error.code());
- message.setSessionId(sessionId);
-
List<FetchResponseData.FetchableTopicResponse> topicResponseList = new ArrayList<>();
- List<FetchRequest.TopicAndPartitionData<PartitionData<T>>> topicsData =
- FetchRequest.TopicAndPartitionData.batchByTopic(partIterator);
- topicsData.forEach(partitionDataTopicAndPartitionData -> {
- List<FetchResponseData.FetchablePartitionResponse> partitionResponses = new ArrayList<>();
- partitionDataTopicAndPartitionData.partitions.forEach((partitionId, partitionData) -> {
- // Since PartitionData alone doesn't know the partition ID, we set it here
- partitionData.partitionResponse.setPartition(partitionId);
+ partIterator.forEachRemaining(entry -> {
+ PartitionData<T> partitionData = entry.getValue();
+ // Since PartitionData alone doesn't know the partition ID, we set it here
+ partitionData.partitionResponse.setPartition(entry.getKey().partition());
+ // We have to keep the order of input topic-partition. Hence, we batch the partitions only if the last
+ // batch is in the same topic group.
+ if (!topicResponseList.isEmpty() && topicResponseList.get(topicResponseList.size() - 1)
+ .topic().equals(entry.getKey().topic())) {
+ topicResponseList.get(topicResponseList.size() - 1)
Review comment:
Maybe we could have a variable called `previousTopic` to make this code more readable.
----------------------------------------------------------------
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 #9818: MINOR: refactor FetchResponse#toMessage to avoid creating a lot of un…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9818:
URL: https://github.com/apache/kafka/pull/9818#discussion_r551410221
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
##########
@@ -331,28 +331,32 @@ public int sessionId() {
private static <T extends BaseRecords> FetchResponseData toMessage(int throttleTimeMs, Errors error,
Iterator<Map.Entry<TopicPartition, PartitionData<T>>> partIterator,
int sessionId) {
- FetchResponseData message = new FetchResponseData();
- message.setThrottleTimeMs(throttleTimeMs);
- message.setErrorCode(error.code());
- message.setSessionId(sessionId);
-
List<FetchResponseData.FetchableTopicResponse> topicResponseList = new ArrayList<>();
- List<FetchRequest.TopicAndPartitionData<PartitionData<T>>> topicsData =
- FetchRequest.TopicAndPartitionData.batchByTopic(partIterator);
- topicsData.forEach(partitionDataTopicAndPartitionData -> {
- List<FetchResponseData.FetchablePartitionResponse> partitionResponses = new ArrayList<>();
- partitionDataTopicAndPartitionData.partitions.forEach((partitionId, partitionData) -> {
- // Since PartitionData alone doesn't know the partition ID, we set it here
- partitionData.partitionResponse.setPartition(partitionId);
+ partIterator.forEachRemaining(entry -> {
+ PartitionData<T> partitionData = entry.getValue();
+ // Since PartitionData alone doesn't know the partition ID, we set it here
+ partitionData.partitionResponse.setPartition(entry.getKey().partition());
+ // We have to keep the order of input topic-partition. Hence, we batch the partitions only if the last
+ // batch is in the same topic group.
+ if (!topicResponseList.isEmpty() && topicResponseList.get(topicResponseList.size() - 1)
+ .topic().equals(entry.getKey().topic())) {
+ topicResponseList.get(topicResponseList.size() - 1)
Review comment:
Nice suggestion. will address it later.
----------------------------------------------------------------
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 #9818: MINOR: refactor FetchResponse#toMessage to avoid creating a lot of un…
Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9818:
URL: https://github.com/apache/kafka/pull/9818#issuecomment-753583153
@ijuma This PR is from #9469 and it brings the better refactor :)
----------------------------------------------------------------
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