MINOR: refactor FetchResponse#toMessage to avoid creating a lot of un…

[chia7712]

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)

[chia7712]

@ijuma This PR is from #9469 and it brings the better refactor :)

[ijuma]

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.

[chia7712]

@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 FetchableTopicResponses since the t0-p0 and t0-p1 can be merged. By contrast, t0-p0, t1-p1 and t0-p2 ought to generate three FetchableTopicResponses.

It seems to me the reason of generating the intermediate collections before is that reusing the data in Struct/Schema is not convenient.

[ijuma]

On closer inspection, I now remember that batchByTopic already relies on the ordering, so the new implementation is equivalent. LGTM overall. A nit below that should be easy to resolve.

[ijuma]

Maybe we could have a variable called previousTopic to make this code more readable.

[chia7712]

Nice suggestion. will address it later.