handle credential unavailable

Author: spike83

README.md

@@ -33,6 +33,7 @@ sasl.jaas.config= \
 		clientId='ignored' \
 		clientSecret='ignored' \
 		useWorkloadIdentity='true' \
+        scope='${CONFLUENT_CLOUD_APP_ID}/.default' \
 		extension_logicalCluster='lkc-xxxxxx' \
 		extension_identityPoolId='pool-xxxx';
 ```
@@ -43,7 +44,7 @@ Use with Schema Registry
 
 example:
 ```
-echo '{"make": "Ford", "model": "Mustang", "price": 10000}' |kafka-avro-console-producer \
+echo '{"make": "Ford", "model": "Mustang", "price": 10000}' | kafka-avro-console-producer \
   --bootstrap-server <bootstrap>.confluent.cloud:9092 \
   --property schema.registry.url=https://<registry>.confluent.cloud \
   --property bearer.auth.credentials.source='CUSTOM' \

src/main/java/io/confluent/oauth/HttpAccessTokenRetriever.java

@@ -173,28 +173,29 @@ public HttpAccessTokenRetriever(String clientId,
 
     @Override
     public String retrieve() throws IOException {
-        // TODO: should we use the AZURE_FEDERATED_TOKEN_FILE variable to enable workload identity? And AZURE_AUTHORITY_HOST to use for the endpoint url?
-        if (this.useWorkloadIdentity){
-            log.debug("using workload identity to get token");
-            // AccessToken https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/core/azure-core/src/main/java/com/azure/core/credential/AccessToken.java
-            TokenCredential workloadIdentityCredential = WorkloadIdentityUtils.createWorkloadIdentityCredentialFromEnvironment();
-            TokenRequestContext tokenRequestContext = WorkloadIdentityUtils.createTokenRequestContextFromEnvironment(scope);
-            AccessToken azureIdentityAccessToken = workloadIdentityCredential.getTokenSync(tokenRequestContext);
-            log.trace("useWorkloadIdentity token, got token from AzureAD: '{}'", azureIdentityAccessToken.getToken());
-            return azureIdentityAccessToken.getToken();
-        }
+        String responseBody;
 
-        String authorizationHeader = formatAuthorizationHeader(clientId, clientSecret);
-        String requestBody = requestMethod == "GET" ? null : formatRequestBody(scope);
-        Retry<String> retry = new Retry<>(loginRetryBackoffMs, loginRetryBackoffMaxMs);
+        try {
+            // TODO: should we use the AZURE_FEDERATED_TOKEN_FILE variable to enable workload identity? And AZURE_AUTHORITY_HOST to use for the endpoint url?
+            if (this.useWorkloadIdentity){
+                log.debug("using workload identity to get token");
+                // AccessToken https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/core/azure-core/src/main/java/com/azure/core/credential/AccessToken.java
+                TokenCredential workloadIdentityCredential = WorkloadIdentityUtils.createWorkloadIdentityCredentialFromEnvironment();
+                TokenRequestContext tokenRequestContext = WorkloadIdentityUtils.createTokenRequestContextFromEnvironment(scope);
+                AccessToken azureIdentityAccessToken = workloadIdentityCredential.getTokenSync(tokenRequestContext);
+                log.trace("useWorkloadIdentity token, got token from AzureAD: '{}'", azureIdentityAccessToken.getToken());
+                return azureIdentityAccessToken.getToken();
+            }
 
-        final Map<String, String> requestHeaders = new HashMap<>(headers);
-        requestHeaders.put(AUTHORIZATION_HEADER, authorizationHeader);
 
+            String authorizationHeader = formatAuthorizationHeader(clientId, clientSecret);
+            String requestBody = requestMethod == "GET" ? null : formatRequestBody(scope);
+            Retry<String> retry = new Retry<>(loginRetryBackoffMs, loginRetryBackoffMaxMs);
+
+            final Map<String, String> requestHeaders = new HashMap<>(headers);
+            requestHeaders.put(AUTHORIZATION_HEADER, authorizationHeader);
 
-        String responseBody;
 
-        try {
             responseBody = retry.execute(() -> {
                 HttpURLConnection con = null;
 
@@ -218,6 +219,9 @@ public String retrieve() throws IOException {
                 throw (IOException) e.getCause();
             else
                 throw new KafkaException(e.getCause());
+        } catch (CredentialUnavailableException ex){
+            log.error("Error getting token from AzureAD: {}", ex.getMessage());
+            throw new KafkaException(ex);
         }
 
         return parseAccessToken(responseBody);