apoorvmittal10 commented on code in PR #17583:
URL: https://github.com/apache/kafka/pull/17583#discussion_r1817026225
##########
core/src/main/java/kafka/server/share/DelayedShareFetch.java:
##########
@@ -79,16 +79,16 @@ public void onExpiration() {
*/
@Override
public void onComplete() {
- log.trace("Completing the delayed share fetch request for group {},
member {}, " +
- "topic partitions {}", shareFetchData.groupId(),
- shareFetchData.memberId(),
shareFetchData.partitionMaxBytes().keySet());
+ log.trace("Completing the delayed share fetch request for group {},
member {}, "
+ + "topic partitions {}", shareFetchData.groupId(),
shareFetchData.memberId(),
+ topicPartitionDataFromTryComplete != null ?
topicPartitionDataFromTryComplete.keySet() : "null");
if (shareFetchData.future().isDone())
return;
Map<TopicIdPartition, FetchRequest.PartitionData> topicPartitionData;
// tryComplete did not invoke forceComplete, so we need to check if we
have any partitions to fetch.
- if (topicPartitionDataFromTryComplete.isEmpty())
+ if (topicPartitionDataFromTryComplete == null ||
topicPartitionDataFromTryComplete.isEmpty())
Review Comment:
I have added tests.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
