MP Reactive Messaging: NullPointerException during Kafka partition rebalance

[Azquelt]

Describe the bug

During partition reassignment, it's possible for the Kafka connector for MicroProfile Reactive Messaging to not update its internal state correctly, resulting in it stopping receiving messages and a NullPointerException being logged.

The following log messages may be seen:

java.lang.NullPointerException
	at com.ibm.ws.microprofile.reactive.messaging.kafka.KafkaInput.lambda$wrapInMessageStream$10(KafkaInput.java:274)
	at com.ibm.ws.microprofile.reactive.messaging.kafka.KafkaInput$$Lambda$476/0x00000000c2be7230.test(Unknown Source)
	at io.smallrye.reactive.streams.stages.FilterStageFactory$$Lambda$481/0x00000000c3268d30.test(Unknown Source)

User provided listener com.ibm.ws.microprofile.reactive.messaging.kafka.adapter.impl.ConsumerRebalanceListenerImpl failed on invocation of onPartitionsRevoked for partitions [live-partition-test-in-0]
java.lang.NullPointerException
	at com.ibm.ws.microprofile.reactive.messaging.kafka.KafkaInput.onPartitionsRevoked(KafkaInput.java:346)
	at com.ibm.ws.microprofile.reactive.messaging.kafka.adapter.impl.ConsumerRebalanceListenerImpl.onPartitionsRevoked(ConsumerRebalanceListenerImpl.java:55)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:315)

Steps to Reproduce
This is hard to reproduce deliberately but is occurring occasionally in our automated tests. It requires that something calls KafkaConsumer.wakeup() while KafkaInput.onPartitionsAssigned() is calling KafkaConsumer.position().

Expected behavior
The partition assignment should succeed correctly and without the above errors.

Diagnostic information:

• OpenLiberty Version: all versions
• Affected feature(s): mpReactiveMessaging-1.0

Additional context

For RTC 290930