MINOR: Initialize fetchPartitionStatus as a Map type to reduce collection conversions

[DL1231]

see #19876 (comment)
Initialize fetchPartitionStatus as a Map type to reduce unnecessary
collection conversions.

Reviewers: Ismael Juma ismael@juma.me.uk, Kamal Chandraprakash
kamal.chandraprakash@gmail.com, Chia-Ping Tsai chia7712@gmail.com

[chia7712]

Is it possible to have duplicate TPs in this "fetch" path?

[ijuma]

Also, we are doing two conversions here - you can avoid one by using a view or iterator after asScala and using toBuffer instead of toSeq (the latter can result in a lazy collection being created which can result in problems).

[DL1231]

The source of fetchPartitionStatus comes from fetchData().

kafka/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java

Lines 395 to 420 in 53e1172

	public Map<TopicIdPartition, PartitionData> fetchData(Map<Uuid, String> topicNames) {
	final LinkedHashMap<TopicIdPartition, PartitionData> fetchData = new LinkedHashMap<>();
	final short version = version();
	data.topics().forEach(fetchTopic -> {
	String name;
	if (version < 13) {
	name = fetchTopic.topic(); // can't be null
	} else {
	name = topicNames.get(fetchTopic.topicId());
	}
	fetchTopic.partitions().forEach(fetchPartition ->
	// Topic name may be null here if the topic name was unable to be resolved using the topicNames map.
	fetchData.put(new TopicIdPartition(fetchTopic.topicId(), new TopicPartition(name, fetchPartition.partition())),
	new PartitionData(
	fetchTopic.topicId(),
	fetchPartition.fetchOffset(),
	fetchPartition.logStartOffset(),
	fetchPartition.partitionMaxBytes(),
	optionalEpoch(fetchPartition.currentLeaderEpoch()),
	optionalEpoch(fetchPartition.lastFetchedEpoch())
	)
	)
	);
	});
	return fetchData;
	}

Its type is LinkedHashMap<TopicIdPartition, PartitionData>, so there will be no duplicate TPs.

[ijuma]

Why do we convert to a Scala List if we convert to a Java collection right after? We should probably use Java's Stream here and avoid the Scala collections altogether.

[DL1231]

Thanks for catching this! I've fixed it. PTAL when you get a chance.

[kamalcph]

LGTM

[chia7712]

@DL1231 please rebase code to include latest fixes

[ijuma]

I think you can use the Stream.toList() directly (no need for collect). This method was added to Stream in Java 16.

[chia7712]

@DL1231 thanks for updates

[chia7712]

Perhaps readFromLog could accept a util.Iterator instead of a Seq. This would help avoid creating an extra collection in this hot operation.

[ijuma]

Iterators can be exhausted and hence are a bit more brittle. I would only use them as a method parameter if there is a meaningful difference in performance.

[chia7712]

fair enough

[chia7712]

LGTM

I plan to merge this later, assuming there are not future comments