Heaven.Blog

assignment.put(memberId, new ArrayList<TopicPartition>());

for (Map.Entry<String, List<String>> topicEntry : consumersPerTopic.entrySet()) {

String topic = topicEntry.getKey();

List<String> consumersForTopic = topicEntry.getValue();

Integer numPartitionsForTopic = partitionsPerTopic.get(topic);

if (numPartitionsForTopic == null)

continue;

Collections.sort(consumersForTopic);

int numPartitionsPerConsumer = numPartitionsForTopic / consumersForTopic.size();  //topic分区数除以消费者总数

int consumersWithExtraPartition = numPartitionsForTopic % consumersForTopic.size();  //计算额外分区

List<TopicPartition> partitions = AbstractPartitionAssignor.partitions(topic, numPartitionsForTopic);

for (int i = 0, n = consumersForTopic.size(); i < n; i++) {

int start = numPartitionsPerConsumer * i + Math.min(i, consumersWithExtraPartition);

int length = numPartitionsPerConsumer + (i + 1 > consumersWithExtraPartition ? 0 : 1);

assignment.get(consumersForTopic.get(i)).addAll(partitions.subList(start, start + length));

}

}

}

C2-0：T1（4，5，6） T2（4，5，6）

C2-1：T1（7，8，9） T2（7，8，9）

public void poll(long now) {

invokeCompletedOffsetCommitCallbacks();// note: 用于测试

// note: Step1 通过 subscribe() 方法订阅 topic,并且 coordinator 未知,初始化 Consumer Coordinator

if (subscriptions.partitionsAutoAssigned() && coordinatorUnknown()) {

// note: 获取 GroupCoordinator 地址,并且建立连接

ensureCoordinatorReady();

now = time.milliseconds();

}

// note: Step2 判断是否需要重新加入 group,如果订阅的 partition 变化或则分配的 partition 变化时,需要 rejoin

if (needRejoin()) {

// due to a race condition between the initial metadata fetch and the initial rebalance,

// we need to ensure that the metadata is fresh before joining initially. This ensures

// that we have matched the pattern against the cluster’s topics at least once before joining.

// note: rejoin group 之前先刷新一下 metadata（对于 AUTO_PATTERN 而言）

if (subscriptions.hasPatternSubscription())

client.ensureFreshMetadata();

// note: 确保 group 是 active; 加入 group; 分配订阅的 partition

ensureActiveGroup();

now = time.milliseconds();

}

// note: Step3 检查心跳线程运行是否正常,如果心跳线程失败,则抛出异常,反之更新 poll 调用的时间

pollHeartbeat(now);

// note: Step4 自动 commit 时,当定时达到时,进行自动 commit

maybeAutoCommitOffsetsAsync(now);

}

long remainingMs = timeoutMs;

while (coordinatorUnknown()) {

// note:  获取 GroupCoordinator,并建立连接

RequestFuture<Void> future = lookupCoordinator();

client.poll(future, remainingMs);

if (future.failed()) {// note: 如果获取的过程中失败了

if (future.isRetriable()) {

remainingMs = timeoutMs – (time.milliseconds() – startTimeMs);

if (remainingMs <= 0)

break;

log.debug(“Coordinator discovery failed for group {}, refreshing metadata”, groupId);

client.awaitMetadataUpdate(remainingMs);

} else

throw future.exception();

} else if (coordinator != null && client.connectionFailed(coordinator)) {

// we found the coordinator, but the connection has failed, so mark

// it dead and backoff before retrying discovery

coordinatorDead();

time.sleep(retryBackoffMs);

}

remainingMs = timeoutMs – (time.milliseconds() – startTimeMs);

if (remainingMs <= 0)

break;

}

return !coordinatorUnknown();

}

protected synchronized RequestFuture<Void> lookupCoordinator() {

if (findCoordinatorFuture == null) {

// find a node to ask about the coordinator

Node node = this.client.leastLoadedNode();//NOTE: 找一个节点,发送 groupCoordinator 的请求

if (node == null) {

// TODO: If there are no brokers left, perhaps we should use the bootstrap set

// from configuration?

log.debug(“No broker available to send GroupCoordinator request for group {}”, groupId);

return RequestFuture.noBrokersAvailable();

} else

findCoordinatorFuture = sendGroupCoordinatorRequest(node);//NOTE: 发送请求，并对 response 进行处理

}

return findCoordinatorFuture;

}

// initiate the group metadata request

log.debug(“Sending GroupCoordinator request for group {} to broker {}”, groupId, node);

