当前位置: 首页 > news >正文

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

作者:vivo 互联网大数据团队- Chen Jianbo

本文是《vivo Pulsar万亿级消息处理实践》系列文章第3篇。

Pulsar是Apache基金会的开源分布式流处理平台和消息中间件,它实现了Kafka的协议,可以让使用Kafka API的应用直接迁移至Pulsar,这使得Pulsar在Kafka生态系统中更加容易被接受和使用。KoP提供了从Kafka到Pulsar的无缝转换,用户可以使用Kafka API操作Pulsar集群,保留了Kafka的广泛用户基础和丰富生态系统。它使得Pulsar可以更好地与Kafka进行整合,提供更好的消息传输性能、更强的兼容性及可扩展性。vivo在使用Pulsar KoP的过程中遇到过一些问题,本篇主要分享一个分区消费指标缺失的问题。

系列文章:

  1. vivo Pulsar万亿级消息处理实践(1)-数据发送原理解析和性能调优

  2. vivo Pulsar万亿级消息处理实践(2)-从0到1建设Pulsar指标监控链路

文章太长?1分钟看图抓住核心观点👇

图片

一、问题背景

在一次版本灰度升级中,我们发现某个使用KoP的业务topic的消费速率出现了显著下降,具体情况如下图所示:

什么原因导致正常的升级重启服务器会出现这个问题呢?直接查看上报采集的数据报文:

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

我们看到,KoP消费指标kop_server_MESSAGE

_OUT、kop_server_BYTES_OUT是有上报的,但指标数据里的group标签变成了空串(缺少消费组名称),分区的消费指标就无法展示了。是什么原因导致了消费组名称缺失?

二、问题分析

1、找到问题代码

我们去找下这个消费组名称是在哪里获取的,是否逻辑存在什么问题。根据druid中的kop_subscription对应的消费指标kop_server_

MESSAGE_OUT、kop_server_BYTES_OUT,找到相关代码如下:

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 threadgroupNameFuture.whenCompleteAsync((groupName, ex) -> {if (ex != null) {log.error("Get groupId failed.", ex);groupName = "";}
.....// 获得消费组名称后,记录消费组对应的消费指标decodeResult.updateConsumerStats(topicPartition,entries.size(),groupName,statsLogger);

代码的逻辑是,从requestHandler的currentConnectedGroup(map)中通过host获取groupName,不存在则通过MetadataStore(带缓存的zk存储对象)获取,如果zk缓存也没有,再发起zk读请求(路径为/client_group_id/host-clientId)。读取到消费组名称后,用它来更新消费组指标。从复现的集群确定走的是这个分支,即是从metadataStore(带缓存的zk客户端)获取不到对应zk节点/client_group_id/xxx。

2、查找可能导致zk节点/client_group_id/xxx节点获取不到的原因

有两种可能性:一是没写进去,二是写进去但是被删除了。

    @Overrideprotected 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) -> {....});});
...

从代码看到,clientId与groupId的关联关系是通过handleFindCoordinatorRequest(FindCoordinator)写进去的,而且只有这个方法入口。由于没有找到warn日志,排除了第一种没写进去的可能性。看看删除的逻辑:

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);}});});}
}

删除是在requsetHandler.close方法中执行,也就是说连接断开就会触发zk节点删除。

但有几个疑问:

  • /client_group_id/xxx 到底是干嘛用的?消费指标为什么要依赖它

  • 为什么要在handleFindCoordinatorRequest写入?

  • 节点/client_group_id/xxx为什么要删除,而且是在连接断开时删除,删除时机是否有问题?

首先回答第1个问题,通过阅读代码可以知道,/client_group_id/xxx 这个zk节点是用于在不同broker实例间交换数据用的(相当redis cache),用于临时存放IP+clientId与groupId的映射关系。由于fetch接口(拉取数据)的request没有groupId的,只能依赖加入Group过程中的元数据,在fetch消费时才能知道当前拉数据的consumer是哪个消费组的。

3、复现

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

因为节点是在requsetHandle.close方法中执行删除,broker节点关闭会触发连接关闭,进而触发删除。假设:客户端通过brokerA发起FindCoordinator请求,写入zk节点/client_group

_id/xxx,同时请求返回brokerB作为Coordinator,后续与brokerB进行joinGroup、syncGroup等交互确定消费关系,客户端在brokerA、brokerB、brokerC都有分区消费。这时重启brokerA,分区均衡到BrokerC上,但此时/client_group_id/xxx因关闭broker而断开连接被删除,consumer消费刚转移到topic1-partition-1的分区就无法获取到groupId。

按照假设,有3个broker,开启生产和消费,通过在FindCoordinator返回前获取node.leader()的返回节点BrokerB,关闭brokerA后,brokerC出现断点复现,再关闭brokerC,brokerA也会复现(假设分区在brokerA与brokerC之间转移)。

图片

