fn fetch_logs<'a>(
        &'a mut self,
        _offset: i64,
        _max_bytes: i32,
        _isolation: Isolation
    ) -> BoxStream<'a,FetchablePartitionResponse<DefaultRecords>> {

        empty().boxed()

    /*
    et offsets = leader.fetch_offsets().await?;
    let last_offset = offsets.last_stable_offset();

    let mut current_offset = if opt.from_beginning {
        offsets.start_offset()
    } else {
        offsets.last_stable_offset()
    };

    debug!("entering loop");

    loop {
        debug!("fetching with offset: {}", current_offset);

        let fetch_logs_res = leader.fetch_logs(current_offset, opt.max_bytes).await?;

        current_offset = fetch_logs_res.last_stable_offset;

        debug!("fetching last offset: {} and response", last_offset);

        // process logs response

        process_fetch_topic_response(out.clone(), &topic, fetch_logs_res, &opt).await?;
        let read_lock = end.read().unwrap();
        if *read_lock {
            debug!("detected end by ctrl-c, exiting loop");
            break;
        }

        /*
        if !opt.continuous {
            if last_offset > last_offset {
                debug!("finishing fetch loop");
                break;
            }
        }
        */

        
        if !opt.continuous {
            debug!("finishing fetch loop");
            break;
        }
        

        debug!("sleeping 200 ms between next fetch");
        sleep(Duration::from_millis(200)).await;
    
        */


    }
/// Fetch log records from a target server
Fetch log records from a target server
Fetch log records from a target server
fn fetch_logs<'a>(
        &'a mut self,
        _offset: i64,
        _max_bytes: i32,
        _isolation: Isolation
    ) -> BoxStream<'a,FetchablePartitionResponse<DefaultRecords>> {
        empty().boxed()
    /*
    et offsets = leader.fetch_offsets().await?;
    let last_offset = offsets.last_stable_offset();
    let mut current_offset = if opt.from_beginning {
        offsets.start_offset()
    } else {
        offsets.last_stable_offset()
    };
    debug!("entering loop");
    loop {
        debug!("fetching with offset: {}", current_offset);
        let fetch_logs_res = leader.fetch_logs(current_offset, opt.max_bytes).await?;
        current_offset = fetch_logs_res.last_stable_offset;
        debug!("fetching last offset: {} and response", last_offset);
        // process logs response
        process_fetch_topic_response(out.clone(), &topic, fetch_logs_res, &opt).await?;
        let read_lock = end.read().unwrap();
        if *read_lock {
            debug!("detected end by ctrl-c, exiting loop");
            break;
        }
        /*
        if !opt.continuous {
            if last_offset > last_offset {
                debug!("finishing fetch loop");
                break;
            }
        }
        */
        if !opt.continuous {
            debug!("finishing fetch loop");
            break;
        }
        debug!("sleeping 200 ms between next fetch");
        sleep(Duration::from_millis(200)).await;
        */
    }
fn fetch_logs < ' a > ( & ' a mut self , _offset : i64 , _max_bytes : i32 , _isolation : Isolation ) -> BoxStream < ' a , FetchablePartitionResponse < DefaultRecords > > { empty ( ) . boxed ( ) /*
    et offsets = leader.fetch_offsets().await?;
    let last_offset = offsets.last_stable_offset();

    let mut current_offset = if opt.from_beginning {
        offsets.start_offset()
    } else {
        offsets.last_stable_offset()
    };

    debug!("entering loop");

    loop {
        debug!("fetching with offset: {}", current_offset);

        let fetch_logs_res = leader.fetch_logs(current_offset, opt.max_bytes).await?;

        current_offset = fetch_logs_res.last_stable_offset;

        debug!("fetching last offset: {} and response", last_offset);

        // process logs response

        process_fetch_topic_response(out.clone(), &topic, fetch_logs_res, &opt).await?;
        let read_lock = end.read().unwrap();
        if *read_lock {
            debug!("detected end by ctrl-c, exiting loop");
            break;
        }

        /*
        if !opt.continuous {
            if last_offset > last_offset {
                debug!("finishing fetch loop");
                break;
            }
        }
        */

        
        if !opt.continuous {
            debug!("finishing fetch loop");
            break;
        }
        

        debug!("sleeping 200 ms between next fetch");
        sleep(Duration::from_millis(200)).await;
    
        */ }
Fetch log records from a target server
Fetch log records from a target server
&'a mut self
_offset
i64
_max_bytes
i32
_isolation
Isolation
&'a mut self
_offset
i64
_max_bytes
i32
_isolation
Isolation