GroupCoordinatorRequest.Builder requestBuilder =

new GroupCoordinatorRequest.Builder(this.groupId);

return client.send(node, requestBuilder)

.compose(new GroupCoordinatorResponseHandler());

//NOTE: compose 的作用是将 GroupCoordinatorResponseHandler 类转换为 RequestFuture

//NOTE: 实际上就是为返回的 Future 类重置 onSuccess() 和 onFailure() 方法

}

private class GroupCoordinatorResponseHandler extends RequestFutureAdapter<ClientResponse, Void> {

@Override

public void onSuccess(ClientResponse resp, RequestFuture<Void> future) {

log.debug(“Received GroupCoordinator response {} for group {}”, resp, groupId);

GroupCoordinatorResponse groupCoordinatorResponse = (GroupCoordinatorResponse) resp.responseBody();

// use MAX_VALUE – node.id as the coordinator id to mimic separate connections

// for the coordinator in the underlying network client layer

// TODO: this needs to be better handled in KAFKA-1935

Errors error = Errors.forCode(groupCoordinatorResponse.errorCode());

clearFindCoordinatorFuture();

if (error == Errors.NONE) {

// note: 如果正确获取 GroupCoordinator 时, 建立连接,并更新心跳时间

synchronized (AbstractCoordinator.this) {

AbstractCoordinator.this.coordinator = new Node(

Integer.MAX_VALUE – groupCoordinatorResponse.node().id(),

groupCoordinatorResponse.node().host(),

groupCoordinatorResponse.node().port());

log.info(“Discovered coordinator {} for group {}.”, coordinator, groupId);

client.tryConnect(coordinator);//note: 初始化 tcp 连接

heartbeat.resetTimeouts(time.milliseconds());//note: 更新心跳时间

}

future.complete(null);

} else if (error == Errors.GROUP_AUTHORIZATION_FAILED) {

future.raise(new GroupAuthorizationException(groupId));

} else {

log.debug(“Group coordinator lookup for group {} failed: {}”, groupId, error.message());

future.raise(error);

}

}

@Override

public void onFailure(RuntimeException e, RequestFuture<Void> future) {

clearFindCoordinatorFuture();

super.onFailure(e, future);

}

}

Integer.MAX_VALUE – groupCoordinatorResponse.node().id(),

groupCoordinatorResponse.node().host(),

groupCoordinatorResponse.node().port());

// always ensure that the coordinator is ready because we may have been disconnected

// when sending heartbeats and does not necessarily require us to rejoin the group.

ensureCoordinatorReady();//NOTE: 确保 GroupCoordinator 已经连接

startHeartbeatThreadIfNeeded();//NOTE: 启动心跳发送线程（并不一定发送心跳,满足条件后才会发送心跳）

joinGroupIfNeeded();//NOTE: 发送 JoinGroup 请求,并对返回的信息进行处理

}

public void joinGroupIfNeeded() {

while (needRejoin() || rejoinIncomplete()) {

//再次确保GroupCoordinator准备就绪

ensureCoordinatorReady();

// call onJoinPrepare if needed. We set a flag to make sure that we do not call it a second

// time if the client is woken up before a pending rebalance completes. This must be called

// on each iteration of the loop because an event requiring a rebalance (such as a metadata

// refresh which changes the matched subscription set) can occur while another rebalance is

// still in progress.

//note: 触发 onJoinPrepare, 包括 offset commit 和 rebalance listener

if (needsJoinPrepare) {

onJoinPrepare(generation.generationId, generation.memberId);

needsJoinPrepare = false;

}

// note: 初始化 JoinGroup 请求,并发送该请求

RequestFuture<ByteBuffer> future = initiateJoinGroup();

client.poll(future);

resetJoinGroupFuture();//NOTE: 重置 joinFuture 为空

if (future.succeeded()) {//note: join succeed,这一步时,时间上 sync-group 已经成功了

needsJoinPrepare = true;

onJoinComplete(generation.generationId, generation.memberId, generation.protocol, future.value());

} else {

RuntimeException exception = future.exception();

if (exception instanceof UnknownMemberIdException ||

exception instanceof RebalanceInProgressException ||

exception instanceof IllegalGenerationException)

continue;

else if (!future.isRetriable())

throw exception;

time.sleep(retryBackoffMs);

}

}

}

