问题标签 [aws-msk]

您好，我目前正在尝试通过融合连接器将数据从 kafka(MSK)(AWS) 发送到 elasticsearch。我有一个来自 sql 数据库的数据流。当我运行 confluent 连接器时，它会运行一段时间将数据发送到 elasticsearch，但会在一两分钟后停止，并出现以下错误。

sending LeaveGroup request to coordinator b-2.*****.amazonaws.com:9092 due to consumer poll timeout has expired. This means the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time processing messages. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.

有很多数据需要流式传输，并且想知道是否有更好的方法或特定设置来一次发送多个批次，或者修复此错误的方法可能会有所帮助。我的弹性搜索服务器太小了吗？

任何其他信息请告诉我谢谢。

I have a Kafka cluster with:

• 2 brokers b-1 and b-2.
• 2 topics with both: PartitionCount:1 ReplicationFactor:2 min.insync.replicas=1

Here is what happened:

Within the code, I got this error when my producer performed a poll around that time:

Broker b-2 logs have this:

My understanding here is that (1) b-2 went down i.e. unable to connect to Zookeeper (2) Messages were produced to b-1 successfully during this time. (3) b-1 was also trying to forward messages to b-2during this downtime due to the replication factor set to 2 (4) All these forwarded messages (ProduceRequests) got timed-out after 600s

My question:

1. Is my understanding correct and how I can prevent this from happening again?
2. If I had 3 brokers here, would b-1 have tried to connect to b-3 right away rather than waiting for b-2? Is that a good workaround? (Assuming topic replication factor = 2 everywhere)

我们创建了一个新的 AWS MSK (kafka) 集群

我们创建了一个新配置并分配给该集群：

使用 python 脚本，我们想发送一条新消息：

它不起作用并说：

如果我们在发送消息之前创建主题，它可以正常工作。

我有三个 MSK 集群；开发、非产品和产品。它们都有以下集群配置——没有主题级别的配置。

Dev 和 Nonprod 正在清除设置中定义的超过 100 小时的消息log.retention.hours=100。

我们有更多的流量通过我们的生产集群，旧消息没有被删除。集群上仍有数十万条超过 400 小时的消息。我考虑过添加更多的配置设置，例如

为了更快地滚动段，因为段可能还没有滚动并且不能标记为删除 - 但是这个相同的配置在其他集群中运行良好，尽管没有接收到那么多的流量。

我正在尝试将我的 python 消费者连接到 AWS MSK 集群。我怎样才能做到这一点？

让 AWS MSK 集群运行我正在尝试使用 python 和 kafka python 使用来自 MSK 集群的消息。

我得到的错误

AWS MSK 的官方文档没有明确指出正在使用哪种类型的存储卷，所以我想知道它是哪一种：

• SSD（通用？）
• 硬盘？

相关链接：

• https://aws.amazon.com/msk/faqs/
• https://docs.aws.amazon.com/msk/latest/developerguide/what-is-msk.html

我们有一个使用 MSK（kafka 服务器）消息的 lambda 函数。它工作没有问题。但有时它会无缘无故停止工作（没有日志），只显示：

问题：Lambda 内部错误。请联系 Lambda 客户支持。

我们几乎记录了每一行代码（lambda函数），没有错误。

我们没有付费支持服务。但我认为如果它与内部 AWS 错误有关，他们必须提供免费支持。无论如何，我们能解决它吗？

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

我已经阅读了有关连接到 MSK 的方法的 AWS指南，但不确定哪种方法适合这个确切的用例。我尝试了几种方法-

1. 将 Advertisementd.listerners 设置为实例的公共 IP。但是在创建 KafkaConsumer 后列出主题时仍然会出现此错误- Connection to node -1 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)

2. 在安全组入站规则中打开相关端口（在我的情况下为 PLAINTEXT 为 9092）。Connection refused执行此操作后尝试 telnet 到端口 9092 时仍然出现错误。

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

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

我正在使用 Kafka 连接从 MSK 集群获取消息，并按照https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-kafka-connector-msk/将它们放在 Kinesis 消防软管上

我的设置工作正常，但是当我尝试通过系统发送大量消息时，我收到警告

卡夫卡版本是 2.6.0

我对 Kafka 还很陌生，所以不确定解决这个问题的最佳方法是什么。我曾考虑过增加 kafka 连接工作程序超时，但这似乎不太正确。

我需要将 Talend for Big Data 7.x 与 AWS MSK Kafka 一起使用

我看到了 Apache Kafka 的组件，但没有看到 AWS

https://help.talend.com/r/ZKOgsQ1JCBAK1GzyokKlYQ/Fju2qsByhmN~2LawpQxWoQ

不确定我是否可以将其与 AWS 一起使用

任何意见表示赞赏