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))
}