AndrewJSchofield commented on code in PR #19148:
URL: https://github.com/apache/kafka/pull/19148#discussion_r1987113040
##########
clients/src/main/java/org/apache/kafka/common/requests/ShareFetchRequest.java:
##########
@@ -226,23 +189,18 @@ public int maxWait() {
return data.maxWaitMs();
}
- public Map<TopicIdPartition, ShareFetchRequest.SharePartitionData>
shareFetchData(Map<Uuid, String> topicNames) {
+ public List<TopicIdPartition> shareFetchData(Map<Uuid, String> topicNames)
{
if (shareFetchData == null) {
synchronized (this) {
if (shareFetchData == null) {
// Assigning the lazy-initialized `shareFetchData` in the
last step
// to avoid other threads accessing a half-initialized
object.
- final LinkedHashMap<TopicIdPartition,
ShareFetchRequest.SharePartitionData> shareFetchDataTmp = new LinkedHashMap<>();
+ final List<TopicIdPartition> shareFetchDataTmp = new
ArrayList<>();
data.topics().forEach(shareFetchTopic -> {
String name =
topicNames.get(shareFetchTopic.topicId());
shareFetchTopic.partitions().forEach(shareFetchPartition -> {
// Topic name may be null here if the topic name
was unable to be resolved using the topicNames map.
- shareFetchDataTmp.put(new
TopicIdPartition(shareFetchTopic.topicId(), new TopicPartition(name,
shareFetchPartition.partitionIndex())),
- new ShareFetchRequest.SharePartitionData(
- shareFetchTopic.topicId(),
-
shareFetchPartition.partitionMaxBytes()
- )
- );
+ shareFetchDataTmp.add(new
TopicIdPartition(shareFetchTopic.topicId(), new TopicPartition(name,
shareFetchPartition.partitionIndex())));
Review Comment:
nit: This could be `new TopicIdPartition(shareFetchTopic.topicId(),
shareFetchPartition.partitionIndex(), name)`.
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -3918,10 +3918,8 @@ class KafkaApisTest extends Logging {
)
when(sharePartitionManager.newContext(any(), any(), any(), any(),
any())).thenReturn(
- new ShareSessionContext(new ShareRequestMetadata(memberId,
shareSessionEpoch), Map(
- new TopicIdPartition(topicId, new TopicPartition(topicName,
partitionIndex)) ->
- new ShareFetchRequest.SharePartitionData(topicId, partitionMaxBytes)
- ).asJava)
+ new ShareSessionContext(new ShareRequestMetadata(memberId,
shareSessionEpoch), List(
+ new TopicIdPartition(topicId, new TopicPartition(topicName,
partitionIndex))).asJava)
Review Comment:
nit: You could use the constructor `TopicIdPartition(Uuid, int, String)`
which would tidy the code up a bit.
##########
server/src/test/java/org/apache/kafka/server/share/fetch/ShareFetchTestUtils.java:
##########
@@ -68,27 +50,22 @@ public static LinkedHashMap<TopicIdPartition, Integer>
orderedMap(int partitionM
* @param rotationAt The position to rotate the keys at.
*/
public static void validateRotatedMapEquals(
Review Comment:
nit: You're rotating a list not a map now, so the method name is inaccurate.
--
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.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
