问题标签 [aws-msk]

我无法将架构注册表连接到 MSK。我在一个单独的实例中运行 Docker，并启用了与 MSK 服务的连接。

尝试了来自 confluent docker.io/confluentinc/cp-schema-registry:latest 的 docker 镜像。

与 SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL 连接时遇到的错误如下

还尝试将 SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL 替换为 SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS

仍然失败并出现以下错误

请对此提出任何解决方案。

我们有一个 MSK 集群设置。我知道引导服务器名称看起来像b-3.abcd-dev-kafka-dev.lq2z7z.c4.kafka.eu-central-1.amazonaws.com（名称已更改）。

这些服务器应该位于我的子网中，因此我希望它们只能在我的内部网络中解析。dig <name> 1.1.1.1但是，当向 cloudflares DNS 服务器发出请求时，DNS 名称似乎也会解析。这是 AWS MSK 集群的正常行为，还是我需要修复的某个地方存在 DNS 泄漏？

我有一个 Spring Boot 应用程序，它会向Kafka (AWS MSK). 从我的本地机器上，我可以将消息发送到MSK同一个应用程序无法从AWS Fargate.

我正在使用 AWS MSK，并且我想启用 ACL，但在打开 ACL 时我无法创建主题。我正在使用命令行工具进行所有操作。这是我正在做的事情的摘要：

• 创建一个新的集群
• 创建一个主题 - 这很好用
• 在 resource=CLUSTER 和 operation=ALL 上为 client1 开启 ACL
• 使用 AdminClient 创建主题（通过提供 --bootstrap-server 选项） - 这会产生超时异常
• 重新尝试创建相同的主题 - 这给出了一个错误，说主题已经存在
• 使用 AdminClient 列出主题 - 这不返回任何主题
• 使用 Zookeeper 连接创建主题 - 这有效
• 使用 Zookeeper 连接列出主题 - 这将返回我创建的所有主题（即使是那些超时的主题）

所以问题是主题是在 Zookeeper 上创建的，但代理无法访问它。大概是由于我缺少一些 ACL 规则。

我运行的命令的原始输出：

再次运行相同的命令：

通过 AdminClient 的主题列表：

通过 Zookeeper 连接的主题列表：

这是我的 ACL 规则：

我错过了什么？

使用包构建 KafkaProducer 时出现此错误kafka-python：

代码如下：

我正在使用 Python 3.7 和 AWS MSK 集群。

我正在使用 ECS 集群为我们的 MSK Kafka 集群构建在 Amazon 中托管的冗余模式注册表。

SchemaRegistry TaskDefinition 需要定义一个主机名，该主机名在运行时对每个任务都是唯一的。

注意：当通过 cli 直接在 EC2 实例中运行 docker 容器时，使用 $Hostname 有效，因为 shell 替换为唯一的完全限定主机名；但我很难弄清楚如何在 ECS 和 CloudFormation 中进行这项工作。

对于那些使用过亚马逊的 MSK（Kafka 托管流）的人的快速问题：

文档/常见问题解答清楚地表明您可以轻松扩展 MSK 集群中的代理数量。您可以轻松地将数字缩小吗？我不是在谈论自动缩放，而是手动缩放。

所以我是 Kafka 和 MSK 的新手，但我试图在 AWS 云上探索更少的东西服务器。我的用例 AWS DMS-->KAFKA(MSK)-->ElastciSearch/DynamoDB 和 S3。我希望使用 lambda 函数来做到这一点。但最近我开始知道端到端不能或难以适应 lambda 函数。

那么有什么方法可以使用带有 lambda 函数的 Kafka 连接来实现这一点，这样就不需要 EC2 了吗？

基本上我正在寻找以无服务器方式连接的 Kafka。

Kinesis 被排除在此选项之外，不在此范围内。

请向 Kafka 提出建议

我在 aws 上运行 MSK，我可以从 MSK 发送记录。我只是想使用 Kafka 连接，以便进入 MSK 的记录将转到 Elastic Search 。我做了以下事情，但我不确定我的连接器是否正常工作，因为我无法在 Elastic Search 中看到记录 这是我要发送的记录

我已经像这样安装了Kafka连接

并更改 url 和主题名称

之后我刚刚开始像下面这样的kafka连接

这给了我输出INFO Usage: ConnectStandalone worker.properties connector1.properties [connector2.properties ...] (org.apache.kafka.connect.cli.ConnectStandalone:62)

所以我没有提到我的经纪人详细信息和 Zookeper 详细信息，那么我将如何连接到 MSK。

请帮助我理解这一点。我错过了什么。我没有架构转换，所以我没有修改schema-registry.properties。

当我尝试使用以下命令时，出现以下错误

我必须将记录从 Aurora/Mysql 发送到 MSK，然后从那里发送到 Elastic 搜索服务

Aurora -->Kafka-connect--->AWS MSK--->kafka connect --->弹性搜索

Aurora 表结构中的记录是这样的，
我认为记录会以这种格式进入 AWS MSK。

所以为了通过弹性搜索消费，我需要使用正确的模式，所以我必须使用模式注册表。

我的问题

问题 1

对于上述类型的消息模式注册表，我应该如何使用模式注册表？我是否必须为此创建 JSON 结构，如果是，我将其保留在哪里。这里需要更多帮助才能理解这一点？

我已编辑

提到了 zookeper，但我没有什么是kafkastore.topic=_schema 如何将其链接到自定义架构。

即使我开始并得到这个错误

这是我所期待的，因为我没有对 schema 做任何事情。

我确实安装了 jdbc 连接器，当我启动时出现以下错误

问题 2 我可以在一个 ec2 上创建两个连接器（jdbc 和弹性 serach 一个）。如果是，我必须在单独的 cli 中同时启动吗？

问题 3 当我打开 vim /usr/local/confluent/etc/kafka-connect-jdbc/source-quickstart-sqlite.properties 我只看到如下属性值

在上面的属性文件中我可以提到架构名称和表名称？

根据答案，我正在更新我的 Kafka 连接 JDBC 配置

---------------启动 JDBC 连接弹性搜索 -----------------

接着

然后我修改了下面的属性

最后我修改

在这里我修改了以下属性

当我列出主题时，我没有看到为表名列出的任何主题。

错误消息的堆栈跟踪