3

我正在使用 github.com/Shopify/sarama 包与 Kafka 进行交互。在我目前的方法中，我可以连接到代理并毫无问题地获取所有主题名称（下面的消费者代码）。

但是，当我尝试使用管理客户端（下面的管理代码）删除一些主题时，我收到“dial tcp:lookup ip-xx-xx.ec2.internal:no such host”错误。

我不知道为什么会收到此错误。我将不胜感激任何提示或可能的解决方案。

消费者


    config := sarama.NewConfig()
    config.Consumer.Return.Errors = true

    //get broker
    cluster, err := sarama.NewConsumer("localhost:9092", config)
    if err != nil {
        panic(err)
    }

    defer func() {
        if err := cluster.Close(); err != nil {
            panic(err)
        }
    }()

    //get all topic from cluster
    topics, _ := cluster.Topics()

行政


    config := sarama.NewConfig()
    config.Consumer.Return.Errors = true

    config.Version = sarama.V2_4_0_0

    //admin broker
    admin, err := sarama.NewClusterAdmin("localhost:9092", config)
    if err != nil {
        panic(err)
    }


    defer func() {
        if err := admin.Close(); err != nil {
            panic(err)
        }
    }()

    topic := []string{"test-topic"}
    output := admin.DeleteTopic(topic)

    if output == nil {
        fmt.Printf(" delete -  %s\n", topic[0])
    } else {
        fmt.Println(output)
    }

注意我通过转发 ssh 端口将堡垒实例连接到远程。

更新

设置后sarama.Logger = log.New(os.Stdout, "[sarama] ", log.LstdFlags)我得到以下信息：

[sarama] 2020/03/25 02:08:03 Initializing new client
[sarama] 2020/03/25 02:08:03 client/metadata fetching metadata for all topics from broker localhost:9092
[sarama] 2020/03/25 02:08:04 Connected to broker at localhost:9092 (unregistered)
[sarama] 2020/03/25 02:08:04 client/brokers registered new broker #1001 at ip-x-x-x-1.ec2.internal:9092
[sarama] 2020/03/25 02:08:04 client/brokers registered new broker #1003 at ip-x-x-x-2.ec2.internal:9092
[sarama] 2020/03/25 02:08:04 client/brokers registered new broker #1002 at ip-x-x-x-3.ec2.internal:9092
[sarama] 2020/03/25 02:08:04 Successfully initialized new client
[sarama] 2020/03/25 02:08:04 Failed to connect to broker ip-x-x-x-3.ec2.internal:9092: dial tcp: lookup ip-x-x-x-3.ec2.internal: no such host`

更新 2

我的卡夫卡 server.properties：

advertised.listeners=INTERNAL://ip-x-x-x-1.ec2.internal:9091,EXTERNAL_INSECURE://ip-x-x-x-1.ec2.internal:9092
listeners=INTERNAL://:9091,EXTERNAL_INSECURE://:9092
listener.security.protocol.map=INTERNAL:SSL,EXTERNAL_INSECURE:PLAINTEXT
4

2 回答 2

3

当客户端连接到代理（在您的情况下localhost:9092）时，代理会提供集群中所有其他代理的客户端详细信息。您可以在日志中看到这一点：

初始连接

fetching metadata for all topics from broker localhost:9092

经纪人详情：

registered new broker #1001 at ip-x-x-x-1.ec2.internal:9092
registered new broker #1003 at ip-x-x-x-2.ec2.internal:9092
registered new broker #1002 at ip-x-x-x-3.ec2.internal:9092

您遇到的问题是您的客户端随后将使用这些代理详细信息与集群进行进一步通信。代理提供的这些地址称为广告侦听器。即经纪人“宣传”的听众。

这意味着您的客户端必须能够解析并连接到代理在初始连接时返回的侦听器的主机和端口。

破解您的/etc/hosts工作的原因是您的本地客户端可以将这些地址解析回本地主机，然后 SSH 转发工作。但这只是一个黑客。

您应该advertised.listeners在代理配置中将您的地址设置为客户端可以解析的地址（无需任何客户端/etc/hosts黑客）。

要详细了解此内容，请参阅https://rmoff.net/2018/08/02/kafka-listeners-explained/

编辑：为了清楚起见，您应该将advertised.listeners每个代理设置为您的客户端可以解析的地址 - 因此，如果这是通过 localhost SSH 转发，请设置advertised.listeners为localhost:9092.

于 2020-03-25T09:49:50.080 回答
1

因此，将代理地址添加到/etc/hosts我执行代码的本地计算机上的已知主机就可以了。

/etc/主机：

127.0.0.1 ip-x-x-x-1.ec2.internal
127.0.0.1 ip-x-x-x-2.ec2.internal
127.0.0.1 ip-x-x-x-3.ec2.internal

虽然，我仍然不明白为什么我在使用sarama.NewConsumer().

于 2020-03-25T02:15:24.257 回答