private synchronized RequestFuture<ByteBuffer> initiateJoinGroup() {

// we store the join future in case we are woken up by the user after beginning the

// rebalance in the call to poll below. This ensures that we do not mistakenly attempt

// to rejoin before the pending rebalance has completed.

if (joinFuture == null) {

// fence off the heartbeat thread explicitly so that it cannot interfere with the join group.

// Note that this must come after the call to onJoinPrepare since we must be able to continue

// sending heartbeats if that callback takes some time.

// note: rebalance 期间,心跳线程停止

disableHeartbeatThread();

state = MemberState.REBALANCING;//NOTE: 标记为 rebalance

joinFuture = sendJoinGroupRequest();//NOTE: 发送 JoinGroup 请求

joinFuture.addListener(new RequestFutureListener<ByteBuffer>() {

@Override

public void onSuccess(ByteBuffer value) {

// handle join completion in the callback so that the callback will be invoked

// even if the consumer is woken up before finishing the rebalance

synchronized (AbstractCoordinator.this) {

log.info(“Successfully joined group {} with generation {}”, groupId, generation.generationId);

state = MemberState.STABLE;//NOTE: 标记 Consumer 为 stable

if (heartbeatThread != null)

heartbeatThread.enable();

}

}

@Override

public void onFailure(RuntimeException e) {

// we handle failures below after the request finishes. if the join completes

// after having been woken up, the exception is ignored and we will rejoin

synchronized (AbstractCoordinator.this) {

state = MemberState.UNJOINED;//NOTE: 标记 Consumer 为 Unjoined

}

}

});

}

return joinFuture;

}

if (coordinatorUnknown())

return RequestFuture.coordinatorNotAvailable();

// send a join group request to the coordinator

log.info(“(Re-)joining group {}”, groupId);

JoinGroupRequest.Builder requestBuilder = new JoinGroupRequest.Builder(

groupId,

this.sessionTimeoutMs,

this.generation.memberId,

protocolType(),

metadata()).setRebalanceTimeout(this.rebalanceTimeoutMs);

log.debug(“Sending JoinGroup ({}) to coordinator {}”, requestBuilder, this.coordinator);

return client.send(coordinator, requestBuilder)

.compose(new JoinGroupResponseHandler());

}

private class JoinGroupResponseHandler extends CoordinatorResponseHandler<JoinGroupResponse, ByteBuffer> {

@Override

public void handle(JoinGroupResponse joinResponse, RequestFuture<ByteBuffer> future) {

Errors error = Errors.forCode(joinResponse.errorCode());

if (error == Errors.NONE) {

log.debug(“Received successful JoinGroup response for group {}: {}”, groupId, joinResponse);

sensors.joinLatency.record(response.requestLatencyMs());

synchronized (AbstractCoordinator.this) {

if (state != MemberState.REBALANCING) {//NOTE: 如果此时 Consumer 的状态不是 rebalacing,就引起异常

// if the consumer was woken up before a rebalance completes, we may have already left

// the group. In this case, we do not want to continue with the sync group.

future.raise(new UnjoinedGroupException());

} else {

AbstractCoordinator.this.generation = new Generation(joinResponse.generationId(),

joinResponse.memberId(), joinResponse.groupProtocol());

AbstractCoordinator.this.rejoinNeeded = false;

//NOTE: join group 成功,下面需要进行 sync-group,获取分配的 tp 列表。

if (joinResponse.isLeader()) {

onJoinLeader(joinResponse).chain(future);

} else {

onJoinFollower().chain(future);

}

}

}

//下面为出现了问题后的流程,先抛开不看

} else if (error == Errors.GROUP_LOAD_IN_PROGRESS) {

log.debug(“Attempt to join group {} rejected since coordinator {} is loading the group.”, groupId,

coordinator());

// backoff and retry

future.raise(error);

} else if (error == Errors.UNKNOWN_MEMBER_ID) {

// reset the member id and retry immediately

resetGeneration();

log.debug(“Attempt to join group {} failed due to unknown member id.”, groupId);

future.raise(Errors.UNKNOWN_MEMBER_ID);

} else if (error == Errors.GROUP_COORDINATOR_NOT_AVAILABLE

|| error == Errors.NOT_COORDINATOR_FOR_GROUP) {

// re-discover the coordinator and retry with backoff

coordinatorDead();

log.debug(“Attempt to join group {} failed due to obsolete coordinator information: {}”, groupId, error.message());

future.raise(error);

} else if (error == Errors.INCONSISTENT_GROUP_PROTOCOL

|| error == Errors.INVALID_SESSION_TIMEOUT

|| error == Errors.INVALID_GROUP_ID) {

// log the error and re-throw the exception

log.error(“Attempt to join group {} failed due to fatal error: {}”, groupId, error.message());

future.raise(error);

} else if (error == Errors.GROUP_AUTHORIZATION_FAILED) {

future.raise(new GroupAuthorizationException(groupId));

} else {

// unexpected error, throw the exception

future.raise(new KafkaException(“Unexpected error in join group response: ” + error.message()));

}

}

}

