KAFKA-14212: Enhanced HttpAccessTokenRetriever to retrieve error mess…

[smjn]

https://issues.apache.org/jira/browse/KAFKA-14212

• Currently HttpAccessTokenRetriever client side class does not retrieve error response from the token e/p. As a result, seemingly trivial config issues could take a lot of time to diagnose and fix. For example, client could be sending invalid client secret, id or scope.
• This PR aims to remedy the situation by retrieving the error response, if present and logging as well as appending to any exceptions thrown.
• New unit tests have also been added.

Sample Output

The o/p is generated using bin/kafka-console-producer

Invalid clientId

.
.
[2022-09-16 14:51:45,769] WARN handleOutput - error response code: 401, response body: null, error response body: {"errorCode":"invalid_client","errorSummary":"Invalid value for 'client_id' parameter.","errorLink":"invalid_client","errorId":"oaecGwvnlkhTce-YfueXeZVkQ","errorCauses":[]} (org.apache.kafka.common.security.oauthbearer.secured.HttpAccessTokenRetriever)
org.apache.kafka.common.KafkaException: Failed to construct kafka producer
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:462)
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:290)
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:317)
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:302)
	at kafka.tools.ConsoleProducer$.main(ConsoleProducer.scala:44)
	at kafka.tools.ConsoleProducer.main(ConsoleProducer.scala)
Caused by: org.apache.kafka.common.KafkaException: javax.security.auth.login.LoginException: org.apache.kafka.common.KafkaException: org.apache.kafka.common.security.oauthbearer.secured.UnretryableException: java.io.IOException: The response code 401 and error response {"invalid_client" - "Invalid value for 'client_id' parameter."} was encountered reading the token endpoint response; will not attempt further retries
.
.

Invalid client_secret

.
.
[2022-09-16 14:53:02,990] WARN handleOutput - error response code: 401, response body: null, error response body: {"error":"invalid_client","error_description":"The client secret supplied for a confidential client is invalid."} (org.apache.kafka.common.security.oauthbearer.secured.HttpAccessTokenRetriever)
org.apache.kafka.common.KafkaException: Failed to construct kafka producer
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:462)
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:290)
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:317)
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:302)
	at kafka.tools.ConsoleProducer$.main(ConsoleProducer.scala:44)
	at kafka.tools.ConsoleProducer.main(ConsoleProducer.scala)
Caused by: org.apache.kafka.common.KafkaException: javax.security.auth.login.LoginException: org.apache.kafka.common.KafkaException: org.apache.kafka.common.security.oauthbearer.secured.UnretryableException: java.io.IOException: The response code 401 and error response {"invalid_client" - "The client secret supplied for a confidential client is invalid."} was encountered reading the token endpoint response; will not attempt further retries
.
.

Invalid scope

.
.
[2022-09-16 14:54:05,536] WARN handleOutput - error response code: 400, response body: null, error response body: {"error":"invalid_scope","error_description":"One or more scopes are not configured for the authorization server resource."} (org.apache.kafka.common.security.oauthbearer.secured.HttpAccessTokenRetriever)
org.apache.kafka.common.KafkaException: Failed to construct kafka producer
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:462)
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:290)
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:317)
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:302)
	at kafka.tools.ConsoleProducer$.main(ConsoleProducer.scala:44)
	at kafka.tools.ConsoleProducer.main(ConsoleProducer.scala)
Caused by: org.apache.kafka.common.KafkaException: javax.security.auth.login.LoginException: org.apache.kafka.common.KafkaException: org.apache.kafka.common.security.oauthbearer.secured.UnretryableException: java.io.IOException: The response code 400 and error response {"invalid_scope" - "One or more scopes are not configured for the authorization server resource."} was encountered reading the token endpoint response; will not attempt further retries
.
.

Missing scope

.
.
[2022-09-16 14:54:47,630] WARN handleOutput - error response code: 400, response body: null, error response body: {"error":"invalid_scope","error_description":"The authorization server resource does not have any configured default scopes, 'scope' must be provided."} (org.apache.kafka.common.security.oauthbearer.secured.HttpAccessTokenRetriever)
org.apache.kafka.common.KafkaException: Failed to construct kafka producer
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:462)
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:290)
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:317)
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:302)
	at kafka.tools.ConsoleProducer$.main(ConsoleProducer.scala:44)
	at kafka.tools.ConsoleProducer.main(ConsoleProducer.scala)
Caused by: org.apache.kafka.common.KafkaException: javax.security.auth.login.LoginException: org.apache.kafka.common.KafkaException: org.apache.kafka.common.security.oauthbearer.secured.UnretryableException: java.io.IOException: The response code 400 and error response {"invalid_scope" - "The authorization server resource does not have any configured default scopes, 'scope' must be provided."} was encountered reading the token endpoint response; will not attempt further retries
	at org.apache.kafka.common.security.oauthbearer.secured.HttpAccessTokenRetriever.retrieve(HttpAccessTokenRetriever.java:180)
.
.

[omkreddy]

LGTM, Thanks for the PR

[kirktrue]

Nice work, @smjn! Thanks!