问题标签 [confluent-cloud]

我们正在使用一个非常普通的 Confluent Cloud 实例进行内部测试。因为这是基于云的，所以它们会为您提供有关当月进行的数据量的统计数据。不幸的是，没有详细的统计数据——只有进入它们的实例的字节数、从它们的实例中流出的字节数和存储。我们已经传输了大约 2MB 存储在那里的数据，但是我们的传输量非常大，每天大约 4GB。我们没有很多消费者，而且它们都是最新的——似乎没有任何奇怪的事情发生，任何消费者从偏移量 0 或类似的地方重复查询。我的问题是：这是典型的行为吗？是因为投票吗？或者是其他东西？

感谢@riferrei 的评论。我很抱歉造成混乱。为了尝试帮助澄清，请看一下这张图片：

这就是我所得到的。我的解释是，在 3 月份，我们存储了至少 390 KB 的数据，但不多（390 KB = 1024 * 1024 * 0.2766 GB-小时 / 31 天 / 24 小时）。我们传输了 2MB (0.0021 GB)，根据账单，我们传输了 138 GB 的数据，或每天大约 4 GB。我试图了解这怎么可能发生。

我正在使用 Confluent 托管的 Kafka 集群和 Schema Registry 服务。我可以管理连接融合云 kafka 集群，将以下属性添加到生产者配置（Scala）

但无法连接到融合云模式注册表。汇合云模式注册表提供访问密钥和秘密，但我不知道如何设置密钥和秘密。是否有任何用于访问融合云模式注册表的配置设置。

说明页面显示为：-

1. 安装 Confluent Cloud CLI 运行此命令以安装 Confluent Cloud CLI。curl -L --http1.1 https://cnfl.io/ccloud-cli | sh -s -- -b /usr/local/bin

它没有获得许可

“安装：无法创建常规文件'/usr/local/bin/ccloud'：权限被拒绝”

即使我尝试过

同样的错误。我该如何安装它？

我有一些代码，我的 Asp.net Core Web API 使用这些代码将某些事件记录到在 Confluent Cloud 中运行的 Kafka 服务器。当我在本地机器上运行 API 服务器时，它可以很好地使用 Kafka 发送和接收，但是当它在 Azure 应用服务上运行时，我收到“本地：消息超时”错误。我可以修改 Azure App Service 网络以使 Kafka 网络流量正确流动吗？

这是下面的代码片段：

来自文档： https ://docs.confluent.io/current/cloud/metrics-api.html

这个间隔格式是什么？-05:00我从来不知道日期时间延长的时间格式2019-12-19T11:00:00？这是什么？有人可以澄清吗？

提前致谢。

使用spring-cloud-stream-binder-kafka. 可能有人可以看到什么是错的？

当我使用https://www.confluent.io/blog/schema-registry-avro-in-spring-boot-application-tutorial/中的示例时， 它工作正常，我可以在 Confluent Cloud 上看到消息

但是，当使用配置添加相同的连接详细信息时spring-cloud-stream-binder-kafka，它会返回未经授权的错误。

我的下面的配置给出了上述错误。不知道出了什么问题？

Confluent 中运行良好的示例：

我有点问题

我在 NestJS 中使用 Kafka 和 CQRS。

我的问题是：在写入过程中处理错误的最佳方法是什么？

1. 我将请求发送到 web api 到我的 nestjs 应用程序

2. 我有并且事件ObjectedCreatedEvent在创建后在事件总线上发送并将其写入kafka（confluent.cloud）。

3. Kafka 响应错误 87，因为针对 json 模式的消息验证失败（到目前为止还可以）

4. 如何正确响应发生错误的 web api？

刚刚在此过程中启动了另一个侦听器？我希望 kafka（融合云）至少必须针对此类事情进行讨论。

我只会有一个事件ObjectValdiationFailedEvent并将其放到事件总线上。

关于这个主题有很多问题，但是，这不是重复的问题！

我面临的问题是我尝试使用 Java 14 和 Kafka 2.5.0 设置 SpringBoot 项目，而我的Consumer 返回一个空的记录列表。这里的大多数答案表明一些被遗忘的属性，经常轮询或将偏移模式设置为 early。

我看不出与docs.confluent.io有任何逻辑差异，尽管我的配置设置看起来不合常规（请参阅下面片段中的jaas.conf设置）。

然而，这有效。我没有收到任何异常（Kafka 或其他），并且建立了连接。

这是我实际投票的地方：

我添加了打印件和其他杂物以尝试找出问题。我在调试模式下运行我的程序并将断点放在这一行：

结果，正如人们所期望的那样，我看到了每秒计数的打印行。但由于我的消费者没有返回任何记录，它永远不会进入forEach，因此永远不会触发我的断点。

我绝对可以在云中看到我的主题，有两个分区。消息源源不断地生成，所以我知道我应该能够收到一些东西。

我知道连接到集群需要一些时间，但是当前时间设置为一刻钟，我至少应该收到一些东西，对吧？作为替代方案，我尝试切换consumer.subscribe()到consumer.assign()指定我的 TopicPartition 的方法，将使用者设置为consumer.seekToBeginning(). 它运行良好，但也没有返回任何内容。

在最常见的示例中没有发现的另一件事是我使用自己的类。因此KafkaConsumer<String, String>，我根据本教程实现了自定义（反）序列化程序，而不是。

会不会是我的配置设置？轮询超时有问题吗？（反）序列化，还是完全其他的？我真的无法确定为什么我得到零记录的任何原因。任何反馈将不胜感激！

我在尝试从一个融合云集群复制到另一个时遇到问题。我一直在关注如何执行此操作的融合文档，但是我遇到了一致的失败，我不得不假设这是由我的配置中的错误引起的。

配置如下：

连接器启动，但不断记录以下错误：

谷歌搜索并没有真正发现任何似乎适用于这种情况的东西。祈祷这里有人至少可以为我指明正确的方向。

谢谢，-瑞安

我正在使用融合云，目前我想防止在我的 kafka 主题中摄取具有错误模式的消息。为此，我正在使用融合的模式管理功能，我想验证消息是否与指定的模式不匹配，它应该被拒绝。我可以在本地融合，但在云上看不到拒绝。

1. 云当前是否支持模式验证？
2. 如果没有，我们是否必须通过 kafka 流实现它来自己验证模式？
3. 以上答案是否参考了他们的官方文档？或者任何指导此场景用于融合云（非本地）的文档将不胜感激。

提前致谢