问题标签 [aiven]

我正在使用 spring cloud stream 和Aiven的模式注册表，它使用confluent 的模式注册表。Aiven 的模式注册表由密码保护。根据这些说明，需要设置这两个配置参数才能成功访问模式注册表服务器。

当我只使用vanilla java的kafka驱动时一切都很好，但是如果我使用Spring cloud stream，我不知道如何注入这两个参数。此刻，我正在把"basic.auth.user.info"和"basic.auth.credentials.source"放在文件中"spring.cloud.stream.kafka.binder.configuration"。application.yml

这样做，我"401 Unauthorized"就在模式想要注册的地方上线。

更新1：

根据 'Alin's 的建议，我更新了 SchemaRegistryClient 的 bean 的配置方式，使其能够感知 SSL 上下文。

这有助于消除应用程序启动时的错误并注册架构。然而，每当应用程序想要向 Kafka 推送消息时，就会再次抛出一个新错误。最后，这也被 mmelsen 的回答解决了。

我在 Aiven (GCP) 上的云中运行了一个托管的 kafka 集群。我正在尝试使用 Debezium SQL Server 连接器从我们的本地数据库流式传输 CDC 记录。

有人知道如何从云中的 kafka 连接到本地数据库吗？这可以通过某种 VPN 连接来完成，还是我需要通过谷歌云配置 VPC？感谢任何建议。

下面的代码片段工作正常。但有时它不是从 Kafka 的 topic 读取消息。我没有收到任何错误。在 Kafka 方面（我们使用的是 Aiven Managed Kafka），消费者组已与主题相关联，并且消费者脚本运行良好。

我需要您的指导来解决上述问题。

卡夫卡版本 - 2.0.1

节点模块版本 - “node-rdkafka”：“^2.7.0”

代码参考 - https://github.com/Blizzard/node-rdkafka

我们将 Aiven 用作 Kafka 即服务提供商，它们支持凭证轮换。例如每 24 小时。

到目前为止，我们还没有找到在需要轮换时在 Spring Kafka 中创建新消费者/生产者的方法。基本上，一个 TopicAuthorizationException 被抛出consumer.poll。

我们需要做的，基本上是创建一个具有更新信任库和密钥库路径的新消费者。密钥库和信任库是从 kubernetes 机密中加载的。

任何人都知道我们如何才能使这项工作？

我想在我的 ElasticSearch 实例 ( Aiven )中测试新的通配符字段类型。

我试过这个：

我得到了这样的回应：

以下是有关该实例的信息：

得到 /

获取 /_许可证

我的理解是这个功能是X-Pack提供的，我不知道Aiven的服务中是否包含这个功能。有什么办法可以使这项工作？

我们已经对两个 VPC 进行了对等。GCP 和第三方 VPC (Aiven)。

我们在 aiven 中很少有服务需要与 gcp vpc 中的资源进行通信，例如 elasticsearch、redis 等。

比方说，

DNS-CNAME： test-elasticsearch.example.io
ES-URL： test-elasticsearch.aivencloud.com

我们在 GCP VPC 中有一个云 DNS 区域。我为第三方服务 URL 配置了 CNAME，只能通过 GCP VPC 访问。

我成功地能够从 GCP VPC ping elasticsearch url 但是，我无法 ping CNAME。

仅供参考：dns 区域运行良好。我通过使用私有IP创建相同的记录来测试它。

在此，让我知道我可以做些什么来解决这个问题。

更新：编辑 1. 没有 CNAME 在我的云 DNS 中工作。

编辑 2. 公共子网中的资源可以解析，但不能解析私有子网资源

好心的帮助

我有 2 个服务：生产者和消费者。

据我了解，message.ts是生产者产生消息的时间（而不是kafka-broker 收到消息的时间）。

问题

1. 当消费者消费消息时，我怎么知道它在 kafka-broker 内的时间（没有网络延迟：从生产者到 kafka-broker 以及从 kafka-broker 到消费者）？

2. 我从我的消费者 vm 到 kafka 代理执行了 ping 操作。ping 结果为 0.7ms（毫秒）。从每一方到kafka broker的网络延迟是0.3ms吗？我假设 kafka 传输是TCP这样的，所以所有东西都有一个“ACK”消息。而且我假设没有“ACK”，每一方都不会做任何事情，所以我得出结论，每个大小的网络延迟与 ping 结果相同：0.7ms（毫秒）。我对么？

我下载 了aiven-kafka-connect-jdbc并手动构建了jar。将相同的jar放在kafka-connect插件目录中，并在connect-standalone.properties中提到了插件路径。但是当我运行连接时，它会抛出错误，提示“找不到任何实现连接器且名称与 io.aiven.connect.jdbc.JdbcSourceConnector 匹配的类”

我的配置中缺少什么？

连接-standalone.properties

连接器属性

kafka connect插件的文件夹结构

我有一个 Kafka 集群（在 AWS 上使用 Aivan）：

卡夫卡硬件

• 我的消费者和 Kafka Broker 之间的 Ping 是 0.7 毫秒。

后群

我有一个这样的话题：

• 它包含大约 3000 个实体的数据。
• 实体生命周期为一周。
• 每周将有不同的 3000 个实体（平均）。
• 每个实体总共可能有 15k 到 50k 条消息。
• 每秒最多可以有 500 条消息。

建筑学

我的团队构建了一个架构，这样会有一组消费者。他们将解析这些数据，执行一些转换（没有任何过滤！！），然后将最终消息发送回 kafka 到topic=<entity-id>.

这意味着我将数据上传回 kafka 到仅包含特定实体数据的主题。

问题

在任何给定时间，kafka 中最多可以有 3-4k 个主题（每个唯一实体 1 个主题）。

1. 我的卡夫卡能处理好吗？如果不是，我需要改变什么？
2. 我是否需要删除一个主题，或者随着时间的推移有（很多！！）未使用的主题是可以的？
3. 每个消费最终消息的消费者将同时消费 100 个主题。我知道 kafka 客户可以同时使用多个主题，但我不确定最佳实践是什么。
4. 请分享您的担忧。

要求

• 请关注这个架构的潜在问题，尽量不要谈论替代架构（更少的话题，更多的消费者等）。

我们目前正在将 ElasticSearch 与由 Aiven 托管的 PHP Symfony 一起使用，但 Aiven 目前不再提供 Elasticsearch。此优惠已被 OpenSearch 1.0 取代。因此，Aiven 计划最迟在 2022 年 3 月之前将所有现有的 ElasticSearch 项目迁移到 OpenSearch。

目前我们正在使用 Ruflin 的“Elastica”包。使用相同的包应该可行，但我们需要找到一个新包，因为 Elastica 将不支持 ElasticSearch 的任何分叉实例（参考：https ://github.com/ruflin/Elastica/issues/1962 ）。

但是，似乎没有一个包可以支持来自 ElasticSearch 的 OpenSearch 分叉实例。请问现在有同样情况的人吗？

非常感谢。