feat: allow StreamConsumer to start at "earliest"/"latest" offset

[crepererum]

This is required to provide a race-free way to start at the earliest or
latest offset, because fetching the offset and then creating a stream
might fail because in the meantime the Kafka retention policy might
already have removed the offsets in question.

[tustvold]

Looks good to me, had some minor comments

[tustvold]

I don't understand this, I'm not sure under what circumstance Kafka would return no records, but if this did occur, wouldn't we just get stuck in an infinite loop?

[crepererum]

So the issue here is that "latest" offset: If you have a new or empty partition, then Kafka returns the start of that empty interval (e.g. 0 if it's an empty partition) if you ask it, also see

rskafka/tests/client.rs

Lines 241 to 251 in 15f424f

	assert_eq!(
	partition_client
	.get_offset(OffsetAt::Earliest)
	.await
	.unwrap(),
	0
	);
	assert_eq!(
	partition_client.get_offset(OffsetAt::Latest).await.unwrap(),
	0
	);

Now if you don't remember that offset, the time your produce something to the partition the "latest" offset (which is the latest record + 1) will also increase, so you're always chasing the tail of the partition but never fetch any records. So when you first get an answer to "what's the earliest/latest offset?" you need to remember that (except when this turns out to be wrong due to "offset out of range").

When the partition stays empty, then sure this loops forever and asks Kafka again and again for data. However this is not a hot loop since the fetch request has a timeout attached, so the broker will wait some milliseconds until it tells us "there are no records here, like the last X time you've asked me".

[tustvold]

The interaction with the offset tracking in buffer is perhaps a little funky. Perhaps this might be easier to understand if these were tracked automatically, and we added a delete records API... I dunno 😅

[crepererum]

You mean you wanna mock a full Kafka broker? 😉