The auth docs are wrong - causing service unavailability issues

[bgavrilMS]

This is a bug in your guidance

See https://github.com/Azure/azure-event-hubs-spark/blob/master/docs/use-aad-authentication-to-connect-eventhubs.md

Repro steps

1. Use service principals with Azure (with secret or with cert)
2. Ensure AuthBySecretCallBack gets called many times (which seems to be the case)

Actual: app.acquireToken always fetches the token from the token issuer service (AAD), over HTTP. While this is normally fast, P50 is ~1 second, any availability issues and spikes in latency destabilize the app.

Expected: app object should be reused, so as to serve tokens from the memory cache.

Impact: Azure Active Directory (AAD) has been receiving incidents related to availability and latency of products using this guidance. By not using cached tokens, apps are frail and have single point of failure in the identity provider.

Root cause

The acquireToken method creates a ConfidentialClientApplication object app each time and requests a token. Once the token is retrieved, the token is cached, but the app object goes out scope. Creating the ConfidentialClientApplication everytime negates the use of cached tokens. A token is available for at least 1h, usually for 24h.

Suggested fix

(pseudocode, sorry I don't know Scala)

  
 static Map<string, ConfidentialClientApplication> apps;

 override def acquireToken(audience: String, authority: String, state: Any): CompletableFuture[String] = try {
 
 ConfidentialClientApplication app =  apps[authority];

   if (app == null) {
     app = ConfidentialClientApplication
         .builder(clientId, ClientCredentialFactory.createFromSecret(this.clientSecret))
         .authority("https://login.microsoftonline.com/" + authority) 
         .build
     apps.Add(authority, app);
   }


    val parameters = ClientCredentialParameters.builder(Collections.singleton(audience + ".default")).build

    app.acquireToken(parameters).thenApply((result: IAuthenticationResult) => result.accessToken())
  } catch {
    case e: Exception =>
      val failed = new CompletableFuture[String]
      failed.completeExceptionally(e)
      failed
  }