消息队列流处理后端微服务消息路由【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pu/pulsar点击查看免费下载导读PIP-91PIP-91: Separate lookup timeout from operation timeout是 Apache Pulsar 客户端侧的一项关键改进提案它针对客户端连接 Topic 时必经的分区元数据请求PMR与Lookup 请求在超时与重试处理上的不一致问题提出将两者所用的超时上限与操作超时operation timeout解耦、让TooManyRequests对这两类请求可重试并为二进制协议 Lookup 提供独立的连接池。读完本文你将理解 Pulsar 客户端 Topic 路由的底层机制、超时与退避backoff的相互作用以及如何在 Java 客户端中通过lookupTimeout()配置项精准控制查找行为让客户端在 Broker 滚动重启等大规模 Topic 迁移场景下更具韧性。一、背景一次 Topic 连接背后发生了什么在 Pulsar 中客户端连接一个 Topic 时并不直接向拥有该 Topic 的 Broker发消息而是先做一次lookup查找以定位当前持有该 Topic 元数据与所有权ownership的 Broker。整个过程包含两个请求分区元数据请求Partition Metadata Request简称 PMR查询该 Topic 是否为分区 Topic、有多少个分区Lookup 请求针对 Topic或其每个分区向集群询问哪个 Broker 负责该 Topic。这两个请求有一个共同特点它们可以由集群中的任意一个 Broker 服务而不像生产produce与消费consume请求那样必须路由到 Topic 的 owner Broker。因此它们的请求频率并不与系统流量成正比而是与Topic 在集群中的迁移/移动频率成正比。例如当集群发生滚动重启rolling restart时即使流过系统的消息数量保持不变所有客户端也必须重新对所有 Topic 做一次 lookup——此时查找类请求会出现瞬时的大规模爆发。从语义上看PMR 与 Lookup 都是无副作用side effect free的对同一个 Topic 的多次重复请求不会改变系统状态。正因为如此PIP-91 认为这两类请求应当比其他请求对瞬时错误transient errors具有更强的韧性——重试它们不会带来状态污染却能显著提升集群动荡期客户端的自愈能力。二、现状问题两套请求的超时与重试不一致PIP-91 指出PMR 与 Lookup 在超时与重试策略上存在不一致甚至缺陷的处理PMR 的超时重试请求超时发生在配置的操作超时operation timeout默认 30 秒之后理论上 PMR 只在操作超时窗口内重试因此不应该发生重试。但实际存在一个bug退避时间backoff time被计入超时预算而等待请求超时所耗的时间却未被计入导致 PMR 在超时后实际上仍然会重试。此外PMR明确不对TooManyRequests响应进行重试。Lookup 的超时重试同样存在重试逻辑但 Lookup 允许的最大总时长与操作超时相等导致超时后实际上没有有效的重试机会。与 PMR 相反Lookup 对TooManyRequests是有重试的。重试打在同一后端的风险当客户端通过负载均衡器loadbalancer接入 Pulsar 时所有 Lookup 都使用同一个 IP 端点。虽然请求会被均衡分发到各后端 Broker但一旦发生瞬时错误由于连接是持久persistent的重试会精确地落到刚才出问题的同一个后端无法换一台 Broker 重试。三、PIP-91 提案的三个核心改动针对上述问题PIP-91 提出三项改动新增独立的 Lookup 超时lookup timeout其值是操作超时的倍数multiple of operation timeout用于替换 PMR 与 Lookup 所用 Backoff 对象中的最大重试时长上限让TooManyRequests对 PMR 与 Lookup 都可重试使客户端在 Broker 限流/过载时能够换一台 Broker 再试让二进制协议binary protoLookup 使用独立的连接池并在请求出错时关闭对应连接可配置。关闭并重建连接可以让负载均衡器重新为客户端挑选一台新的后端从而规避重试永远落在故障节点的问题。默认行为保持不变lookup timeout 的默认值匹配 operation timeout未显式配置的客户端不会感知到行为变化。四、仓库源码中的落地实现PIP-91 提出的 lookup timeout 配置在 Apache Pulsar Java 客户端中已经完整落地以下是关键实现证据均在当前仓库中可查。4.1 配置字段lookupTimeoutMs在 ClientConfigurationData.java 中定义了lookupTimeoutMs字段默认值为-1Schema( name operationTimeoutMs, description Client operation timeout (in milliseconds). ) private long operationTimeoutMs 30000; Schema( name lookupTimeoutMs, description Client lookup timeout (in milliseconds). ) private long lookupTimeoutMs -1;读取时的逻辑体现了默认回落到 operation timeout的设计见 同文件 L572-L578public long getLookupTimeoutMs() { if (lookupTimeoutMs 0) { return lookupTimeoutMs; } else { return operationTimeoutMs; } }也就是说未配置时 lookup timeout 完全等于 operation timeout默认 30 秒与 PIP-91 提案中的默认行为不变完全一致一旦显式设置Lookup 类请求便获得独立的超时预算。4.2 Builder APIlookupTimeout(int, TimeUnit)公共 API 层在 ClientBuilder.java 中声明/** * param lookupTimeout lookup timeout * param unit time unit for {code lookupTimeout} */ ClientBuilder lookupTimeout(int lookupTimeout, TimeUnit unit);具体实现位于 ClientBuilderImpl.java将其转换为毫秒写入配置Override public ClientBuilder lookupTimeout(int lookupTimeout, TimeUnit unit) { conf.setLookupTimeoutMs(unit.toMillis(lookupTimeout)); return this; }典型用法示例PulsarClient client PulsarClient.builder() .serviceUrl(pulsar://broker.example.com:6650) // 显式拉高 lookup 超时例如 60 秒为集群滚动重启留出重试余量 .lookupTimeout(60, TimeUnit.SECONDS) // operationTimeout 保持默认 30 秒互不影响 .build();4.3 消费端使用lookup deadlinegetLookupTimeoutMs()被客户端各核心组件用来计算查找截止时间lookup deadline例如ProducerImpl.javathis.lookupDeadline System.currentTimeMillis() client.getConfiguration().getLookupTimeoutMs();ConsumerImpl.java 与之对称TransactionMetaStoreHandler.java 在定位事务元数据存储transaction meta store时同样使用该截止时间WatchTcAssignmentsDiscovery.java 在事务协调器TC分配发现流程中读取getLookupTimeoutMs()。这表明 lookup timeout 不仅作用于普通 Topic 的 PMR/Lookup也覆盖了事务协调器发现等内部查找路径。4.4 PMR 路径用 lookup timeout 作为退避上限在 PulsarClientImpl.java 的getPartitionedTopicMetadata中PMR 的重试 Backoff 直接以getLookupTimeoutMs()作为超时预算AtomicLong opTimeoutMs new AtomicLong(conf.getLookupTimeoutMs()); Backoff backoff Backoff.builder() .initialDelay(Duration.ofNanos(conf.getInitialBackoffIntervalNanos())) .mandatoryStop(Duration.ofMillis(opTimeoutMs.get() * 2)) .maxBackoff(Duration.ofNanos(conf.getMaxBackoffIntervalNanos())) .build();可以看到mandatoryStop强制终止时间由 lookup timeout 决定这正对应 PIP-91 中用 lookup timeout 替换 operation timeout 作为 Backoff 最大重试时长的提议。4.5 HTTP 查找路径对于通过http/https服务地址进行的查找HttpClient.java 也读取了 lookup timeout 并构造超时预算Duration budget当配置值不大于 0 时回落到 60 秒兜底值。4.6 关于独立连接池配置PIP-91 中提出的第三个改动二进制协议 Lookup 使用独立连接池、出错时关闭连接属于可配置项当前仓库中未发现与该配置同名的字段读者应以本仓库实际提供的客户端配置为准并关注 PIP 后续演进版本中的实现情况。五、测试验证Lookup 重试行为PIP-91 的测试计划指出该改动将做单元测试新增 mock 以便从ServerCnx轻松触发特定行为超时、TooManyRequests。仓库中的集成测试 LookupRetryTest.java 正是对这一行为的验证。该测试构造了一个模拟 Broker分别对分区元数据请求第 324 行与 Lookup 请求第 338 行主动返回ServerError.TooManyRequests例如ctx.writeAndFlush(newPartitionMetadataResponse(ServerError.TooManyRequests, too many, requestId)); ctx.writeAndFlush(newLookupErrorResponse(ServerError.TooManyRequests, too many, requestId));测试中客户端通过.lookupTimeout(10, TimeUnit.SECONDS)见测试文件第 106、260、273 行显式设置 lookup timeout并验证在连续TooManyRequests下的重试与最终失败行为第 224-238 行的注释说明了MaxNumberOfRejectedRequestPerConnection1时客户端应在连续TooManyRequests下失败、以及指数退避下 1 次与 4 次拒绝的差异。这为TooManyRequests可重试以及重试次数受退避策略控制提供了可复现的测试依据。此外ServiceUrlQuarantineTest.java 同样涉及 lookup timeout 相关配置的验证说明该配置已进入多场景的回归测试覆盖。六、兼容性与迁移注意事项PIP-91 明确指出该提案可能带来的一处行为变化PMR 超时计算的 bug 修复修复后 PMR 的整体超时时间更严格不再把退避时间以外的等待时间漏算进预算因此部分此前被 bug 掩盖的瞬时错误将重新暴露为超时异常。对于这类用户解决方案是显式把 lookup timeout 调高如设置为操作超时的数倍这正是本提案的核心收益——lookup timeout 与 operation timeout 解耦后你可以单独为查找类请求放大重试窗口而无需放大所有生产/消费操作的超时。其余改动TooManyRequests可重试、独立连接池为增强项默认配置下不改变既有行为。七、总结与实践建议PIP-91 的落地让 Apache Pulsar Java 客户端的查找链路变得更加健壮和可调理解查找请求的本质PMR 与 Lookup 无副作用、可由任意 Broker 服务理应比 produce/consume 拥有更大的重试余量善用lookupTimeout()在集群滚动重启、Broker 频繁上下线、或客户端经负载均衡器接入的场景下将其显式设置为 operation timeout 的 25 倍可显著降低查不到 Topic导致的客户端报错关注重试落点独立连接池与出错即断连的机制能促使客户端重新握手并让负载均衡器换一台后端避免重试始终命中故障节点回归风险可控默认行为完全兼容lookupTimeoutMs -1时回落至 operation timeout唯一需要留意的变化是 PMR 超时计算的 bug 修复。关于该提案的完整设计动机与讨论可阅读 pip/pip-91.md 原文对配置字段与 Backoff 实现的细节可继续深入 ClientConfigurationData.java 与 PulsarClientImpl.java 源码。赞分享消息队列流处理后端微服务消息路由【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pu/pulsar点击查看免费下载相关推荐深入解析 Apache Pulsar PIP-57Broker Zookeeper 会话超时处理机制的改进深入解析 Apache Pulsar PIP 57Broker Zookeeper 会话超时处理机制的改进 导读 本篇文章以 Apache Pulsar 仓库消息队列流处理后端微服务消息路由Apache Pulsar PIP-292 深度解读在 WebSocket 插件中强制执行 Token 过期时间Apache Pulsar PIP 292 深度解读在 WebSocket 插件中强制执行 Token 过期时间 本文围绕 Apache Pulsar 的 P消息队列流处理后端微服务消息路由Apache Pulsar 中 PIP-130让 ack 超时重投递支持 Backoff 退避策略ackTimeoutRedeliveryBackoffApache Pulsar 中 PIP 130让 ack 超时重投递支持 Backoff 退避策略ackTimeoutRedeliveryBackoff消息队列流处理后端微服务消息路由创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Apache Pulsar PIP-91 深度解读：将 Lookup 超时从 Operation 超时中分离
消息队列流处理后端微服务消息路由【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pu/pulsar点击查看免费下载导读PIP-91PIP-91: Separate lookup timeout from operation timeout是 Apache Pulsar 客户端侧的一项关键改进提案它针对客户端连接 Topic 时必经的分区元数据请求PMR与Lookup 请求在超时与重试处理上的不一致问题提出将两者所用的超时上限与操作超时operation timeout解耦、让TooManyRequests对这两类请求可重试并为二进制协议 Lookup 提供独立的连接池。读完本文你将理解 Pulsar 客户端 Topic 路由的底层机制、超时与退避backoff的相互作用以及如何在 Java 客户端中通过lookupTimeout()配置项精准控制查找行为让客户端在 Broker 滚动重启等大规模 Topic 迁移场景下更具韧性。一、背景一次 Topic 连接背后发生了什么在 Pulsar 中客户端连接一个 Topic 时并不直接向拥有该 Topic 的 Broker发消息而是先做一次lookup查找以定位当前持有该 Topic 元数据与所有权ownership的 Broker。整个过程包含两个请求分区元数据请求Partition Metadata Request简称 PMR查询该 Topic 是否为分区 Topic、有多少个分区Lookup 请求针对 Topic或其每个分区向集群询问哪个 Broker 负责该 Topic。这两个请求有一个共同特点它们可以由集群中的任意一个 Broker 服务而不像生产produce与消费consume请求那样必须路由到 Topic 的 owner Broker。因此它们的请求频率并不与系统流量成正比而是与Topic 在集群中的迁移/移动频率成正比。例如当集群发生滚动重启rolling restart时即使流过系统的消息数量保持不变所有客户端也必须重新对所有 Topic 做一次 lookup——此时查找类请求会出现瞬时的大规模爆发。从语义上看PMR 与 Lookup 都是无副作用side effect free的对同一个 Topic 的多次重复请求不会改变系统状态。正因为如此PIP-91 认为这两类请求应当比其他请求对瞬时错误transient errors具有更强的韧性——重试它们不会带来状态污染却能显著提升集群动荡期客户端的自愈能力。二、现状问题两套请求的超时与重试不一致PIP-91 指出PMR 与 Lookup 在超时与重试策略上存在不一致甚至缺陷的处理PMR 的超时重试请求超时发生在配置的操作超时operation timeout默认 30 秒之后理论上 PMR 只在操作超时窗口内重试因此不应该发生重试。但实际存在一个bug退避时间backoff time被计入超时预算而等待请求超时所耗的时间却未被计入导致 PMR 在超时后实际上仍然会重试。此外PMR明确不对TooManyRequests响应进行重试。Lookup 的超时重试同样存在重试逻辑但 Lookup 允许的最大总时长与操作超时相等导致超时后实际上没有有效的重试机会。与 PMR 相反Lookup 对TooManyRequests是有重试的。重试打在同一后端的风险当客户端通过负载均衡器loadbalancer接入 Pulsar 时所有 Lookup 都使用同一个 IP 端点。虽然请求会被均衡分发到各后端 Broker但一旦发生瞬时错误由于连接是持久persistent的重试会精确地落到刚才出问题的同一个后端无法换一台 Broker 重试。三、PIP-91 提案的三个核心改动针对上述问题PIP-91 提出三项改动新增独立的 Lookup 超时lookup timeout其值是操作超时的倍数multiple of operation timeout用于替换 PMR 与 Lookup 所用 Backoff 对象中的最大重试时长上限让TooManyRequests对 PMR 与 Lookup 都可重试使客户端在 Broker 限流/过载时能够换一台 Broker 再试让二进制协议binary protoLookup 使用独立的连接池并在请求出错时关闭对应连接可配置。关闭并重建连接可以让负载均衡器重新为客户端挑选一台新的后端从而规避重试永远落在故障节点的问题。默认行为保持不变lookup timeout 的默认值匹配 operation timeout未显式配置的客户端不会感知到行为变化。四、仓库源码中的落地实现PIP-91 提出的 lookup timeout 配置在 Apache Pulsar Java 客户端中已经完整落地以下是关键实现证据均在当前仓库中可查。4.1 配置字段lookupTimeoutMs在 ClientConfigurationData.java 中定义了lookupTimeoutMs字段默认值为-1Schema( name operationTimeoutMs, description Client operation timeout (in milliseconds). ) private long operationTimeoutMs 30000; Schema( name lookupTimeoutMs, description Client lookup timeout (in milliseconds). ) private long lookupTimeoutMs -1;读取时的逻辑体现了默认回落到 operation timeout的设计见 同文件 L572-L578public long getLookupTimeoutMs() { if (lookupTimeoutMs 0) { return lookupTimeoutMs; } else { return operationTimeoutMs; } }也就是说未配置时 lookup timeout 完全等于 operation timeout默认 30 秒与 PIP-91 提案中的默认行为不变完全一致一旦显式设置Lookup 类请求便获得独立的超时预算。4.2 Builder APIlookupTimeout(int, TimeUnit)公共 API 层在 ClientBuilder.java 中声明/** * param lookupTimeout lookup timeout * param unit time unit for {code lookupTimeout} */ ClientBuilder lookupTimeout(int lookupTimeout, TimeUnit unit);具体实现位于 ClientBuilderImpl.java将其转换为毫秒写入配置Override public ClientBuilder lookupTimeout(int lookupTimeout, TimeUnit unit) { conf.setLookupTimeoutMs(unit.toMillis(lookupTimeout)); return this; }典型用法示例PulsarClient client PulsarClient.builder() .serviceUrl(pulsar://broker.example.com:6650) // 显式拉高 lookup 超时例如 60 秒为集群滚动重启留出重试余量 .lookupTimeout(60, TimeUnit.SECONDS) // operationTimeout 保持默认 30 秒互不影响 .build();4.3 消费端使用lookup deadlinegetLookupTimeoutMs()被客户端各核心组件用来计算查找截止时间lookup deadline例如ProducerImpl.javathis.lookupDeadline System.currentTimeMillis() client.getConfiguration().getLookupTimeoutMs();ConsumerImpl.java 与之对称TransactionMetaStoreHandler.java 在定位事务元数据存储transaction meta store时同样使用该截止时间WatchTcAssignmentsDiscovery.java 在事务协调器TC分配发现流程中读取getLookupTimeoutMs()。这表明 lookup timeout 不仅作用于普通 Topic 的 PMR/Lookup也覆盖了事务协调器发现等内部查找路径。4.4 PMR 路径用 lookup timeout 作为退避上限在 PulsarClientImpl.java 的getPartitionedTopicMetadata中PMR 的重试 Backoff 直接以getLookupTimeoutMs()作为超时预算AtomicLong opTimeoutMs new AtomicLong(conf.getLookupTimeoutMs()); Backoff backoff Backoff.builder() .initialDelay(Duration.ofNanos(conf.getInitialBackoffIntervalNanos())) .mandatoryStop(Duration.ofMillis(opTimeoutMs.get() * 2)) .maxBackoff(Duration.ofNanos(conf.getMaxBackoffIntervalNanos())) .build();可以看到mandatoryStop强制终止时间由 lookup timeout 决定这正对应 PIP-91 中用 lookup timeout 替换 operation timeout 作为 Backoff 最大重试时长的提议。4.5 HTTP 查找路径对于通过http/https服务地址进行的查找HttpClient.java 也读取了 lookup timeout 并构造超时预算Duration budget当配置值不大于 0 时回落到 60 秒兜底值。4.6 关于独立连接池配置PIP-91 中提出的第三个改动二进制协议 Lookup 使用独立连接池、出错时关闭连接属于可配置项当前仓库中未发现与该配置同名的字段读者应以本仓库实际提供的客户端配置为准并关注 PIP 后续演进版本中的实现情况。五、测试验证Lookup 重试行为PIP-91 的测试计划指出该改动将做单元测试新增 mock 以便从ServerCnx轻松触发特定行为超时、TooManyRequests。仓库中的集成测试 LookupRetryTest.java 正是对这一行为的验证。该测试构造了一个模拟 Broker分别对分区元数据请求第 324 行与 Lookup 请求第 338 行主动返回ServerError.TooManyRequests例如ctx.writeAndFlush(newPartitionMetadataResponse(ServerError.TooManyRequests, too many, requestId)); ctx.writeAndFlush(newLookupErrorResponse(ServerError.TooManyRequests, too many, requestId));测试中客户端通过.lookupTimeout(10, TimeUnit.SECONDS)见测试文件第 106、260、273 行显式设置 lookup timeout并验证在连续TooManyRequests下的重试与最终失败行为第 224-238 行的注释说明了MaxNumberOfRejectedRequestPerConnection1时客户端应在连续TooManyRequests下失败、以及指数退避下 1 次与 4 次拒绝的差异。这为TooManyRequests可重试以及重试次数受退避策略控制提供了可复现的测试依据。此外ServiceUrlQuarantineTest.java 同样涉及 lookup timeout 相关配置的验证说明该配置已进入多场景的回归测试覆盖。六、兼容性与迁移注意事项PIP-91 明确指出该提案可能带来的一处行为变化PMR 超时计算的 bug 修复修复后 PMR 的整体超时时间更严格不再把退避时间以外的等待时间漏算进预算因此部分此前被 bug 掩盖的瞬时错误将重新暴露为超时异常。对于这类用户解决方案是显式把 lookup timeout 调高如设置为操作超时的数倍这正是本提案的核心收益——lookup timeout 与 operation timeout 解耦后你可以单独为查找类请求放大重试窗口而无需放大所有生产/消费操作的超时。其余改动TooManyRequests可重试、独立连接池为增强项默认配置下不改变既有行为。七、总结与实践建议PIP-91 的落地让 Apache Pulsar Java 客户端的查找链路变得更加健壮和可调理解查找请求的本质PMR 与 Lookup 无副作用、可由任意 Broker 服务理应比 produce/consume 拥有更大的重试余量善用lookupTimeout()在集群滚动重启、Broker 频繁上下线、或客户端经负载均衡器接入的场景下将其显式设置为 operation timeout 的 25 倍可显著降低查不到 Topic导致的客户端报错关注重试落点独立连接池与出错即断连的机制能促使客户端重新握手并让负载均衡器换一台后端避免重试始终命中故障节点回归风险可控默认行为完全兼容lookupTimeoutMs -1时回落至 operation timeout唯一需要留意的变化是 PMR 超时计算的 bug 修复。关于该提案的完整设计动机与讨论可阅读 pip/pip-91.md 原文对配置字段与 Backoff 实现的细节可继续深入 ClientConfigurationData.java 与 PulsarClientImpl.java 源码。赞分享消息队列流处理后端微服务消息路由【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pu/pulsar点击查看免费下载相关推荐深入解析 Apache Pulsar PIP-57Broker Zookeeper 会话超时处理机制的改进深入解析 Apache Pulsar PIP 57Broker Zookeeper 会话超时处理机制的改进 导读 本篇文章以 Apache Pulsar 仓库消息队列流处理后端微服务消息路由Apache Pulsar PIP-292 深度解读在 WebSocket 插件中强制执行 Token 过期时间Apache Pulsar PIP 292 深度解读在 WebSocket 插件中强制执行 Token 过期时间 本文围绕 Apache Pulsar 的 P消息队列流处理后端微服务消息路由Apache Pulsar 中 PIP-130让 ack 超时重投递支持 Backoff 退避策略ackTimeoutRedeliveryBackoffApache Pulsar 中 PIP 130让 ack 超时重投递支持 Backoff 退避策略ackTimeoutRedeliveryBackoff消息队列流处理后端微服务消息路由创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考