[BUG] Event Hub Producer client fails to refresh auth token

[srnagar]

A long-running Event Hub producer client fails to refresh auth token after a few minutes of running and stops sending events with the following exception.

Exception:

Exception in thread "amqp://event-processor-interop.servicebus.windows.net/interop-test-tokenManager" reactor.core.Exceptions$BubblingException: java.lang.NullPointerException
	at reactor.core.Exceptions.bubble(Exceptions.java:154)
	at reactor.core.publisher.Operators.onErrorDropped(Operators.java:521)
	at reactor.core.publisher.MonoFlatMap$FlatMapInner.onError(MonoFlatMap.java:247)
	at reactor.core.publisher.Operators$MonoSubscriber.onError(Operators.java:1633)
	at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreInner.onError(MonoIgnoreThen.java:235)
	at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:153)
	at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1592)
	at reactor.core.publisher.MonoProcessor.subscribe(MonoProcessor.java:383)
	at reactor.core.publisher.Mono.subscribe(Mono.java:4087)
	at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:172)
	at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56)
	at reactor.core.publisher.Mono.subscribe(Mono.java:4087)
	at reactor.core.publisher.FluxFlatMap.trySubscribeScalarMap(FluxFlatMap.java:199)
	at reactor.core.publisher.MonoFlatMap.subscribeOrReturn(MonoFlatMap.java:53)
	at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:48)
	at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:150)
	at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1592)
	at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:147)
	at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56)
	at reactor.core.publisher.Mono.subscribe(Mono.java:4087)
	at reactor.core.publisher.Mono.subscribeWith(Mono.java:4193)
	at reactor.core.publisher.Mono.subscribe(Mono.java:4059)
	at reactor.core.publisher.Mono.subscribe(Mono.java:3995)
	at reactor.core.publisher.Mono.subscribe(Mono.java:3967)
	at com.azure.core.amqp.implementation.ActiveClientTokenManager$RefreshAuthorizationToken.run(ActiveClientTokenManager.java:124)
	at java.base/java.util.TimerThread.mainLoop(Timer.java:556)
	at java.base/java.util.TimerThread.run(Timer.java:506)
Caused by: java.lang.NullPointerException
	at com.azure.core.amqp.implementation.ActiveClientTokenManager$RefreshAuthorizationToken.lambda$run$1(ActiveClientTokenManager.java:146)
	at reactor.core.publisher.LambdaMonoSubscriber.doError(LambdaMonoSubscriber.java:152)
	at reactor.core.publisher.LambdaMonoSubscriber.onError(LambdaMonoSubscriber.java:147)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onError(FluxMapFuseable.java:134)
	at reactor.core.publisher.MonoFlatMap$FlatMapMain.secondError(MonoFlatMap.java:185)
	at reactor.core.publisher.MonoFlatMap$FlatMapInner.onError(MonoFlatMap.java:251)
	at reactor.core.publisher.Operators$MonoSubscriber.onError(Operators.java:1633)
	at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreInner.onError(MonoIgnoreThen.java:235)
	at reactor.core.publisher.MonoFlatMap$FlatMapMain.secondError(MonoFlatMap.java:185)
	at reactor.core.publisher.MonoFlatMap$FlatMapInner.onError(MonoFlatMap.java:251)
	at reactor.core.publisher.Operators$MonoSubscriber.onError(Operators.java:1633)
	at reactor.core.publisher.MonoIgnoreThen$ThenAcceptInner.onError(MonoIgnoreThen.java:306)
	at reactor.core.publisher.MonoCreate$DefaultMonoSink.error(MonoCreate.java:183)
	at reactor.core.publisher.MonoCreate.subscribe(MonoCreate.java:60)
	at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:153)
	at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56)
	at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:150)
	... 21 more

Code sample:

   public static void main(String args[]) throws InterruptedException {
        EventHubProducerAsyncClient producer = new EventHubClientBuilder()
            .connectionString(CONNECTION_STRING)
            .buildAsyncProducerClient();
        System.out.println("Creating batch and sending events");
        producer.createBatch()
            .flatMap(batch -> {
                batch.tryAdd(new EventData("Test event"));
                return producer.send(batch);
            }).repeat(() -> true)
//            .take(Duration.ofMinutes(300))
            .subscribe(unused -> {
                    System.out.println("Message sent");
                },
                error -> System.out.println("Error sending event " + error.getMessage()),
                () -> System.out.println("Complete sending"));
        TimeUnit.MINUTES.sleep(300);
        producer.close();
        System.out.println("Closing application");
    }