Consumer closes with an exception NoSuchElementException #419

Use case:
Poller is configured and subscribed for 2 topics.
After polling if message needs further processing then push to separate topic.
Polling for messages 60>k messages per hour.

Version: 0.5.2.2

Configuration:
Order by Partitions and max concurrency is 12.

Consumer Configuration:

props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

During the high load of messages > 60k per hour I have seen that the consumer closes down and there are following exceptions seen

I think the PartitionState.getOffsetHighestSequentialSucceeded() method implementation can be changed to follow as the current implementation is not thread safe and have possibility of failing with NoSuchElementException.

Current Implementation:

public long getOffsetHighestSequentialSucceeded() {
 if (this.incompleteOffsets.isEmpty()) {
            return this.offsetHighestSeen;
        } else {
            return this.incompleteOffsets.first() - 1;
        }
}

Possible New Implementation:

public long getOffsetHighestSequentialSucceeded() {
        Long offset = this.incompleteOffsets.ceiling(0l);
        return offset==null ? offsetHighestSeen : offset-1;
}

If the implementation map is empty this will return null and exception can be avoided.

Exceptions:

"message":"Unknown error","logger_name":"io.confluent.parallelconsumer.internal.BrokerPollSystem","thread_name":"pc-broker-poll","level":"ERROR","level_value":40000,"stack_trace":"java.util.NoSuchElementException: null\n\tat java.base/java.util.concurrent.ConcurrentSkipListMap.firstKey(Unknown Source)\n\tat java.base/java.util.concurrent.ConcurrentSkipListSet.first(Unknown Source)\n\tat io.confluent.parallelconsumer.state.PartitionState.getOffsetHighestSequentialSucceeded(PartitionState.java:267)\n\tat io.confluent.parallelconsumer.state.PartitionState.getNextExpectedPolledOffset(PartitionState.java:235)\n\tat io.confluent.parallelconsumer.state.PartitionState.createOffsetAndMetadata(PartitionState.java:225)\n\tat io.confluent.parallelconsumer.state.PartitionState.getCommitDataIfDirty(PartitionState.java:218)\n\tat io.confluent.parallelconsumer.state.PartitionStateManager.collectDirtyCommitData(PartitionStateManager.java:373)\n\tat io.confluent.parallelconsumer.state.WorkManager.collectCommitDataForDirtyPartitions(WorkManager.java:198)\n\tat io.confluent.parallelconsumer.internal.AbstractOffsetCommitter.retrieveOffsetsAndCommit(AbstractOffsetCommitter.java:31)\n\tat io.confluent.parallelconsumer.internal.ConsumerOffsetCommitter.maybeDoCommit(ConsumerOffsetCommitter.java:179)\n\tat java.base/java.util.Optional.ifPresent(Unknown Source)\n\tat io.confluent.parallelconsumer.internal.BrokerPollSystem.maybeDoCommit(BrokerPollSystem.java:323)\n\tat io.confluent.parallelconsumer.internal.BrokerPollSystem.controlLoop(BrokerPollSystem.java:118)\n\tat java.base/java.util.concurrent.FutureTask.run(Unknown Source)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)\n\tat java.base/java.lang.Thread.run(Unknown Source)\n

