无法使用segmentio的kafka-go连接到Confluence Kafka

我可以使用confluence cli 连接到confluence kafka 集群，但无法使用segmentio 的kafka-go 库。 我收到以下错误。

with SASL: SASL handshake failed: EOF

这是我在 go 中的函数

package consumer

import (
    "context"
    "fmt"
    "log"
    "os"
    "time"

    "github.com/segmentio/kafka-go"
    "github.com/segmentio/kafka-go/sasl/plain"
)
func Consume(ctx context.Context) {
    // create a new logger that outputs to stdout
    // and has the `kafka reader` prefix
    l := log.New(os.Stdout, "kafka reader: ", 0)
    mechanism := plain.Mechanism{
        Username: "my-api-key",
        Password: "my-api-secret",
    }

    dialer := &kafka.Dialer{
        Timeout:       10 * time.Second,
        DualStack:     true,
        SASLMechanism: mechanism,
    }

    r := kafka.NewReader(kafka.ReaderConfig{
        Brokers: []string{brokerAddress}, // brokerAddress given in confluent cloud cluster settings. 
        Topic:   []string{"steps"}[0],
        // assign the logger to the reader
        Logger: l,
        Dialer: dialer,
    })
    for {
        // the `ReadMessage` method blocks until we receive the next event
        msg, err := r.ReadMessage(ctx)
        if err != nil {
            panic("could not read message " + err.Error())
        }
        // after receiving the message, log its value
        fmt.Println("received: ", string(msg.Value))
    }
}

我尝试生成新密钥、使用我的帐户用户名和密码、减少分区，但没有任何效果。

知元AI

AI智能语音聊天 对讲问答 AI绘画 AI写作 AI创作助手工具

下载

正确答案

看来您的服务器的 TLS 版本不被接受，您可以使用 MinVersion 强制 go-kafka 接受它：

dialer := &kafka.Dialer{
        Timeout:       10 * time.Second,
        DualStack:     true,
        SASLMechanism: mechanism,
        TLS: &tls.Config{
            MinVersion: tls.VersionTLS12,
        },
    }