title: How to configure Spring Boot 2 SASL OAUTHBEARER to connect to KAFKA AZURE EVENT HUB?
body: <p>I am trying to connect a simple Spring Boot kafka Consumer to Azure Event Hub Kafka
  I must authenticate to Azure Event Hub via Azure AD before polling or consuming message from topic</p>
  <p>In my application.yaml</p>
  <pre><code>server: 
    port: 9000
  spring:
    kafka:
      bootstrap-servers: xxxx.servicebus.windows.net:9093
      properties:
        sasl.jaas.config: org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;
        sasl.mechanism: OAUTHBEARER
        security.protocol: SASL_SSL
        request.timeout.ms: 60000
        **sasl.login.callback.handler.class**: com.xxx.config.CustomAuthenticateCallbackHandler
      consumer:
        group-id: compass
        properties:
          spring.json:
            use.type.headers: false
            value.default.type: com.xxx.consumerMessage
        auto-offset-reset: earliest
        key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
        value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  topic:
    name: global.xxx
  </code></pre>
  <p>I added this class to implement Kafka custom OAUTHBEARER to get and validate Token from Azure AD :</p>
  <pre><code>package com.xxxx..consumer.config;
  //Copyright (c) Microsoft Corporation. All rights reserved.
  //Licensed under the MIT License.
  import java.io.IOException;
  import java.net.MalformedURLException;
  import java.net.URI;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.Executors;
  import java.util.concurrent.ScheduledExecutorService;
  import java.util.concurrent.TimeoutException;
  
  import javax.security.auth.callback.Callback;
  import javax.security.auth.callback.UnsupportedCallbackException;
  import javax.security.auth.login.AppConfigurationEntry;
  
  import org.apache.kafka.clients.producer.ProducerConfig;
  import org.apache.kafka.common.KafkaException;
  import org.apache.kafka.common.security.auth.AuthenticateCallbackHandler;
  import org.apache.kafka.common.security.oauthbearer.OAuthBearerToken;
  import org.apache.kafka.common.security.oauthbearer.OAuthBearerTokenCallback;
  
  import com.microsoft.aad.msal4j.ClientCredentialFactory;
  import com.microsoft.aad.msal4j.ClientCredentialParameters;
  import com.microsoft.aad.msal4j.ConfidentialClientApplication;
  import com.microsoft.aad.msal4j.IAuthenticationResult;
  import com.microsoft.aad.msal4j.IClientCredential;
  
  public class CustomAuthenticateCallbackHandler implements AuthenticateCallbackHandler {
  
      final static ScheduledExecutorService EXECUTOR_SERVICE = Executors.newScheduledThreadPool(1);
  
      private String authority;
      private String appId;
      private String appSecret;
      private ConfidentialClientApplication aadClient;
      private ClientCredentialParameters aadParameters;
  
      @Override
      public void configure(Map&lt;String, ?&gt; configs, String mechanism, List&lt;AppConfigurationEntry&gt; jaasConfigEntries) {
          String bootstrapServer = Arrays.asList(configs.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG)).get(0).toString();
          bootstrapServer = bootstrapServer.replaceAll(&quot;\\[|\\]&quot;, &quot;&quot;);
          URI uri = URI.create(&quot;https://&quot; + bootstrapServer);
          String sbUri = uri.getScheme() + &quot;://&quot; + uri.getHost();
          this.aadParameters =
                  ClientCredentialParameters.builder(Collections.singleton(sbUri + &quot;/.default&quot;))
                  .build();
  
          this.authority = &quot;https://login.microsoftonline.com/xxxxxx/&quot;; // replace &lt;tenant-id&gt; with your tenant id
          this.appId = &quot;xxxxxx&quot;; // also called client id
          this.appSecret = &quot;xxxxx&quot;; // also called client secret
      }
  
      public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
          for (Callback callback: callbacks) {
              if (callback instanceof OAuthBearerTokenCallback) {
                  try {
                      OAuthBearerToken token = getOAuthBearerToken();
                      OAuthBearerTokenCallback oauthCallback = (OAuthBearerTokenCallback) callback;
                      oauthCallback.token(token);
                  } catch (InterruptedException | ExecutionException | TimeoutException e) {
                      e.printStackTrace();
                  }
              } else {
                  throw new UnsupportedCallbackException(callback);
              }
          }
      }
  
      OAuthBearerToken getOAuthBearerToken() throws MalformedURLException, InterruptedException, ExecutionException, TimeoutException
      {
          if (this.aadClient == null) {
              synchronized(this) {
                  if (this.aadClient == null) {
                      IClientCredential credential = ClientCredentialFactory.createFromSecret(this.appSecret);
                      this.aadClient = ConfidentialClientApplication.builder(this.appId, credential)
                              .authority(this.authority)
                              .build();
                  }
              }
          }
  
          IAuthenticationResult authResult = this.aadClient.acquireToken(this.aadParameters).get();
          System.out.println(&quot;TOKEN ACQUIRED&quot;);
  
          return new OAuthBearerTokenImp(authResult.accessToken(), authResult.expiresOnDate());
      }
  
      public void close() throws KafkaException {
          // NOOP
      }
  }
  </code></pre>
  <p>My doubts are :</p>
  <ul>
  <li><p>Is this the best solution to do that with Spring , should I write by myself the class which implements  AuthenticateCallbackHandler</p>
  </li>
  <li><p>If I change the **client Id ** constant to wrong value in <strong>CustomAuthenticateCallbackHandler.Java</strong>  , in the console an exception is added :</p>
  <p>at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  at java.base/java.lang.reflect.Method.invoke(Method.java:566)
  at org.springframework.boot.devtools.restart.RestartLauncher.run(RestartLauncher.java:49)
  Caused by: com.microsoft.aad.msal4j.MsalServiceException: AADSTS700016: Application with identifier 'xxxxxx00f832e67A' was not found in the directory 'e4e1abd9-eac7-4a71-ab52-da5c998aa7ba'. This can happen if the application has not been installed by the administrator of the tenant or consented to by any user in the tenant. You may have sent your authentication request to the wrong tenant.</p>
  </li>
  </ul>
  <p>But after that, the console shows: Successfully logged in, is it normal to not authenticate and get Polling?</p>
  <pre><code>2021-07-07 15:00:07.924  INFO 16108 --- [  restartedMain] .o.i.e.ExpiringCredentialRefreshingLogin : ***Successfully logged in.***
  2021-07-07 15:00:07.974  WARN 16108 --- [  restartedMain] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'spring.json.value.default.type' was supplied but isn't a known config.
  2021-07-07 15:00:07.975  WARN 16108 --- [  restartedMain] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'spring.json.use.type.headers' was supplied but isn't a known config.
  2021-07-07 15:00:07.976  INFO 16108 --- [  restartedMain] o.a.kafka.common.utils.AppInfoParser     : Kafka version: 2.7.1
  2021-07-07 15:00:07.976  INFO 16108 --- [  restartedMain] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId: xxx
  2021-07-07 15:00:07.977  INFO 16108 --- [  restartedMain] o.a.kafka.common.utils.AppInfoParser     : Kafka startTimeMs: 1625662807975
  2021-07-07 15:00:07.978  INFO 16108 --- [  restartedMain] o.a.k.clients.consumer.KafkaConsumer     : [Consumer clientId=consumer-xxxx-1, groupId=xxx] Subscribed to topic(s): global.finance.copa
  2021-07-07 15:00:07.996  INFO 16108 --- [  restartedMain] c.l.c.consumer.KafkaOnAzureApplication   : Started KafkaOnAzureApplication in 1.703 seconds (JVM running for 2.555)
  2021-07-07 15:00:08.178  WARN 16108 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-xxx-1, groupId=xxxx] Connection to node -1 (xxx.servicebus.windows.net/xxxxx:9093) could not be established. 
  </code></pre>
  <p>The SASL OUATHBEARER:<br />
  <a href="https://i.stack.imgur.com/VJCBw.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/VJCBw.png" alt="The SASL OUATHBEARER" /></a></p>