Error from poll control thread, will attempt controlled shutdown, then rethrow. Error: Error in BrokerPollSystem system.","logger_name":"io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor","thread_name":"pc-control","level":"ERROR","level_value":40000,"stack_trace":"io.confluent.parallelconsumer.internal.InternalRuntimeError: Error in BrokerPollSystem system.\n\tat io.confluent.parallelconsumer.internal.BrokerPollSystem.supervise(BrokerPollSystem.java:100)\n\tat io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor.controlLoop(AbstractParallelEoSStreamProcessor.java:703)\n\tat io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor.lambda$supervisorLoop$5(AbstractParallelEoSStreamProcessor.java:639)\n\tat java.base/java.util.concurrent.FutureTask.run(Unknown Source)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)\n\tat java.base/java.lang.Thread.run(Unknown Source)\nCaused by: java.util.concurrent.ExecutionException: java.util.NoSuchElementException\n\tat java.base/java.util.concurrent.FutureTask.report(Unknown Source)\n\tat java.base/java.util.concurrent.FutureTask.get(Unknown Source)\n\tat io.confluent.parallelconsumer.internal.BrokerPollSystem.supervise(BrokerPollSystem.java:98)\n\t... 6 common frames omitted\nCaused by: java.util.NoSuchElementException: null\n\tat java.base/java.util.concurrent.ConcurrentSkipListMap.firstKey(Unknown Source)\n\tat java.base/java.util.concurrent.ConcurrentSkipListSet.first(Unknown Source)\n\tat io.confluent.parallelconsumer.state.PartitionState.getOffsetHighestSequentialSucceeded(PartitionState.java:267)\n\tat io.confluent.parallelconsumer.state.PartitionState.getNextExpectedPolledOffset(PartitionState.java:235)\n\tat io.confluent.parallelconsumer.state.PartitionState.createOffsetAndMetadata(PartitionState.java:225)\n\tat io.confluent.parallelconsumer.state.PartitionState.getCommitDataIfDirty(PartitionState.java:218)\n\tat io.confluent.parallelconsumer.state.PartitionStateManager.collectDirtyCommitData(PartitionStateManager.java:373)\n\tat io.confluent.parallelconsumer.state.WorkManager.collectCommitDataForDirtyPartitions(WorkManager.java:198)\n\tat io.confluent.parallelconsumer.internal.AbstractOffsetCommitter.retrieveOffsetsAndCommit(AbstractOffsetCommitter.java:31)\n\tat io.confluent.parallelconsumer.internal.ConsumerOffsetCommitter.maybeDoCommit(ConsumerOffsetCommitter.java:179)\n\tat java.base/java.util.Optional.ifPresent(Unknown Source)\n\tat io.confluent.parallelconsumer.internal.BrokerPollSystem.maybeDoCommit(BrokerPollSystem.java:323)\n\tat io.confluent.parallelconsumer.internal.BrokerPollSystem.controlLoop(BrokerPollSystem.java:118)\n\t... 4 common frames omitted\n","dd.service":"uc3-connector","dd.version":"0.0.58-4"}

message":"Execution or timeout exception waiting for broker poller thread to finish","logger_name":"io.confluent.parallelconsumer.internal.BrokerPollSystem","thread_name":"pc-control","level":"ERROR","level_value":40000,"stack_trace":"java.util.concurrent.ExecutionException: java.util.NoSuchElementException\n\tat java.base/java.util.concurrent.FutureTask.report(Unknown Source)\n\tat java.base/java.util.concurrent.FutureTask.get(Unknown Source)\n\tat io.confluent.parallelconsumer.internal.BrokerPollSystem.closeAndWait(BrokerPollSystem.java:244)\n\tat io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor.doClose(AbstractParallelEoSStreamProcessor.java:515)\n\tat io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor.lambda$supervisorLoop$5(AbstractParallelEoSStreamProcessor.java:642)\n\tat java.base/java.util.concurrent.FutureTask.run(Unknown Source)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)\n\tat java.base/java.lang.Thread.run(Unknown Source)\nCaused by: java.util.NoSuchElementException: null\n\tat java.base/java.util.concurrent.ConcurrentSkipListMap.firstKey(Unknown Source)\n\tat java.base/java.util.concurrent.ConcurrentSkipListSet.first(Unknown Source)\n\tat io.confluent.parallelconsumer.state.PartitionState.getOffsetHighestSequentialSucceeded(PartitionState.java:267)\n\tat io.confluent.parallelconsumer.state.PartitionState.getNextExpectedPolledOffset(PartitionState.java:235)\n\tat io.confluent.parallelconsumer.state.PartitionState.createOffsetAndMetadata(PartitionState.java:225)\n\tat io.confluent.parallelconsumer.state.PartitionState.getCommitDataIfDirty(PartitionState.java:218)\n\tat io.confluent.parallelconsumer.state.PartitionStateManager.collectDirtyCommitData(PartitionStateManager.java:373)\n\tat io.confluent.parallelconsumer.state.WorkManager.collectCommitDataForDirtyPartitions(WorkManager.java:198)\n\tat io.confluent.parallelconsumer.internal.AbstractOffsetCommitter.retrieveOffsetsAndCommit(AbstractOffsetCommitter.java:31)\n\tat io.confluent.parallelconsumer.internal.ConsumerOffsetCommitter.maybeDoCommit(ConsumerOffsetCommitter.java:179)\n\tat java.base/java.util.Optional.ifPresent(Unknown Source)\n\tat io.confluent.parallelconsumer.internal.BrokerPollSystem.maybeDoCommit(BrokerPollSystem.java:323)\n\tat io.confluent.parallelconsumer.internal.BrokerPollSystem.controlLoop(BrokerPollSystem.java:118)\n\t... 4 common frames omitted