Invalid client connection when SASL token is rejected by broker

[d1egoaz]

Versions

Please specify real version numbers or git SHAs, not just "Latest" since that changes fairly regularly.
Sarama Version:
digest = "1:2ec153af6a806c3d63d4299f2549bcb29d75d9703097341be309a46db3481488"
version = "v1.22.1"

Kafka Version: 2.0.1
Go Version: 1.12

Configuration

What configuration values are you using for Sarama and Kafka?
SASL authentication

Problem Description

When creating a client connection to kafka using SASL and the token provider provides an invalid token, sarama is not returning an error, it returns a client with no brokers, and the connection is still open, so if we try to talk to kafka, i.e, Topics() you're getting an empty list, where this should raise an error as the client connection is invalid

when a token provider returns a valid token, it can get the list of brokers and connect to them succesfully:

{"level":"info","msg":"Initializing new client","test":"kafkaClient"}
{"level":"info","msg":"client/metadata fetching metadata for all topics from broker localhost:9094\n","test":"kafkaClient"}
{"level":"info","msg":"Successful SASL handshake. Available mechanisms: [OAUTHBEARER]","test":"kafkaClient"}
{"level":"info","msg":"Connected to broker at localhost:9094 (unregistered)\n","test":"kafkaClient"}
{"level":"info","msg":"client/brokers registered new broker #1 at localhost:9094","test":"kafkaClient"}
{"level":"info","msg":"Successfully initialized new client","test":"kafkaClient"}
{"level":"info","msg":"client.Closed() false"}
{"level":"info","msg":"brokers count: 1"}
{"level":"info","msg":"topic list [__consumer_offsets __identities]","test":"kafkaClient"}

when token provider returns an invalid token:

{"level":"info","msg":"Initializing new client","test":"kafkaClient"}
{"level":"info","msg":"client/metadata fetching metadata for all topics from broker localhost:9094\n","test":"kafkaClient"}
{"level":"info","msg":"Successful SASL handshake. Available mechanisms: [OAUTHBEARER]","test":"kafkaClient"}
{"level":"info","msg":"Received SASL auth response: {\"status\":\"invalid_request\", \"scope\":\"Could not verify token\", \"openid-configuration\":\"https://accounts.google.com/.well-known/openid-configuration\"}","test":"kafkaClient"}
{"level":"info","msg":"Connected to broker at localhost:9094 (unregistered)\n","test":"kafkaClient"}
{"level":"info","msg":"Successfully initialized new client","test":"kafkaClient"}
{"level":"info","msg":"client.Closed() false"}
{"level":"info","msg":"brokers count 0"}
{"level":"info","msg":"topic list []","test":"kafkaClient"}

as we can see, broker is rejecting the client authentication, and this seems it's an unrecoverable error for the client.

It seems this method reads the invalid request response, but it doesn't check the content to see if the broker rejected the user credentials/token:
https://github.com/Shopify/sarama/blob/c82066c1586636238edf9256e1ab6e410e080e49/broker.go#L1263-L1265

Not sure if the solution would be to add some checks to this method to see if the broker accepted the user's token, and somehow make the NewClient function return an error. and close the connection.

Also, not sure if the current workflow is expected, and the client needs to check something before start talking to kafka,

[mk6i]

@d1egoaz Thanks for raising this issue. I believe it's due to the Sarama client not properly closing the token exchange as described by https://tools.ietf.org/html/rfc7628#section-4.3. I will work on fixing this.

[mk6i]

@d1egoaz feel free to close the issue if the fix merged to master solves your problem.

[d1egoaz]

that's great @mkaminski1988
Please let me find some time this week to test the fix

Thanks you very much for this!

[d1egoaz]

Just tested the new version and it works great!
kafka_client_test.go:194: couldn't connect to kafka, kafka server: SASL Authentication failed.

Thanks again!