[Bug]: Offset::end() with topic replicas reads from beginning. about fluvio HOT 6 CLOSED

use std::sync::Arc;
use fluvio::{metadata::topic::TopicSpec, FluvioAdmin, Offset, RecordKey};
use futures::TryStreamExt;
use std::time::Duration;

const DELAY_MILLIS: u64 = 1000;
const MAX_RECORDS: u8 = 15;
const REC_TRIGGER_OFFSET: u8 = 11;  // start sometime after FLV_SHORT_RECONCILLIATION setting in seconds
const TOPIC_NAME: &str = "dectest-offset";
const TOPIC_REPLICAS: u32 = 3;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    reset_topic().await?;

    println!("Number of topic replicas: {}", TOPIC_REPLICAS);

    let notify_start = Arc::new(tokio::sync::Notify::new());
    tokio::spawn(consume(TestOffset::Beginning, notify_start.clone()));
    tokio::spawn(consume(TestOffset::End, notify_start.clone()));

    let producer = fluvio::producer(TOPIC_NAME).await?;
    for index in 0..MAX_RECORDS {
        producer.send(RecordKey::NULL, index.to_string()).await?;
        println!("[PRODUCER] sent: {}", index);
        tokio::time::sleep(Duration::from_millis(DELAY_MILLIS)).await;
    }

    Ok(())
}

#[derive(PartialEq)]
enum TestOffset {
    Beginning,
    End,
}

async fn consume(offset: TestOffset, onotify: Arc<tokio::sync::Notify>) -> anyhow::Result<()> {
    // wait for consume signal to start
    if offset == TestOffset::End {
        onotify.notified().await;
    }

    let mut stream = fluvio::consumer(TOPIC_NAME, 0)
        .await?
        .stream(match offset {
            TestOffset::Beginning => Offset::beginning(),
            TestOffset::End => Offset::end(),
        })
        .await?;

    while let Some(record) = stream.try_next().await? {
        let index = record.get_value().as_utf8_lossy_string().parse::<u8>()?;

        println!(
            "[CONSUMER_{}] recieved: {}",
            match offset {
                TestOffset::Beginning => "BEGINNING",
                TestOffset::End => "END",
            },
            index
        );

        if offset == TestOffset::Beginning && index == REC_TRIGGER_OFFSET {
            onotify.notify_waiters();
        }

        if index == MAX_RECORDS - 1 {
            break;
        }
    }

    Ok(())
}

async fn reset_topic() -> anyhow::Result<()> {
    let admin = FluvioAdmin::connect().await?;
    let _ = admin.delete::<TopicSpec>(TOPIC_NAME).await;
    admin
        .create(TOPIC_NAME.to_string(), false, TopicSpec::new_computed(1, TOPIC_REPLICAS, None))
        .await?;

    Ok(())
}