复现要几个条件:

  1. broker数量要足够多(不小于3个)

  2. broker内部有zk缓存metadataCache默认为5分钟,可以把时间调小为1毫秒,相当于没有cache

  3. findCoordinator返回的必须是其他broker的IP

  4. 重启的必须是接收到findCoordinator请求那台broker,而不是真正的coordinator,这时会从zk删除节点

  5. 分区转移到其他broker,这时新的broker会重新读取zk节点数据

到此,我们基本上清楚了问题原因:连接关闭导致zk节点被删除了,别的broker节点需要时就读取不到了。那怎么解决?

三、问题解决

方案一

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

图片

consumer统一由Coordinator管理,由于FindCoordinator接口不一定是Coordinator处理的,如果换成由Coordinator处理的JoinGroup接口是否就可以了,这样consumer断开与Coordinator的连接就应该删除数据。但实现验证时却发现,客户端在断连后也不会再重连,所以没法重新写入zk,不符合预期。

方案二

还是由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());
}

调用入口有两个,其中handleLeaveGroup是主动离开,onExpireHeartbeat是超时被动离开,客户端正常退出或者宕机都可以调用removeMemberAndUpdateGroup方法触发删除。

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 {...// 触发删除消费者consumerremoveMemberAndUpdateGroup(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());// 触发删除消费者consumerremoveMemberAndUpdateGroup(group, member);}return null;});
}

但这个方案有个问题是,日志运维关闭broker也会触发一个onExpireHeartbeat事件删除zk节点,与此同时客户端发现Coordinator断开了会马上触发FindCoordinator写入新的zk节点,但如果删除晚于写入的话,会导致误删除新写入的节点。我们干脆在关闭broker时,使用ShutdownHook加上shuttingdown状态防止关闭broker时删除zk节点,只有客户端断开时才删除。

这个方案修改上线半个月后,还是出现了一个客户端的消费指标无法上报的情况。后来定位发现,如果客户端因FullGC出现卡顿情况,客户端可能会先于broker触发超时,也就是先超时的客户端新写入的数据被后监听到超时的broker误删除了。因为写入与删除并不是由同一个节点处理,所以无法在进程级别做并发控制,而且也无法判断哪次删除对应哪次的写入,所以用zk也是很难实现并发控制。

方案三

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

既然我们很难控制好写入与删除的先后顺序,我们可以做数据一致性检查,类似于交易系统里的对账。因为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;  });  }  }  
}

经过方案三的优化上线,即使是历史存在问题的消费组,个别分区消费流量指标缺少group字段的问题也得到了修复。具体效果如下图所示:

图片

四、总结

经过多个版本的优化和线上验证,最终通过方案三比较完美的解决了这个消费指标问题。在分布式系统中,并发问题往往难以模拟和复现,我们也在尝试多个版本后才找到有效的解决方案。如果您在这方面有更好的经验或想法,欢迎提出,我们共同探讨和交流。

http://www.dtcms.com/a/271824.html

相关文章:

  • 快速上手MongoDB与.NET/C#整合
  • 【AI大模型】LLM模型架构深度解析:BERT vs. GPT vs. T5
  • searxng 对接openweb-UI实现大模型通过国内搜索引擎在线搜索
  • 搜索引擎vs向量数据库:LangChain混合检索架构实战解析
  • 计算机视觉 之 数字图像处理基础
  • 基于 SpringBoot + Vue 的 IT 技术交流和分享平台的设计与实现
  • TCP-与-UDP-协议详解:原理、区别与应用场景全解析
  • 北斗舞动在线监测装置:电力安全的“智慧守护者”
  • SpringMVC @ExceptionHandler 典型用法
  • 了解去中心化金融在现代经济中的作用——安全交易新时代
  • 编写bat文件自动打开chrome浏览器,并通过selenium抓取浏览器操作chrome
  • 双指针-18.四数之和-力扣(LeetCode)
  • linux系统---ISCSI存储服务
  • Language Models are Few-Shot Learners: 开箱即用的GPT-3(二)
  • 节点小宝:手机图片备份至电脑功能实测体验
  • 同一类型,每条数据,执行不同逻辑
  • 偏振相机,偏振图像是怎么样的
  • WebGPU了解
  • 智能体决策机制深度剖析:ReAct、Plan-and-Execute与自适应策略
  • 云蝠智能VoiceAgent重构企业电话客服体系
  • PLC框架-1.3.2 报文750控制汇川伺服的转矩上下限
  • 【前缀和 BFS 并集查找】P3127 [USACO15OPEN] Trapped in the Haybales G|省选-
  • XSS(跨站脚本攻击)
  • RabbitMQ 消息队列:从入门到Spring Boot实战
  • Java 枚举详解:从基础到实战,掌握类型安全与优雅设计
  • 7-语言模型
  • CRT 不同会导致 fopen 地址不同
  • 技术演进中的开发沉思-30 MFC系列:五大机制
  • 删除k8s安装残留
  • Spring Boot:将应用部署到Kubernetes的完整指南