vivo Pulsar 万亿级消息处理实践（3）-KoP指标异常修复

kop_server_MESSAGE_OUT{group="",partition="0",tenant="kop",topic="persistent://kop-tenant/kop-ns/service-raw-stats"} 3
kop_server_BYTES_OUT{group="",partition="0",tenant="kop",topic="persistent://kop-tenant/kop-ns/service-raw-stats"} 188

private void handleEntries(final List<Entry> entries,
                               final TopicPartition topicPartition,
                               final FetchRequest.PartitionData partitionData,
                               final KafkaTopicConsumerManager tcm,
                               final ManagedCursor cursor,
                               final AtomicLong cursorOffset,
                               final boolean readCommitted) {
....
        // 处理消费数据时，获取消费组名称
        CompletableFuture<String> groupNameFuture = requestHandler
                .getCurrentConnectedGroup()
                .computeIfAbsent(clientHost, clientHost -> {
                    CompletableFuture<String> future = new CompletableFuture<>();
                    String groupIdPath = GroupIdUtils.groupIdPathFormat(clientHost, header.clientId());
                    requestHandler.getMetadataStore()
                            .get(requestHandler.getGroupIdStoredPath() + groupIdPath)
                            .thenAccept(getResultOpt -> {
                                if (getResultOpt.isPresent()) {
                                    GetResult getResult = getResultOpt.get();
                                    future.complete(new String(getResult.getValue() == null
                                            ? new byte[0] : getResult.getValue(), StandardCharsets.UTF_8));
                                } else {
                                    // 从zk节点 client_group_id/xxx 获取不到消费组，消费组就是空的
                                    future.complete("");
                                }
                            }).exceptionally(ex -> {
                                future.completeExceptionally(ex);
                                return null;
                            });
                    returnfuture;
                });


        // this part is heavyweight, and we should not execute in the ManagedLedger Ordered executor thread
        groupNameFuture.whenCompleteAsync((groupName, ex) -> {
            if (ex != null) {
                log.error("Get groupId failed.", ex);
                groupName = "";
            }
.....
            // 获得消费组名称后，记录消费组对应的消费指标
            decodeResult.updateConsumerStats(topicPartition,
                    entries.size(),
                    groupName,
                    statsLogger);

    @Override
    protected void handleFindCoordinatorRequest(KafkaHeaderAndRequest findCoordinator,
                                                CompletableFuture<AbstractResponse> resultFuture) {
...
        // Store group name to metadata store for current client, use to collect consumer metrics.
        storeGroupId(groupId, groupIdPath)
                .whenComplete((stat, ex) -> {
                    if (ex != null) {
                        // client_group_id/xxx节点写入失败
                        log.warn("Store groupId failed, the groupId might already stored.", ex);
                    }
                    findBroker(TopicName.get(pulsarTopicName))
                            .whenComplete((node, throwable) -> {
                                ....
                            });
                });
...

protected void close(){
    if (isActive.getAndSet(false)) {
        ...
        currentConnectedClientId.forEach(clientId -> {
            String path = groupIdStoredPath + GroupIdUtils.groupIdPathFormat(clientHost, clientId);
            // 删除zk上的 client_group_id/xxx 节点
            metadataStore.delete(path, Optional.empty())
                    .whenComplete((__, ex) -> {
                        if (ex != null) {
                            if (ex.getCause() instanceof MetadataStoreException.NotFoundException) {
                                if (log.isDebugEnabled()) {
                                    log.debug("The groupId store path doesn't exist. Path: [{}]", path);
                                }
                                return;
                            }
                            log.error("Delete groupId failed. Path: [{}]", path, ex);
                            return;
                        }
                        if (log.isDebugEnabled()) {
                            log.debug("Delete groupId success. Path: [{}]", path);
                        }
                    });
        });
    }
}

3、复现

若要解决问题，最好能够稳定地复现出问题，这样才能确定问题的根本原因，并且确认修复是否完成。

既然知道把消费者与FindCoordinator的连接进行绑定不合适的，那么是否应该把FindCoordinator写入zk节点换成由JoinGroup写入，断连即删除。

还是由FindCoordinator写入zk节点，但删除改为GroupCoordinator监听consumer断开触发。

因为consumer统一由Coordinator管理，它能监听到consumer加入或者离开。GroupCoordinator的removeMemberAndUpdateGroup方法是coordinator对consumer成员管理。

private void removeMemberAndUpdateGroup(GroupMetadata group,
                                        MemberMetadata member) {
    group.remove(member.memberId());
    switch (group.currentState()) {
        case Dead:
        case Empty:
            return;
        case Stable:
        case CompletingRebalance:
            maybePrepareRebalance(group);
            break;
        case PreparingRebalance:
            joinPurgatory.checkAndComplete(new GroupKey(group.groupId()));
            break;
        default:
            break;
    }
    // 删除 client_group_id/xxx 节点
    deleteClientIdGroupMapping(group, member.clientHost(), member.clientId());
}

public CompletableFuture<Errors> handleLeaveGroup(
    String groupId,
    String memberId
) {
    return validateGroupStatus(groupId, ApiKeys.LEAVE_GROUP).map(error ->
        CompletableFuture.completedFuture(error)
    ).orElseGet(() -> {
        return groupManager.getGroup(groupId).map(group -> {
            return group.inLock(() -> {
                if (group.is(Dead) || !group.has(memberId)) {
                    return CompletableFuture.completedFuture(Errors.UNKNOWN_MEMBER_ID);
                } else {
                    ...
                
                    // 触发删除消费者consumer
                    removeMemberAndUpdateGroup(group, member);
                    return CompletableFuture.completedFuture(Errors.NONE);
                }
            });
        })
        ....
    });
}

void onExpireHeartbeat(GroupMetadata group,
                       MemberMetadata member,
                       long heartbeatDeadline) {
    group.inLock(() -> {
        if (!shouldKeepMemberAlive(member, heartbeatDeadline)) {
            log.info("Member {} in group {} has failed, removing it from the group",
                member.memberId(), group.groupId());
            // 触发删除消费者consumer
            removeMemberAndUpdateGroup(group, member);
        }
        return null;
    });
}

其实这并不是新的方案，只是在方案二基础上优化：数据一致性检查。

既然我们很难控制好写入与删除的先后顺序，我们可以做数据一致性检查，类似于交易系统里的对账。因为GroupCoordinator是负责管理consumer成员的，维护着consumer的实时状态，就算zk节点被误删除，我们也可以从consumer成员信息中恢复，重新写入zk节点。

private void checkZkGroupMapping(){  
    for (GroupMetadata group : groupManager.currentGroups()) {  
        for (MemberMetadata memberMetadata : group.allMemberMetadata()) {  
            String clientPath = GroupIdUtils.groupIdPathFormat(memberMetadata.clientHost(), memberMetadata.clientId());  
            String zkGroupClientPath = kafkaConfig.getGroupIdZooKeeperPath() + clientPath;  
            // 查找zk中是否存在节点
            metadataStore.get(zkGroupClientPath).thenAccept(resultOpt -> {  
                if (!resultOpt.isPresent()) {  
                    // 不存在则进行补偿修复
                    metadataStore.put(zkGroupClientPath, memberMetadata.groupId().getBytes(UTF\_8), Optional.empty())  
                            .thenAccept(stat -> {  
                                log.info("repaired clientId and group mapping: {}({})",  
                                        zkGroupClientPath, memberMetadata.groupId());  
                            })  
                            .exceptionally(ex -> {  
                                log.warn("repaired clientId and group mapping failed: {}({})",  
                                        zkGroupClientPath, memberMetadata.groupId());  
                                return null;  
                            });  
                }  
            }).exceptionally(ex -> {  
                log.warn("repaired clientId and group mapping failed: {} ", zkGroupClientPath, ex);  
                return null;  
            });  
        }  
    }  
}