Kafka消费者增量拉取

case class FetchSession(val id: Int, // session编号是随机32位数字，防止未授权的客户端伪造数据
                        val privileged: Boolean,
                        val partitionMap: FetchSession.CACHE_MAP,
                        val creationMs: Long,
                        var lastUsedMs: Long,
                        var epoch: Int) // 自增

// FetchSessionHandler.Builder.build()
if (nextMetadata.isFull()) { // epoch=0或者-1
    sessionPartitions = next; // next为之前调动add添加的分区
    next = null; // 本地full拉取，下次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());
        if (response.error() == Errors.FETCH_SESSION_ID_NOT_FOUND) {
            nextMetadata = FetchMetadata.INITIAL;
        } else {
            nextMetadata = nextMetadata.nextCloseExisting();
        }
        return false;
    } else if (nextMetadata.isFull()) {
        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) {
            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. 客户端正常处理全量拉取的响应
            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.
            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. 客户端正常处理增量拉取的响应结果
            log.debug("Node {} sent an incremental fetch response for session {}{}",
                node, response.sessionId(), responseDataToLogString(response));
            nextMetadata = nextMetadata.nextIncremental();
            return true;
        }
    }
}

// KafkaApis.handleFetchRequest
    val fetchContext = fetchManager.newContext(
      fetchRequest.metadata,
      fetchRequest.fetchData,
      fetchRequest.toForget,
      fetchRequest.isFromFollower)

    // 针对不同的拉取上下文，分别更新并生成响应数据
    unconvertedFetchResponse = fetchContext.updateAndGenerateResponseData(partitions)

override def updateAndGenerateResponseData(updates: FetchSession.RESP_MAP): FetchResponse[Records] = {
  def createNewSession: FetchSession.CACHE_MAP = {
    val cachedPartitions = new FetchSession.CACHE_MAP(updates.size)
    updates.entrySet.asScala.foreach(entry => {
      val part = entry.getKey
      val respData = entry.getValue
      val reqData = fetchData.get(part)
      cachedPartitions.mustAdd(new CachedPartition(part, reqData, respData))
    })
    cachedPartitions
  }
  val responseSessionId = cache.maybeCreateSession(time.milliseconds(), isFromFollower,
      updates.size, () => createNewSession)
  debug(s"Full fetch context with session id $responseSessionId returning " +
    s"${partitionsToLogString(updates.keySet)}")
  new FetchResponse(Errors.NONE, updates, 0, responseSessionId)
}

def maybeCreateSession(now: Long,
                       privileged: Boolean,
                       size: Int,
                       createPartitions: () => FetchSession.CACHE_MAP): Int =
synchronized {
  // If there is room, create a new session entry.
  if ((sessions.size < maxEntries) ||
      tryEvict(privileged, EvictableKey(privileged, size, 0), now)) {
    val partitionMap = createPartitions()
    // 这里创建一个新的session时，同时也会增加epoch，从0到1
    val session = new FetchSession(newSessionId(), privileged, partitionMap,
        now, now, JFetchMetadata.nextEpoch(INITIAL_EPOCH))
    debug(s"Created fetch session ${session.toString}")
    sessions.put(session.id, session)
    touch(session, now)
    session.id
  } else {
    debug(s"No fetch session created for privileged=$privileged, size=$size.")
    INVALID_SESSION_ID
  }
}