Improvement of seek functionality in kafka consumer

[a1shadows]

For the newly added seek semantics in the kafka consume step, there are many things that are not ideal. I think we should address those.

1. Upon retry on a consume step with seek, it reseeks to the timestamp and consumes duplicate records. Ideally it should only seek when the current offset is less than the offset to seek to. There is really no real reason I can see for a consumer to seek to an already consumed records
2. Insufficient CI tests to cover edge scenarios around seek.
3. Currently, the consumer only assigns to the partitions that have a valid offset > the seek offset for timestamp. Ideally, it should assign to all the partitions, seeking to end for partitions that don't have messages yet with offset > seekOffset

[a1shadows]

@authorjapps Let me know if you can think of any other cases I should cover