PartitionConsumer::stream should not block while there are no records

[nicholastmosher]

If I create a new empty topic and run the following code, it will block indefinitely:

let consumer = fluvio::consumer("my-topic", 0).await?;

// This call blocks until a record appears in the partition
let mut stream = consumer.stream(Offset::beginning()).await?;

This is not desirable. The expected behavior would be for the stream object to be returned as soon as a connection to the partition leader is created. Waiting should only happen if we explicitly await on fetching elements on the stream, e.g. here:

while let Some(Ok(record)) = stream.next().await { ... }