When parallel consumer does not close kafka consumer if commmit fails during close

When AbstractParallelEoSStreamProcessor.close(Duration timeout) is being executed it performs commitOffsetsThatAreReady() - if this operation fails then maybeCloseConsumer() is not executed - so consumer is not closed.
Not closing consumer means that it will stay in consumer group for max.poll.interval.ms if it is not executing polls - it can also prevent from joining consumer group by other consumers.

In my case issue was like this:
1. Kafka removed and added ACL for consumer user (not sure if it was terraform)
2. Poolsystem failed.
3. My application found out that parallel processor is in failed state (isClosedOrFailed()) and tried to close it and create it again.
4. Close failed on commit (due to timeout - not sure why), consumer was not closed but was not performing polls
5. New parallel consumer could not join the group for 5 minutes (until old consumer was not removed from the group in max.poll.interval.ms)

Of course as I am providing kafka consumer to parallel listener I can close it myself - which I will do if close of parallel consumer fail but I think that it should be done by parallel consumer.