-record(stream_state, {
    messages :: integer(),
    bytes :: integer(),
    first_seq :: integer(),
    first_ts :: binary(),
    last_seq :: integer(),
    last_ts :: binary(),
    consumer_count :: integer()
}).