使用segmentio kafka-go连接Confluent Kafka失败，报SASL握手EOF错误

核心问题分析 #

Confluent Cloud集群强制要求SASL_SSL协议连接，你的代码仅配置了SASL PLAIN身份验证机制，但未启用SSL加密，导致连接被集群直接拒绝，触发EOF错误。而Confluent CLI默认内置了SSL配置，因此可以正常连接。

修复步骤 #

• 添加SSL/TLS配置：在kafka.Dialer中显式启用SSL加密，Confluent Cloud的证书受公共CA信任，无需额外配置自定义证书。
• 校验Broker地址：确保brokerAddress是控制台提供的完整地址，格式为xxx.confluent.cloud:9092，不要遗漏端口号。

修改后的完整代码 #

package consumer

import (
    "context"
    "crypto/tls"
    "fmt"
    "log"
    "os"
    "time"

    "github.com/segmentio/kafka-go"
    "github.com/segmentio/kafka-go/sasl/plain"
)

func Consume(ctx context.Context) {
    l := log.New(os.Stdout, "kafka reader: ", 0)
    mechanism := plain.Mechanism{
        Username: "my-api-key",   // 替换为你的Confluent API Key
        Password: "my-api-secret", // 替换为你的Confluent API Secret
    }

    dialer := &kafka.Dialer{
        Timeout:       10 * time.Second,
        DualStack:     true,
        SASLMechanism: mechanism,
        // 启用SSL加密，使用默认TLS配置即可
        TLS: &tls.Config{
            InsecureSkipVerify: false,
        },
    }

    r := kafka.NewReader(kafka.ReaderConfig{
        Brokers: []string{"xxx.confluent.cloud:9092"}, // 替换为完整Broker地址
        Topic:   "steps",
        Logger:  l,
        Dialer:  dialer,
    })
    defer r.Close() // 新增资源释放逻辑

    for {
        msg, err := r.ReadMessage(ctx)
        if err != nil {
            log.Printf("读取消息失败: %v", err)
            continue // 替换panic为优雅的错误处理
        }
        fmt.Println("收到消息: ", string(msg.Value))
    }
}

额外检查项 #

• 确认API Key/Secret属于目标集群，不要混淆不同环境的密钥
• 检查防火墙是否允许出站访问9092端口
• 坚持使用API Key/Secret进行身份验证，Confluent Cloud不支持账户用户名密码直接连接

内容的提问来源于stack exchange，提问作者Meet