如何使用 KafkaConsumer java SDK 使用来自 AWS MSK 集群的事件表单？答案

【问题标题】：How to consume events form from AWS MSK cluster using KafkaConsumer java SDK?如何使用 KafkaConsumer java SDK 使用来自 AWS MSK 集群的事件表单？
【发布时间】：2021-06-08 00:51:17
【问题描述】：

我在 AWS MSK 中创建了一个集群并创建了一个主题。我可以在与集群相同的 VPC 中生成和使用来自 EC2 实例的事件。具体用例是我想使用 Java 来消费事件。 KafkaClientAWS SDK 没有任何事件消费的方法，所以我这里不能用。在创建KafkaConsumer 时，我无法直接在 Properties() 中传递引导服务器，因为 IP 位于 VPC 内部。

我已经通过这个 AWS guide 了解连接到 MSK 的方法，但不确定哪种方法适合这个确切的用例。我尝试了几种方法-

将 Advertisementd.listerners 设置为实例的公共 IP。但是在创建 KafkaConsumer 后列出主题时仍然会出现此错误- Connection to node -1 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
在安全组入站规则中打开相关端口（在我的情况下为 PLAINTEXT 的 9092）。执行此操作后尝试 telnet 到端口 9092 时，我仍然收到 Connection refused 错误。

我的问题是上述任何一项工作，我只是在这里遗漏了其他可以解决这个问题的东西。还是以上方法都行不通？

我的最终目标是使用 Java 使用来自 MSK 集群的事件。

【问题讨论】：

你的java进程是运行在同一个VPC还是外部？如果在外面，您是否希望您的 MSK 集群对所有人公开可用？
@RishabhSharma 它需要从外部运行（即不是来自同一个 VPC）。是的，我希望它可以公开（即，如果传递了正确的 client.properties，则应该从任何地方访问它）。为此，我想我们需要让引导服务器的 IP 可公开访问，因为它们目前对 VPC 是私有的。

标签： java apache-kafka kafka-consumer-api aws-msk amazon-msk

【解决方案1】：

由于您需要公开 MSK Endpoints ips，因此您需要改进身份验证过程并将 MSK 集群放置在具有 Internet 网关和正确配置路由表的 VPC 中的公共子网上： https://docs.aws.amazon.com/vpc/latest/userguide/VPC_Route_Tables.html

如果您在自己的 VPC 中连接它，即使是公开的，您也可以使用 CLI 从 MSK 集群中检索引导服务器，并使用 EC2 引导中的用户数据将其作为环境变量注入： https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/user-data.html

使用 CLI 从 MSK 集群检索引导服务器的命令：

aws kafka get-bootstrap-brokers --region us-east-1 --cluster-arn ClusterArn

更多关于 MSK 集群配置的文档： https://docs.aws.amazon.com/msk/latest/developerguide/produce-consume.html

【讨论】：