private RequestFuture<ByteBuffer> onJoinFollower() {

// send follower’s sync group with an empty assignment

SyncGroupRequest.Builder requestBuilder =

new SyncGroupRequest.Builder(groupId, generation.generationId, generation.memberId,

Collections.<String, ByteBuffer>emptyMap());

log.debug(“Sending follower SyncGroup for group {} to coordinator {}: {}”, groupId, this.coordinator,

requestBuilder);

return sendSyncGroupRequest(requestBuilder);

}

//note: 当 consumer 客户端为 leader 时,对 group 下的所有实例进行分配,将 assign 的结果发送到 GroupCoordinator

private RequestFuture<ByteBuffer> onJoinLeader(JoinGroupResponse joinResponse) {

try {

// perform the leader synchronization and send back the assignment for the group

Map<String, ByteBuffer> groupAssignment = performAssignment(joinResponse.leaderId(), joinResponse.groupProtocol(),

joinResponse.members());//NOTE: 进行 assign 操作

SyncGroupRequest.Builder requestBuilder =

new SyncGroupRequest.Builder(groupId, generation.generationId, generation.memberId, groupAssignment);

log.debug(“Sending leader SyncGroup for group {} to coordinator {}: {}”,

groupId, this.coordinator, requestBuilder);

return sendSyncGroupRequest(requestBuilder);//NOTE: 发送 sync-group 请求

} catch (RuntimeException e) {

return RequestFuture.failure(e);

}

}

@Override

protected void onJoinComplete(int generation,

String memberId,

String assignmentStrategy,

ByteBuffer assignmentBuffer) {

// only the leader is responsible for monitoring for metadata changes (i.e. partition changes)

if (!isLeader)

assignmentSnapshot = null;

PartitionAssignor assignor = lookupAssignor(assignmentStrategy);

if (assignor == null)

throw new IllegalStateException(“Coordinator selected invalid assignment protocol: ” + assignmentStrategy);

Assignment assignment = ConsumerProtocol.deserializeAssignment(assignmentBuffer);

// set the flag to refresh last committed offsets

//note: 设置是否需要拉取 last committed offsets 为 true

subscriptions.needRefreshCommits();

// update partition assignment

//note: 更新订阅的 tp list

subscriptions.assignFromSubscribed(assignment.partitions());

// check if the assignment contains some topics that were not in the original

// subscription, if yes we will obey what leader has decided and add these topics

// into the subscriptions as long as they still match the subscribed pattern

//

// TODO this part of the logic should be removed once we allow regex on leader assign

Set<String> addedTopics = new HashSet<>();

for (TopicPartition tp : subscriptions.assignedPartitions()) {

if (!joinedSubscription.contains(tp.topic()))

addedTopics.add(tp.topic());

}

if (!addedTopics.isEmpty()) {

Set<String> newSubscription = new HashSet<>(subscriptions.subscription());

Set<String> newJoinedSubscription = new HashSet<>(joinedSubscription);

newSubscription.addAll(addedTopics);

newJoinedSubscription.addAll(addedTopics);

this.subscriptions.subscribeFromPattern(newSubscription);

this.joinedSubscription = newJoinedSubscription;

}

// update the metadata and enforce a refresh to make sure the fetcher can start

// fetching data in the next iteration

//note: 更新 metadata,确保在下一次循环中可以拉取

this.metadata.setTopics(subscriptions.groupSubscription());

client.ensureFreshMetadata();

// give the assignor a chance to update internal state based on the received assignment

assignor.onAssignment(assignment);

// reschedule the auto commit starting from now

this.nextAutoCommitDeadline = time.milliseconds() + autoCommitIntervalMs;

// execute the user’s callback after rebalance

//note: 执行 listener

ConsumerRebalanceListener listener = subscriptions.listener();

log.info(“Setting newly assigned partitions {} for group {}”, subscriptions.assignedPartitions(), groupId);

try {

Set<TopicPartition> assigned = new HashSet<>(subscriptions.assignedPartitions());

listener.onPartitionsAssigned(assigned);

} catch (WakeupException | InterruptException e) {

throw e;

} catch (Exception e) {

log.error(“User provided listener {} for group {} failed on partition assignment”,

listener.getClass().getName(), groupId, e);

}

}