NullPointerException on partitions revoked

[Amoratinos]

I have an application deployed on a k8s cluster, while the system is under some load and the pod is scaled up I've seen a NullPointerException in the logs when the partitions are being revoked from an existing consumer

java.lang.NullPointerException: Cannot invoke "io.confluent.parallelconsumer.state.WorkContainer.getRetryDueAt()" because "workContainer" is null
    at i.c.parallelconsumer.state.ShardManager.lambda$new$0(ShardManager.java:81)
    at java.util.Comparator.lambda$comparing$77a9974f$1(Comparator.java:473)
    at java.util.Comparator.lambda$thenComparing$36697e65$1(Comparator.java:220)
    at java.util.Comparator.lambda$thenComparing$36697e65$1(Comparator.java:220)
    at java.util.TreeMap.getEntryUsingComparator(TreeMap.java:374)
    at java.util.TreeMap.getEntry(TreeMap.java:344)
    at java.util.TreeMap.remove(TreeMap.java:881)
    at java.util.TreeSet.remove(TreeSet.java:276)
    at i.c.parallelconsumer.state.ShardManager.removeWorkFromShardFor(ShardManager.java:173)
    at i.c.parallelconsumer.state.ShardManager.removeAnyShardEntriesReferencedFrom(ShardManager.java:157)
    at i.c.p.state.PartitionState.onPartitionsRemoved(PartitionState.java:545)
    at i.c.p.state.PartitionStateManager.resetOffsetMapAndRemoveWork(PartitionStateManager.java:245)
    at i.c.p.state.PartitionStateManager.onPartitionsRemoved(PartitionStateManager.java:184)
    at i.c.p.state.PartitionStateManager.onPartitionsRevoked(PartitionStateManager.java:175)
    at i.c.parallelconsumer.state.WorkManager.onPartitionsRevoked(WorkManager.java:109)
    at i.c.p.i.AbstractParallelEoSStreamProcessor.onPartitionsRevoked(AbstractParallelEoSStreamProcessor.java:421)
    ... 16 common frames omitted
Wrapped by: i.c.p.internal.InternalRuntimeException: onPartitionsRevoked event error
    at i.c.p.i.AbstractParallelEoSStreamProcessor.onPartitionsRevoked(AbstractParallelEoSStreamProcessor.java:423)
    at o.a.k.c.c.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:340)
    at o.a.k.c.c.internals.ConsumerCoordinator.onJoinPrepare(ConsumerCoordinator.java:830)
    ... 14 common frames omitted
Wrapped by: org.apache.kafka.common.KafkaException: User rebalance callback throws an error
    at o.a.k.c.c.internals.ConsumerCoordinator.onJoinPrepare(ConsumerCoordinator.java:861)
    at o.a.k.c.c.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:443)
    at o.a.k.c.c.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:385)
    at o.a.k.c.c.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:552)
    at o.a.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1276)
    at o.a.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1240)
    at o.a.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1220)
    at i.c.p.internal.ConsumerManager.poll(ConsumerManager.java:58)
    at i.c.p.internal.BrokerPollSystem.pollBrokerForRecords(BrokerPollSystem.java:205)
    at i.c.p.internal.BrokerPollSystem.handlePoll(BrokerPollSystem.java:160)
    at i.c.p.internal.BrokerPollSystem.controlLoop(BrokerPollSystem.java:136)
    at java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
    at j.u.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
    at java.lang.Thread.run(Thread.java:842)

From what I've checked in the code looks like a null WorkContainer is passed to the retryQueue TreeSet thus the exception looks like a valid one ShardManager

I'm on version 0.5.2.8.

[sangreal]

This should be race condition after checking the code, I am working on the fix

[rkolesnev]

Fix for this is merged now. @Amoratinos - can you please test on build from master?

[Amoratinos]

I can give it a try but I just saw this error once and under heavy load 🤞

[rkolesnev]

Closing as fixed - reopen if it will reoccur.