Kafka消费者加入group流程（上）

/**
 * 消费者加入 Group，它确保了这个 group 的 coordinator 是已知的，并且这个consumer是已经加入group，也用于offset周期性的提交。
 * 如果超时将会立即返回。
 */
public boolean poll(Timer timer) {
	// 可能会更新订阅的元数据信息
	maybeUpdateSubscriptionMetadata();
	// 用于测试
	invokeCompletedOffsetCommitCallbacks();

	if (subscriptions.partitionsAutoAssigned()) {
		/** 检查心跳线程运行是否正常, 如果心跳线程运行失败, 则抛出异常; 反之更新poll 调用的时间 */
		pollHeartbeat(timer.currentTimeMs());
		// coordinator 未知,初始化 Consumer Coordinator
		if (coordinatorUnknown() && !ensureCoordinatorReady(timer)) {
			return false;
		}
		// 判断是否需要重新加入group,如果订阅的 partition 变化或则分配的 partition 变化时,需要rejoin
		if (rejoinNeededOrPending()) {
			// 因为初始化的metadata刷新和初始化Rebalance存在竞态条件，在这里要确保metadata刷新在前面。这样可以保证在入组之前，订阅主题和Broker主题至少有一次匹配的过程。
			if (subscriptions.hasPatternSubscription()) {
				// 对于模式匹配订阅的Consumer，当一个Topic创建后，任何Consumer通过刷新metadata后发现新Topic后，都会触发一次Rebalance。因此此时可能会有大量的Rebalance操作，通过下面的backoff time判断会显著降低Rebalance的频率。
				if (this.metadata.timeToAllowUpdate(timer.currentTimeMs()) == 0) {
					this.metadata.requestUpdate();
				}

				if (!client.ensureFreshMetadata(timer)) {
					return false;
				}

				maybeUpdateSubscriptionMetadata();
			}
			// 确保group是active;加入group;分配到订阅的partition
			if (!ensureActiveGroup(timer)) {
				return false;
			}
		}
	} else {
		// For manually assigned partitions, if there are no ready nodes, await metadata.
		// If connections to all nodes fail, wakeups triggered while attempting to send fetch
		// requests result in polls returning immediately, causing a tight loop of polls. Without
		// the wakeup, poll() with no channels would block for the timeout, delaying re-connection.
		// awaitMetadataUpdate() initiates new connections with configured backoff and avoids the busy loop.
		// When group management is used, metadata wait is already performed for this scenario as
		// coordinator is unknown, hence this check is not required.
		if (metadata.updateRequested() && !client.hasReadyNodes(timer.currentTimeMs())) {
			client.awaitMetadataUpdate(timer);
		}
	}
	// 设置自动commit时，当定时达到时，执行自动commit
	maybeAutoCommitOffsetsAsync(timer.currentTimeMs());
	return true;
}

/**
 * 确保 coordinator ready 去接收请求 （已经连接，并可以发送请求）
 */
protected synchronized boolean ensureCoordinatorReady(final Timer timer) {
    if (!coordinatorUnknown())
        return true;

    do {
        // 获取 GroupCoordinator,并建立连接
        final RequestFuture<Void> future = lookupCoordinator();
        client.poll(future, timer);

        if (!future.isDone()) {
            // ran out of time
            break;
        }
        // 如果获取的过程中失败了
        if (future.failed()) {
            if (future.isRetriable()) {
                log.debug("Coordinator discovery failed, refreshing metadata");
                client.awaitMetadataUpdate(timer);
            } else
                throw future.exception();
        } else if (coordinator != null && client.isUnavailable(coordinator)) {
            // 当找到了 coordinator，但是连接失败了，此时标记为 dead，然后重试
            markCoordinatorUnknown();
            timer.sleep(rebalanceConfig.retryBackoffMs);
        }
    } while (coordinatorUnknown() && timer.notExpired());

    return !coordinatorUnknown();
}

/**
 * 选择最空闲的节点, 发送 groupCoordinator请求
 */
protected synchronized RequestFuture<Void> lookupCoordinator() {
	if (findCoordinatorFuture == null) {
		// 选择一个连接数最少的节点（假装最空闲）
		Node node = this.client.leastLoadedNode();
		if (node == null) {
			log.debug("No broker available to send FindCoordinator request");
			return RequestFuture.noBrokersAvailable();
		} else
			// 发送请求并处理响应
			findCoordinatorFuture = sendFindCoordinatorRequest(node);
	}
	return findCoordinatorFuture;
}

/**
 * 发送 CoordinatorRequest 请求
 * 为 group 寻找 coordinator，发送 GroupMetadata 请求到Broker。
 */
private RequestFuture<Void> sendFindCoordinatorRequest(Node node) {
	// 初始化 GroupMetadata 请求
	log.debug("Sending FindCoordinator request to broker {}", node);
	FindCoordinatorRequest.Builder requestBuilder = new FindCoordinatorRequest.Builder(new FindCoordinatorRequestData().setKeyType(CoordinatorType.GROUP.id()).setKey(this.rebalanceConfig.groupId));
	// 请求发送之后的响应结果，委派给了 Handler 执行
	return client.send(node, requestBuilder).compose(new FindCoordinatorResponseHandler());
}

// 对 GroupCoordinator 的 response 进行处理,回调
private class FindCoordinatorResponseHandler extends RequestFutureAdapter<ClientResponse, Void> {
	@Override
	public void onSuccess(ClientResponse resp, RequestFuture<Void> future) {
		log.debug("Received FindCoordinator response {}", resp);
		clearFindCoordinatorFuture();

		FindCoordinatorResponse findCoordinatorResponse = (FindCoordinatorResponse) resp.responseBody();
		Errors error = findCoordinatorResponse.error();
		if (error == Errors.NONE) {
			synchronized (AbstractCoordinator.this) {
				int coordinatorConnectionId = Integer.MAX_VALUE - findCoordinatorResponse.data().nodeId();
				/** 如果正确获取 GroupCoordinator 时, 建立连接,并更新心跳时间 */
				AbstractCoordinator.this.coordinator = new Node(coordinatorConnectionId, findCoordinatorResponse.data().host(), findCoordinatorResponse.data().port());
				log.info("Discovered group coordinator {}", coordinator);
				// 初始化 tcp 连接
				client.tryConnect(coordinator);
				// 更新心跳时间
				heartbeat.resetSessionTimeout();
			}
			future.complete(null);
		} else if (error == Errors.GROUP_AUTHORIZATION_FAILED) {
			future.raise(new GroupAuthorizationException(rebalanceConfig.groupId));
		} else {
			log.debug("Group coordinator lookup failed: {}", findCoordinatorResponse.data().errorMessage());
			future.raise(error);
		}
	}

	@Override
	public void onFailure(RuntimeException e, RequestFuture<Void> future) {
		clearFindCoordinatorFuture();
		super.onFailure(e, future);
	}
}