问题标签 [apache-pulsar]

我知道在 Kafka 中，消费者从代理主题中提取消息（拉取）？

receive考虑到方法阻塞，我觉得 Pulsar 的工作方式相同。但我找不到确认。有人可以指点我参考或纠正我吗？

谢谢

我有一个围绕代理设计的系统，这样我的生产者使用 Java，消费者使用 Go。

我正在使用 apache-pulsar 作为我的经纪人

Java - 生产者

MessageJava 类在发送到 pulsar 之前转换为字节数组：MessageJava 类的对象调用同一类中定义的getBytes()方法将其转换为 byte[]，然后将该数组发送到 apache-pulsar

我的消费者是用 Go 编写的。

去 - 消费者

从 pulsar 读取字节数组并使用 ConvertAfterReceiving 方法 [定义如下] 转换为 MessageGo 结构，我使用gob进行解码：

问题是我无法解码 byte[] 并将其转换为 MessageGo 结构。它显示错误编码的无符号整数超出范围

我尝试将 MessageJava.entityId 更改为 short/long 并将 MessageGo.EntityId 更改为 int8/int16/int32/int64/uint8/uint16/uint32/uint64 [所有排列]，但都是徒劳的。

//转换导入org.apache.flink.api.scala._

我只是写了这些代码，但是 flink 代理总是抛出一个错误： 原因：org.apache.pulsar.client.admin.PulsarAdminException$NotAuthorizedException: HTTP 401 Unauthorized

我在 Apache Pulsar 消费者中有一个用例。有一个生产者正在向不同的两个集群（集群 A 和集群 B）以及一次具有相同数据的相应主题生成消息（JSON 是生成的消息） .

这里 A 是主集群，B 就像辅助集群。

我正在尝试什么

我正在尝试使用数据，然后将其处理为 CSV 文件，然后每天处理表。应该根据集群的可用性来完成。

问题：

当 Cluster Primary 不可用和/或不向 Primary(A) 轮询消息时，我需要切换到 Secondary(B)。但是如果我从 Secondary(B) 消费，那么我将从 start 获取数据，因为没有确认集群之间共享。因此，我们将遇到数据冗余问题。

假设如果我同时启动并且只存储主要数据和次要数据，则通过等待一定的时间间隔仅考虑主要的故障转移。在这种情况下可能有数据丢失的可能性，因为我正在等待一定的间隔时间。

他们是否有任何建议或想法来解决我的问题，例如打开故障转移并存储到 CVS 和没有数据依赖性和数据丢失的表。

我开发了两个消费者代码来消费数据，一个用于 A，另一个用于 B。

Apache Pulsar 支持的最大消息大小是多少？

我尝试阅读文档。但我找不到任何相关信息。

我浏览了 Apache Pulsar Documentation for Message Delivery Semantics。Apache函数提到的交付语义（至少一次，最多一次和一次有效），如果我们不使用Apache函数，那么有哪些不同的交付语义可用？

我有一个 Pulsar 主题，我通过多个生产者向该主题发布消息。这些生产者只是 goroutines，我正在使用SendAsync生产者上的方法发送消息，但我不断收到Got unexpected send receipt for message警告。我查看了客户端的库代码（Pulsar 的 golang 客户端），发现只有当生产者未在此 map 中注册为侦听器时，才会显示此警告map[uint64]ConnectionListener。但是，当生产者未注册为侦听器时，我如何发布？

我的情况是我在 Pulsar 中压缩了主题（https://pulsar.apache.org/docs/en/concepts-topic-compaction/）。与生产者一起，我正在发送带有密钥的消息。是否可以通过与消费者一起使用密钥来获取特定消息？

我的代码：

我一直在研究 Apache Pulsar 的功能，我很高兴。但是，我刚刚进入身份验证/授权，并在 Java 管理客户端遇到了一个奇怪的错误。

虽然我能够执行各种操作，例如创建/删除租户、创建/删除命名空间、创建/删除分区主题、向租户添加管理员角色等，但在尝试使用

org.apache.pulsar.client.admin.Namespaces.grantPermissionOnNamespace()

我收到此错误：

org.apache.pulsar.client.admin.PulsarAdminException$ServerSideErrorException: HTTP 501 Not Implemented

我不敢相信这么重要的功能不会被实现，所以我猜这个错误只是掩盖了其他东西。

我创建了一个具有新命名空间的新租户（没有任何管理员角色）。我试图添加的角色只是虚构的，我不确定它是否必须先在某个地方注册（但我没有找到任何可以这样做的东西）。我尝试分别并同时授予消费或生产操作，但没有任何效果。

代理上没有配置超级用户，但考虑到我可以在没有它的情况下执行所有其他操作，只有超级用户才能做到这一点似乎有些牵强。目前，所有呼叫都未经过身份验证。这会是个问题吗？但这会引发同样的问题：为什么我可以做其他所有事情，只是不授予权限？

那么我错过了什么？

我们正在运行 Pulsar 2.5.0 版以及 Java pulsar-client-admin Maven 工件 2.5.0 版。

您如何使用 Akka Streams 持续使用来自 Apache Pulsar 的消息并打印每条消息？

下面是我从pulsar4s库中找到的示例代码。您如何打印消费的消息，而不是将消息发布到另一个主题？