一.在分析Rebalance之前，先列出哪几种情况下会触发Rebalance
- 有新的消费者加入Consumer Group
- 有消费者宕机下线。消费者不一定真的宕机了，比如长时间的GC，网络的延迟导致消费者长时间未向GroupCoordinator发送HeartbeatRequest时，GroupCoordinator会认为消费者下线。
- 有消费者主动退出Consumer Group。
- Consumer Group订阅的任何一个Topic出现分区数量的变化。
- 消费者调用unsubscribe()取消对某个Topic的订阅。
二.分析Rebalance的具体操作。
第一个阶段：
Rebalance操作的第一步是查找GroupCoordinator,这个阶段消费者会向Kafka集群中的任意一个Broker发送GroupCoordinatorRequest请求，并处理返回的GroupCoordinatorResponse响应。
GroupCoordinatorRequest消息体的格式比较简单，仅仅含了Consumer Group的id。GroupCoordinatorResponse消息体包含了错误码(short类型),coordinator的节点id,GroupCoordinator的host,GroupCoordinator的端口号。
发送GroupCoordinatorRequest请求的入口是ConsumerCoordinator的ensureCoordinatorReady()方法，具体流程如下：
ensureCoordinatorReady()方法处理流程.jpg
(1)首先检测是否需要重新查找GroupCoordinator。
主要是检测coordinator字段是否为空以及与GroupCoordinator之间的连接是否正常。
/**
* Check if we know who the coordinator is and we have an active connection
* @return true if the coordinator is unknown
*/
public boolean coordinatorUnknown() {
if (coordinator == null)//检测Coordinator字段是否为null
return true;
//检测与GroupCoodinator之间的网络连接是否正常
if (client.connectionFailed(coordinator)) {
//将unsent集合中对应的请求清空并将coordinator字段设置为null
coordinatorDead();
return true;
}
return false;
}
(2)查找集群负载最低的节点的Node节点，并创建GroupCoordinatorRequest请求。
调用client.send方法将请求放入unsent队列中等待发送，并返回RequestFuture<Void>对象。返回的RequestFuture<Void>对象经过了compose()方法适配，原理同HeartbeatCompletionHandler.
protected RequestFuture<Void> lookupCoordinator() {
if (findCoordinatorFuture == null) {
findCoordinatorFuture = sendGroupCoordinatorRequest();
findCoordinatorFuture.addListener(new RequestFutureListener<Void>() {
@Override
public void onSuccess(Void value) {
findCoordinatorFuture = null;
}
@Override
public void onFailure(RuntimeException e) {
findCoordinatorFuture = null;
}
});
}
return findCoordinatorFuture;
}
/**
* Discover the current coordinator for the group. Sends a GroupMetadata request to
* one of the brokers. The returned future should be polled to get the result of the request.
* @return A request future which indicates the completion of the metadata request
*/
private RequestFuture<Void> sendGroupCoordinatorRequest() {
// initiate the group metadata request
// find a node to ask about the coordinator
Node node = this.client.leastLoadedNode();
if (node == null) {
// TODO: If there are no brokers left, perhaps we should use the bootstrap set
// from configuration?
return RequestFuture.noBrokersAvailable();
} else {
// create a group metadata request
log.debug("Sending coordinator request for group {} to broker {}", groupId, node);
GroupCoordinatorRequest metadataRequest = new GroupCoordinatorRequest(this.groupId);
return client.send(node, ApiKeys.GROUP_COORDINATOR, metadataRequest)
.compose(new RequestFutureAdapter<ClientResponse, Void>() {
@Override
public void onSuccess(ClientResponse response, RequestFuture<Void> future) {
handleGroupMetadataResponse(response, future);
}
});
}
}
(3)调用ConsumerNetworkClient.poll(future)方法，将GroupCoordinatorRequest请求发送出去。
这里是用阻塞的方式发送，直到收到GroupCoordinatorResponse响应或异常完成才会从此方法返回。GroupCoordinatorResponse具体流程后面介绍。
(4) 检测RequestFuture<Void>对象的状态。
如果出现RetriableException异常，则调用ConsumerNetworkClient.awaitMetadataUpdate()方法阻塞更新Metadata中记录的集群元数据后跳转到步骤1继续执行。如果不是RetriableException异常则直接报错。
(5) 如果成功找到GroupCoordinator节点，但是网络连接失败，则将unsent中对应的请求清空，并将coordinator字段置为null,准备重新查找GroupCoordinator，退避一段时间后跳转到步骤1继续执行。
ensureCoordinatorReady()方法的具体实现：
/**
* Block until the coordinator for this group is known and is ready to receive requests.
*/
public void ensureCoordinatorReady() {//步骤1：检测GroupCoordinator的状态。
while (coordinatorUnknown()) {
RequestFuture<Void> future = lookupCoordinator();
//步骤2：创建并缓存请求。
//步骤3：阻塞发GroupCoordinatorRequest，并处理GroupCoordinatorResponse。
client.poll(future);
if (future.failed()) {//步骤4：异常处理
if (future.isRetriable())
client.awaitMetadataUpdate();//阻塞更新Metadata中记录的集群元数据
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();//步骤5：连接不到GroupCoordinator，退避一段时间，重试
time.sleep(retryBackoffMs);
}
}
}
下面分析下方法sendGroupCoordinatorRequest()：
/**
* Discover the current coordinator for the group. Sends a GroupMetadata request to
* one of the brokers. The returned future should be polled to get the result of the request.
* @return A request future which indicates the completion of the metadata request
*/
private RequestFuture<Void> sendGroupCoordinatorRequest() {
// initiate the group metadata request
// find a node to ask about the coordinator
//查找负载最低的节点，底层原理是InFlightRequest中未确认请求最少的节点。
Node node = this.client.leastLoadedNode();
if (node == null) {//如果找不到可用的节点，则直接返回一个异常结束的RequestFuture。
// TODO: If there are no brokers left, perhaps we should use the bootstrap set
// from configuration?
return RequestFuture.noBrokersAvailable();
} else {
// create a group metadata request
//创建GroupCoordinatorRequest请求，将GroupCoordinatorRequest缓存到unsent集合，
//
log.debug("Sending coordinator request for group {} to broker {}", groupId, node);
GroupCoordinatorRequest metadataRequest = new GroupCoordinatorRequest(this.groupId);
//使用匿名RequestFutureAdapter
return client.send(node, ApiKeys.GROUP_COORDINATOR, metadataRequest)
.compose(new RequestFutureAdapter<ClientResponse, Void>() {
@Override
public void onSuccess(ClientResponse response, RequestFuture<Void> future) {
//处理 GroupMetadataResponse的入口
handleGroupMetadataResponse(response, future);
}
});
}
}
处理GroupCoordinatorResponse的相关操作
通过对上述sendGroupCoordinatorRequest()方法的分析，handleGroupMetadataResponse(response, future)方法是处理GroupCoordinatorResponse的入口，步骤如下：
- 调用coordinatorUnknown()检测是否已经找到GroupCoordinator且成功连接。如果是则忽略此GroupCoordinatorResponse，因为发送GroupCoordinatorRequest时没有防止重发的机制，可能会有多个GroupCoordinatorResponse，所以这里要防止重复的情况。如果是否则走下面的步骤。
- 解析GroupCoordinatorResponse得到服务端GroupCoordinator的信息。
- 构建Node对象赋值给coordinator对象，并尝试与GroupCoordinator建立连接。
- 启动HeartbeatTask定时任务。
- 调用RequestFuture.complete()方法将正常收到GroupCoordinatorResponse的事件传播出去。
- 如果GroupCoordinatorResponse中的错误码不为NONE,则调用RequestFuture.raise()方法将异常传播出去。最终由ensureCoordinatorReady()方法中的第四步处理。
handleGroupMetadataResponse()方法的具体实现如下：
private void handleGroupMetadataResponse(ClientResponse resp, RequestFuture<Void> future) {
log.debug("Received group coordinator response {}", resp);
if (!coordinatorUnknown()) {
// We already found the coordinator, so ignore the request
//检测是否已经找到GroupCoordinator且成功连接
future.complete(null);
} else {
//解析GroupCoordinatorResponse
GroupCoordinatorResponse groupCoordinatorResponse = new 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());
if (error == Errors.NONE) {
//构建Node对象赋值给coordinator对象
this.coordinator = new Node(Integer.MAX_VALUE - groupCoordinatorResponse.node().id(),
groupCoordinatorResponse.node().host(),
groupCoordinatorResponse.node().port());
log.info("Discovered coordinator {} for group {}.", coordinator, groupId);
//尝试与GroupCoordinator建立连接
client.tryConnect(coordinator);
// start sending heartbeats only if we have a valid generation
if (generation > 0)
heartbeatTask.reset();//启动HeartbeatTask定时任务
future.complete(null);//调用RequestFuture.complete()方法将正常收到GroupCoordinatorResponse的事件传播出去。
} else if (error == Errors.GROUP_AUTHORIZATION_FAILED) {
//如果GroupCoordinatorResponse中的错误码不为NONE,则调用RequestFuture.raise()方法将异常传播出去。
future.raise(new GroupAuthorizationException(groupId));
} else {
future.raise(error);
}
}
}
网友评论