We initialize the variable in constructor then re-assign while creating another LinkedHashMap in acquirablePartitions() method. Are we are initializing partitionsToComplete here to save null check? Can't we re-use already initialized LinkedHashMap()?

Yes, there have been comments above where I left some variables as null, and it was pointed out that I need to initialize them to avoid null checks at different places, then we just need to do empty checks.. Hence, I've implemented it in this manner.

If the concern is with additional null check then I would recommend general helper methods. My concern is with creating additional maps when they are always re-referenced.

    private <K,V> void addToNullableMap(Map<K, V> map, K key, V value) {
        if (map == null) {
            map = new LinkedHashMap<>();
        }
        map.put(key, value);
    }

    private boolean isMapEmpty(Map<?,?> map) {
        return map == null || map.isEmpty();
    }

Again we reset partitionsAlreadyFetched to response from replicaManagerReadResponse. Why to have such instances created when anyways we have to re-assign?

Same reason as above, we are avoiding any null checks. We just check for empty scenarios by doing this.

nit: I personally find the code hard to read with nested if/else blocks, same is the case here. Though I leave it on you.

if (topicPartitionData.isEmpty()) {
     log.trace("Can't acquire records for any partition in the share fetch request for group {}, member {}, " +
                        "topic partitions {}", shareFetchData.groupId(), shareFetchData.memberId(),
                    shareFetchData.partitionMaxBytes().keySet());
    return false;
}

// In case, fetch offset metadata doesn't exist for one or more topic partitions, we do a
// replicaManager.readFromLog to populate the offset metadata and update the fetch offset metadata for
// those topic partitions.
Map<TopicIdPartition, LogReadResult> replicaManagerReadResponse = updateFetchOffsetMetadata(maybeReadFromLog(topicPartitionData));
if (!anyTopicIdPartitionHasLogReadError(replicaManagerReadResponse) && !isMinBytesSatisfied(topicPartitionData)) {
       log.debug("minBytes is not satisfied for the share fetch request for group {}, member {}, " +
                            "topic partitions {}", shareFetchData.groupId(), shareFetchData.memberId(),
       shareFetchData.partitionMaxBytes().keySet());
       releasePartitionLocks(topicPartitionData.keySet());
       return false;
}

partitionsToComplete = topicPartitionData;
partitionsAlreadyFetched = replicaManagerReadResponse;
....
....

I have been asked to add else blocks in the above comments on this PR, hence I don' think I should change it again. #17539 (comment)

I leave it on @junrao to decide then.

Hmm, Apoorv had a good point in his comment (#17539 (comment)). There seems to be a potential problem. It's possible that thread1 calls tryComplete, finds completed to be false, and is about to set partitionsAlreadyFetched. The expiration thread then calls forceComplete and sets completed to true and proceeds to here. Now, thread1 continues and updates partitionsAlreadyFetched. The expiration thread will pick up the wrong partitionsAlreadyFetched.

Hi @junrao , I've created a ticket https://issues.apache.org/jira/browse/KAFKA-17948 to track this issue and if it fine to you, I would prefer to address the issue in a future PR.

Since the ordering is important, let's use LinkedHashMap.

Since the ordering is important, let's use LinkedHashMap.

Since the ordering is important, let's use LinkedHashMap.

This means tryComplete will never get non-empty fetchOffsetMetadata and its calculation of minBytes will be off. We need to think through how to address this.