I can use confluence cli to connect to the confluence kafka cluster, but I cannot use segmentio's kafka-go library. I get the following error.
with SASL: SASL handshake failed: EOF
This is my function in 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)) } }
I tried generating new keys, using my account username and password, reducing partitions, but nothing worked.
It seems that your server's TLS
version is not accepted, you can use MinVersion
to force go -kafka accept it:
dialer := &kafka.Dialer{ Timeout: 10 * time.Second, DualStack: true, SASLMechanism: mechanism, TLS: &tls.Config{ MinVersion: tls.VersionTLS12, }, }
The above is the detailed content of Unable to connect to Confluence Kafka using segmentio's kafka-go. For more information, please follow other related articles on the PHP Chinese website!