fix the issue with offsetOutOfRange

[dethlex]

Summary
Handle ErrOffsetOutOfRange in goka's partition table recovery by falling back to sarama.OffsetOldest instead of crashing the processor

Problem

  Production error:                                                                                                                                                                                                                                                        
  error consuming %TOPIC%: kafka server: The requested offset is outside   
  the range of offsets maintained by the server for the given topic/partition         

Goka's findOffsetToLoad() adjusts offsets below the broker's oldest, but a race condition exists: the oldest offset can shift between the metadata fetch and the actual ConsumePartition call (e.g. due to log compaction or retention cleanup), causing ErrOffsetOutOfRange.

Fix When ConsumePartition returns ErrOffsetOutOfRange, retry with sarama.OffsetOldest (lets the broker resolve the actual earliest offset) instead of failing the partition setup.

[norbertklawikowski]

Thanks for creating the PR!

[norbertklawikowski]

The handling of other errors than sarama.ErrOffsetOutOfRange is missing here.

[norbertklawikowski]

Is it possible that a ErrOffsetOutOfRange can be returned by the Errors channel of the consumer inside loadMessages? It seems to be only possibly returned via the chooseStartingOffset which is called via ConsumePartition, thus while constructing a PartitionConsumer.

Here it would mean that while consuming from a PartitionConsumer this error can happen, and we would retry only one time which doesn't seem to be sufficient in this case. But I can not see that this is possible at all.