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.