Unhandled exception in PartitionsFor in case of network instability

[FDoKE]

Version

4.3.8

Context

.onComplete method never gets called after some internal NPE.
Caused by netowork instability.

partitionInfoConsumer.partitionsFor(topicName)
                    .onComplete(ar -> {
                        ... someWork
                   })

Logs contains:

05:45:18.718 [vert.x-eventloop-thread-9] [ERROR] [io.vertx.core.impl.ContextBase] - Unhandled exception
java.lang.NullPointerException: null
	at io.vertx.kafka.client.common.impl.Helper.from(Helper.java:109)
	at io.vertx.kafka.client.consumer.impl.KafkaConsumerImpl.lambda$partitionsFor$8(KafkaConsumerImpl.java:473)
	at io.vertx.kafka.client.consumer.impl.KafkaReadStreamImpl.lambda$null$1(KafkaReadStreamImpl.java:131)
	at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:264)
	at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:246)
	at io.vertx.core.impl.EventLoopContext.lambda$runOnContext$0(EventLoopContext.java:43)
	at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:174)
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:167)
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:503)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.lang.Thread.run(Thread.java:748)

Looks like leader() contains null.

Do you have a reproducer?

No, it happened while network was unstable.

Steps to reproduce

1. Call partitionsFor
2. onComplete never gets called
3. Logs have unhandled exception (which caused hang)