message: Defer stream initialization in consumer
  Fixes
diff: @@ -5,6 +5,7 @@ use tracing::{debug, error, trace, instrument};
  use once_cell::sync::Lazy;
  use futures_util::future::{Either, err};
  use futures_util::stream::{StreamExt, once, iter};
  +use futures_util::FutureExt;
  use fluvio_spu_schema::server::stream_fetch::{
  DefaultStreamFetchRequest, DefaultStreamFetchResponse, SmartStreamPayload, SmartStreamWasm,
  @@ -441,12 +442,12 @@ impl PartitionConsumer {
  stream_request.wasm_payload = Some(module);
  }
  }
  -
  let mut stream = self
  .pool
  .create_stream_with_version(&replica, stream_request, stream_fetch_version)
  .await?;
  + let ft_stream = async move {
  if let Some(Ok(response)) = stream.next().await {
  let stream_id = response.stream_id;
  @@ -510,12 +511,15 @@ impl PartitionConsumer {
  })
  .map_err(|e| e.into())
  });
  - Ok(Either::Left(iter(vec![Ok(response)]).chain(
  - publish_stream::EndPublishSt::new(update_stream, publisher),
  - )))
  + Either::Left(
  + iter(vec![Ok(response)])
  + .chain(publish_stream::EndPublishSt::new(update_stream, publisher)),
  + )
  } else {
  - Ok(Either::Right(empty()))
  + Either::Right(empty())
  }
  + };
  + Ok(ft_stream.flatten_stream().boxed())
  }
  }
  