如何在 Amazon MSK 上运行 Kafka Camel 连接器答案

【问题标题】：How To Run Kafka Camel Connectors On Amazon MSK如何在 Amazon MSK 上运行 Kafka Camel 连接器
【发布时间】：2021-06-10 17:35:49
【问题描述】：

背景： 我点击了这个链接on setting up AWS MSK and testing a producer and consumer，它的设置和工作正常。我能够通过 2 个单独的 EC2 实例发送和接收消息，它们都使用相同的 Kafka 集群（我的 MSK 集群）。现在，我想建立一个从 Eventhubs 到 AWS Firehose 的数据管道，其形式如下：

Azure Eventhub -> Eventhub-to-Kafka Camel 连接器 -> AWS MSK -> Kafka-to-Kinesis-Firehose Camel 连接器 -> AWS Kinesis Firehose

我能够在不使用 MSK 的情况下（通过常规的旧 Kafka）成功地做到这一点，但由于未说明的原因，现在需要使用 MSK，但我无法让它工作。

问题： 在尝试启动 AWS MSK 和我正在使用的两个 Camel 连接器之间的连接器时，我收到以下错误：

这些是有问题的两个连接器：

目标：让这些连接器与 MSK 一起使用，就像他们在直接使用 Kafka 时所做的那样。

这是 Firehose 的问题：

Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:177)
com.amazonaws.services.kinesisfirehose.model.AmazonKinesisFirehoseException: The security token included in the request is invalid

这是 Azure 的一个：

[2021-05-04 14:09:56,848] WARN Load balancing for event processor failed - If you are using a StorageSharedKeyCredential, and the server returned an error message that says 'Signature did not match', you can compare the string to sign with the one generated by the SDK. To log the string to sign, pass in the context key value pair 'Azure-Storage-Log-String-To-Sign': true to the appropriate method call.
If you are using a SAS token, and the server returned an error message that says 'Signature did not match', you can compare the string to sign with the one generated by the SDK. To log the string to sign, pass in the context key value pair 'Azure-Storage-Log-String-To-Sign': true to the appropriate generateSas method call.
Please remember to disable 'Azure-Storage-Log-String-To-Sign' before going to production as this string can potentially contain PII.
Status code 403, "<?xml version="1.0" encoding="utf-8"?><Error><Code>AuthorizationFailure</Code><Message>This request is not authorized to perform this operation.
Time:2021-05-04T14:09:56.7148317Z</Message></Error>" (com.azure.messaging.eventhubs.PartitionBasedLoadBalancer:344)
[2021-05-04 14:09:56,858] Error was received while reading the incoming data. The connection will be closed. (reactor.netty.channel.ChannelOperationsHandler:319)
java.lang.NoSuchMethodError: org.apache.camel.component.azure.eventhubs.EventHubsConsumer.createExchange(Z)Lorg/apache/camel/Exchange;
        at org.apache.camel.component.azure.eventhubs.EventHubsConsumer.createAzureEventHubExchange(EventHubsConsumer.java:93)

【问题讨论】：

您好，请粘贴实际的错误代码而不是其图像。它提供了更好的可读性和搜索选项
@Ftisiot，添加了错误。感谢您的提醒。

标签： amazon-web-services apache-kafka apache-camel azure-eventhub amazon-msk

【解决方案1】：

MSK 不提供 Kafka Connect 服务。您需要将其安装在您自己的计算机或其他 AWS 计算资源上。从那里，您需要安装 Camel 连接器插件

【讨论】：

谢谢。所以我感到困惑的是，我在文档中的第 6 步 #3 中使用 --bootstrap-server BootstrapBrokerStringTls，在本地运行连接分发，然后尝试运行 Camel Producer 和 Consumer Connector 插件。但是在运行 Camel 连接器时，我不断收到错误消息。这是使用 Amazon MSK 运行 Camel 连接器插件的正确方法吗？没有太多文档可以提供帮助。
Connect 本身不使用--bootstrap-server，所以我不确定您指的是什么，但是将 MSK 地址和任何相关证书放在 connect-distributed.properties 文件中是所有需要的。然后为 Camel JAR 文件更新plugin.path。关于问题中的错误，似乎是一些版本问题，或者您缺少一些其他 JAR 文件，但我从未使用过 Camel，所以不能说
所以我找到了这个，这将有助于在 MSK 上启动 Kafka Connect。 amazonmsk-labs.workshop.aws/en/securityencryption/tlsmauth/… 但随后出现了连接器无法正常工作并引发错误的问题。 stackoverflow.com/questions/58511736/… aws.amazon.com/premiumsupport/knowledge-center/… 的建议均无效
一般建议不要在broker上运行Connect。在任何情况下，您遇到的错误都是 Java/插件特有的，根本不是 Kafka 连接问题
您好，有什么故障排除建议吗？它看起来是与 Kafka 连接器相关的特定于 Java 的。我会为连接器使用另一个 EC2 系统吗？

【解决方案2】：

Kafka Connect 是一个适用于 Kafka（MSK、开源或任何其他 kafka 发行版）的框架。但是，它不附带任何连接器。 Kafka Connect 与开源 kafka 捆绑在一起。

作为最佳实践，切勿在与代理节点相同的服务器上运行 kafka connect。因为它们共享二进制文件。调整代理可能会导致 kafka 代理出现意外问题。此外，Kafka Connect 应用程序是应用程序，您不会在同一节点上运行您的 kafka 消费者或生产者应用程序。所以创建一个 EC2 实例并在那里部署 kafka 连接。

来到 TLS - 如果您启用客户端 TLS 身份验证 - 您需要查找 boostrap_broker_tls。

【讨论】：