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.

[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.