class FetchSession(val id: Int, // sessionid是随机32位数字，用于鉴权，防止客户端伪造 
                   val privileged: Boolean, // 是否授权 
                   val partitionMap: FetchSession.CACHE_MAP,// 缓存数据CachedPartitionMap 
                   val creationMs: Long, // 创建Session的时间 
                   var lastUsedMs: Long, // 上次使用会话的时间，由FetchSessionCache更新 
                   var epoch: Int) // 获取会话序列号

if (nextMetadata.isFull()) { // epoch为0或者-1 
    if (log.isDebugEnabled()) { 
        log.debug("Built full fetch {} for node {} with {}.", 
                  nextMetadata, node, partitionsToLogString(next.keySet())); 
    } 
    sessionPartitions = next; // next为之前动态增加的分区 
    next = null; // 本地全量拉取，下次next为null 
    Map<TopicPartition, PartitionData> toSend = 
        Collections.unmodifiableMap(new LinkedHashMap<>(sessionPartitions)); 
    return new FetchRequestData(toSend, Collections.emptyList(), toSend, nextMetadata); 
}

public boolean handleResponse(FetchResponse<?> response) { 
if (response.error() != Errors.NONE) { 
log.info("Node {} was unable to process the fetch request with {}: {}.", 
node, nextMetadata, response.error()); // 当集群session超过最大阀值，会出现这个异常信息 
if (response.error() == Errors.FETCH_SESSION_ID_NOT_FOUND) { 
nextMetadata = FetchMetadata.INITIAL; 
} else { 
nextMetadata = nextMetadata.nextCloseExisting(); 
} 
return false; 
} 
if (nextMetadata.isFull()) { 
if (response.responseData().isEmpty() && response.throttleTimeMs() > 0) { 
// Normally, an empty full fetch response would be invalid.  However, KIP-219 
// specifies that if the broker wants to throttle the client, it will respond 
// to a full fetch request with an empty response and a throttleTimeMs 
// value set.  We don't want to log this with a warning, since it's not an error. 
// However, the empty full fetch response can't be processed, so it's still appropriate 
// to return false here. 
if (log.isDebugEnabled()) { 
log.debug("Node {} sent a empty full fetch response to indicate that this " + 
"client should be throttled for {} ms.", node, response.throttleTimeMs()); 
} 
nextMetadata = FetchMetadata.INITIAL; 
return false; 
} 
String problem = verifyFullFetchResponsePartitions(response); 
if (problem != null) { 
log.info("Node {} sent an invalid full fetch response with {}", node, problem); 
nextMetadata = FetchMetadata.INITIAL; 
return false; 
} else if (response.sessionId() == INVALID_SESSION_ID) { 
if (log.isDebugEnabled()) 
log.debug("Node {} sent a full fetch response{}", node, responseDataToLogString(response)); 
nextMetadata = FetchMetadata.INITIAL; 
return true; 
} else { 
// The server created a new incremental fetch session. 
if (log.isDebugEnabled()) 
log.debug("Node {} sent a full fetch response that created a new incremental " + 
"fetch session {}{}", node, response.sessionId(), responseDataToLogString(response)); 
nextMetadata = FetchMetadata.newIncremental(response.sessionId()); 
return true; 
} 
} else { 
String problem = verifyIncrementalFetchResponsePartitions(response); 
if (problem != null) { 
log.info("Node {} sent an invalid incremental fetch response with {}", node, problem); 
nextMetadata = nextMetadata.nextCloseExisting(); 
return false; 
} else if (response.sessionId() == INVALID_SESSION_ID) { 
// The incremental fetch session was closed by the server. 
if (log.isDebugEnabled()) 
log.debug("Node {} sent an incremental fetch response closing session {}{}", 
node, nextMetadata.sessionId(), responseDataToLogString(response)); 
nextMetadata = FetchMetadata.INITIAL; 
return true; 
} else { 
// The incremental fetch session was continued by the server. 
// We don't have to do anything special here to support KIP-219, since an empty incremental 
// fetch request is perfectly valid. 
if (log.isDebugEnabled()) 
log.debug("Node {} sent an incremental fetch response with throttleTimeMs = {} " + 
"for session {}{}", node, response.throttleTimeMs(), response.sessionId(), 
responseDataToLogString(response)); 
nextMetadata = nextMetadata.nextIncremental(); 
return true; 
} 
} 
}

val fetchContext = fetchManager.newContext( 
fetchRequest.metadata, 
fetchRequest.fetchData, 
fetchRequest.toForget, 
fetchRequest.isFromFollower) 
// ...... 
if (fetchRequest.isFromFollower) { 
// We've already evaluated against the quota and are good to go. Just need to record it now. 
unconvertedFetchResponse = fetchContext.updateAndGenerateResponseData(partitions) 
val responseSize = KafkaApis.sizeOfThrottledPartitions(versionId, unconvertedFetchResponse, quotas.leader) 
quotas.leader.record(responseSize) 
trace(s"Sending Fetch response with partitions.size=${unconvertedFetchResponse.responseData.size}, " + 
s"metadata=${unconvertedFetchResponse.sessionId}") 
requestHelper.sendResponseExemptThrottle(request, createResponse(0), Some(updateConversionStats))
}