从 Java API 连接 Kafka aws 实例答案

【问题标题】：connect Kafka aws instance from Java API从 Java API 连接 Kafka aws 实例
【发布时间】：2019-11-26 11:05:24
【问题描述】：

我试图通过本地 Spring Boot API 连接 kafka aws 实例。

我可以连接它，但是在收听主题时，它会引发以下异常，但新主题是由 Spring Boot API 成功创建的我也无法发布任何消息。

java.io.IOException: Can't resolve address: ip-xxx-xx-xx-xx.ec2.internal:9092
at org.apache.kafka.common.network.Selector.doConnect(Selector.java:235) ~[kafka-clients-2.0.1.jar:na]

Caused by: java.nio.channels.UnresolvedAddressException: null
    at sun.nio.ch.Net.checkAddress(Net.java:101) ~[na:1.8.0_192]
    at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:622) ~[na:1.8.0_192]
    at org.apache.kafka.common.network.Selector.doConnect(Selector.java:233) ~[kafka-clients-2.0.1.jar:na]
    ... 30 common frames omitted

2019-07-17 15:36:13.581  WARN 3709 --- [           main] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-1, groupId=group_id] Error connecting to node ip-172-31-80-50.ec2.internal:9092 (id: 0 rack: null)

我也允许这个端口自定义 TCP 规则 TCP 2181 0.0.0.0/0 自定义 TCP 规则 TCP 9092 0.0.0.0/0

server:
  port: 8081
spring:
  kafka:
    consumer:
      bootstrap-servers: xx.xx.xx.xx:9092
      group-id: group_id
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      bootstrap-servers: xx.xx.xx.xx:9092
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer


@KafkaListener(topics = "ConsumerTest", groupId = "group_id")
    public void consume(String message) throws IOException {
        logger.info(String.format("#### -> Consumed message -> %s", message));
    }

【问题讨论】：

ping bootstrap-servers 并检查它是否启动
在 aws 实例上运行的引导服务器。我也允许这个端口 Custom TCP Rule TCP 2181 0.0.0.0/0 Custom TCP Rule TCP 9092 0.0.0.0/0 从本地 spring boot api 访问

标签： spring-boot apache-kafka spring-kafka

【解决方案1】：

java.io.IOException: 无法解析地址：ip-xxx-xx-xx-xx.ec2.internal:9092

连接到节点 ip-172-31-80-50.ec2.internal:9092 时出错

当消费者连接到代理时，他们会取回从中读取数据的分区的代理元数据。您的客户在这里得到的是 Kafka 代理的 advertised.listener。因此，当您通过代理的公共地址连接到代理时，它会将机器的内部地址返回给您的客户端。

要解决此问题，您需要在代理上正确设置侦听器。详情请见https://rmoff.net/2018/08/02/kafka-listeners-explained/。

【讨论】：