fn __fetch_group_offsets(req: protocol::OffsetFetchRequest,
                         num_partitions: usize,
                         state: &mut state::ClientState,
                         conn_pool: &mut network::Connections,
                         config: &ClientConfig)
                         -> Result<Vec<TopicPartitionOffset>> {
    let mut attempt = 1;
    loop {
        let now = Instant::now();

        let r = {
            let host = try!(__get_group_coordinator(req.group, state, conn_pool, config, now));
            debug!("fetch_group_offsets: fetching on protocol v{} for '{}' from: {}",
                   req.header.api_version,
                   req.group,
                   host);
            try!(__send_receive::<_, protocol::OffsetFetchResponse>(conn_pool, host, now, &req))
        };

        let mut retry_code = None;
        let mut v = Vec::with_capacity(num_partitions);

        'rproc: for tp in r.topic_partitions {
            for p in tp.partitions {
                let mut o = p.get_offsets(tp.topic.clone());
                match o.offset {
                    Ok(_) => {
                        v.push(o);
                    }
                    Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition)) => {
                        // ~ occurs only on protocol v0 when no offset available
                        // for the group in question; we'll align the behavior
                        // with protocol v1.
                        o.offset = Ok(-1);
                        v.push(o);
                    }
                    Err(Error::Kafka(e @ KafkaCode::GroupLoadInProgress)) => {
                        retry_code = Some(e);
                        break 'rproc;
                    }
                    Err(Error::Kafka(e @ KafkaCode::NotCoordinatorForGroup)) => {
                        debug!("fetch_group_offsets: resetting group coordinator for '{}'",
                               req.group);
                        state.remove_group_coordinator(&req.group);
                        retry_code = Some(e);
                        break 'rproc;
                    }
                    Err(e) => {
                        // ~ immeditaly abort with the error
                        return Err(e);
                    }
                }
            }
        }
        // ~ have we processed the result successfully or shall we
        // retry once more?
        match retry_code {
            Some(e) => {
                if attempt < config.retry_max_attempts {
                    debug!("fetch_group_offsets: will retry request (c: {}) due to: {:?}",
                           req.header.correlation_id,
                           e);
                    attempt += 1;
                    __retry_sleep(config)
                } else {
                    return Err(Error::Kafka(e));
                }
            }
            None => {
                return Ok(v);
            }
        }
    }
}
// ~ 'num_partitions' is merely a hint to pre-allocation inside this method
~ 'num_partitions' is merely a hint to pre-allocation inside this method
~ ' num_partitions ' is merely a hint to pre - allocation inside this method
fn __fetch_group_offsets(req: protocol::OffsetFetchRequest,
                         num_partitions: usize,
                         state: &mut state::ClientState,
                         conn_pool: &mut network::Connections,
                         config: &ClientConfig)
                         -> Result<Vec<TopicPartitionOffset>> {
    let mut attempt = 1;
    loop {
        let now = Instant::now();
        let r = {
            let host = try!(__get_group_coordinator(req.group, state, conn_pool, config, now));
            debug!("fetch_group_offsets: fetching on protocol v{} for '{}' from: {}",
                   req.header.api_version,
                   req.group,
                   host);
            try!(__send_receive::<_, protocol::OffsetFetchResponse>(conn_pool, host, now, &req))
        };
        let mut retry_code = None;
        let mut v = Vec::with_capacity(num_partitions);
        'rproc: for tp in r.topic_partitions {
            for p in tp.partitions {
                let mut o = p.get_offsets(tp.topic.clone());
                match o.offset {
                    Ok(_) => {
                        v.push(o);
                    }
                    Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition)) => {
                        // ~ occurs only on protocol v0 when no offset available
                        // for the group in question; we'll align the behavior
                        // with protocol v1.
                        o.offset = Ok(-1);
                        v.push(o);
                    }
                    Err(Error::Kafka(e @ KafkaCode::GroupLoadInProgress)) => {
                        retry_code = Some(e);
                        break 'rproc;
                    }
                    Err(Error::Kafka(e @ KafkaCode::NotCoordinatorForGroup)) => {
                        debug!("fetch_group_offsets: resetting group coordinator for '{}'",
                               req.group);
                        state.remove_group_coordinator(&req.group);
                        retry_code = Some(e);
                        break 'rproc;
                    }
                    Err(e) => {
                        // ~ immeditaly abort with the error
                        return Err(e);
                    }
                }
            }
        }
        // ~ have we processed the result successfully or shall we
        // retry once more?
        match retry_code {
            Some(e) => {
                if attempt < config.retry_max_attempts {
                    debug!("fetch_group_offsets: will retry request (c: {}) due to: {:?}",
                           req.header.correlation_id,
                           e);
                    attempt += 1;
                    __retry_sleep(config)
                } else {
                    return Err(Error::Kafka(e));
                }
            }
            None => {
                return Ok(v);
            }
        }
    }
}
fn __fetch_group_offsets ( req : protocol :: OffsetFetchRequest , num_partitions : usize , state : & mut state :: ClientState , conn_pool : & mut network :: Connections , config : & ClientConfig ) -> Result < Vec < TopicPartitionOffset > > { let mut attempt = 1 ; loop { let now = Instant :: now ( ) ; let r = { let host = try ! ( __get_group_coordinator ( req group state conn_pool config now ) ) ; debug ! ( " " req header api_version req group host ) ; try ! ( __send_receive protocol OffsetFetchResponse ( conn_pool host now req ) ) } ; let mut retry_code = None ; let mut v = Vec :: with_capacity ( num_partitions ) ; ' rproc : for tp in r . topic_partitions { for p in tp . partitions { let mut o = p . get_offsets ( tp . topic . clone ( ) ) ; match o . offset { Ok ( _ ) => { v . push ( o ) ; } Err ( Error :: Kafka ( KafkaCode :: UnknownTopicOrPartition ) ) => { // ~ occurs only on protocol v0 when no offset available // for the group in question; we'll align the behavior // with protocol v1. o . offset = Ok ( - 1 ) ; v . push ( o ) ; } Err ( Error :: Kafka ( e @ KafkaCode :: GroupLoadInProgress ) ) => { retry_code = Some ( e ) ; break ' rproc ; } Err ( Error :: Kafka ( e @ KafkaCode :: NotCoordinatorForGroup ) ) => { debug ! ( " " req group ) ; state . remove_group_coordinator ( & req . group ) ; retry_code = Some ( e ) ; break ' rproc ; } Err ( e ) => { // ~ immeditaly abort with the error return Err ( e ) ; } } } } // ~ have we processed the result successfully or shall we // retry once more? match retry_code { Some ( e ) => { if attempt < config . retry_max_attempts { debug ! ( " " req header correlation_id e ) ; attempt += 1 ; __retry_sleep ( config ) } else { return Err ( Error :: Kafka ( e ) ) ; } } None => { return Ok ( v ) ; } } } }
~ 'num_partitions' is merely a hint to pre-allocation inside this method
~ ' num_partitions ' is merely a hint to pre - allocation inside this method
req
protocol::OffsetFetchRequest
num_partitions
usize
state
&mut state::ClientState
conn_pool
&mut network::Connections
config
&ClientConfig