pub async fn lookup_start_offsets(
    consumer: Arc<BaseConsumer<KafkaErrCheckContext>>,
    topic: &str,
    with_options: &BTreeMap<String, Value>,
    now: u64,
) -> Result<Option<Vec<i64>>, anyhow::Error> {
    let time_offset = with_options.get("kafka_time_offset");
    if time_offset.is_none() {
        return Ok(None);
    } else if with_options.contains_key("start_offset") {
        bail!("`start_offset` and `kafka_time_offset` cannot be set at the same time.")
    }

    // Validate and resolve `kafka_time_offset`.
    let time_offset = match time_offset.unwrap() {
        Value::Number(s) => match s.parse::<i64>() {
            // Timestamp in millis *before* now (e.g. -10 means 10 millis ago)
            Ok(ts) if ts < 0 => {
                let now: i64 = now.try_into()?;
                let ts = now - ts.abs();
                if ts <= 0 {
                    bail!("Relative `kafka_time_offset` must be smaller than current system timestamp")
                }
                ts
            }
            // Timestamp in millis (e.g. 1622659034343)
            Ok(ts) => ts,
            _ => bail!("`kafka_time_offset` must be a number"),
        },
        _ => bail!("`kafka_time_offset` must be a number"),
    };

    // Lookup offsets
    task::spawn_blocking({
        let topic = topic.to_string();
        move || {
            // There cannot be more than i32 partitions
            let num_partitions =
                get_partitions(consumer.as_ref(), &topic, Duration::from_secs(10))?.len();

            let mut tpl = TopicPartitionList::with_capacity(1);
            tpl.add_partition_range(&topic, 0, num_partitions as i32 - 1);
            tpl.set_all_offsets(Offset::Offset(time_offset))?;

            let offsets_for_times = consumer.offsets_for_times(tpl, Duration::from_secs(10))?;

            // Translate to `start_offsets`
            let start_offsets = offsets_for_times
                .elements()
                .iter()
                .map(|elem| match elem.offset() {
                    Offset::Offset(offset) => Ok(offset),
                    Offset::End => fetch_end_offset(&consumer, &topic, elem.partition()),
                    _ => bail!(
                        "Unexpected offset {:?} for partition {}",
                        elem.offset(),
                        elem.partition()
                    ),
                })
                .collect::<Result<Vec<_>, _>>()?;

            if start_offsets.len() != num_partitions {
                bail!(
                    "Expected offsets for {} partitions, but recevied {}",
                    num_partitions,
                    start_offsets.len(),
                );
            }

            Ok(Some(start_offsets))
        }
    })
    .await?
}
/// Returns start offsets for the partitions of `topic` and the provided
/// `kafka_time_offset` option.
///
/// For each partition, the returned offset is the earliest offset whose
/// timestamp is greater than or equal to the given timestamp for the
/// partition. If no such message exists (or the Kafka broker is before
/// 0.10.0), the current end offset is returned for the partition.
///
/// The provided `kafka_time_offset` option must be a non-zero number:
/// * Non-Negative numbers will used as is (e.g. `1622659034343`)
/// * Negative numbers will be translated to a timestamp in millis
///   before now (e.g. `-10` means 10 millis ago)
///
/// If `kafka_time_offset` has not been configured, an empty Option is
/// returned.
Returns start offsets for the partitions of `topic` and the provided
`kafka_time_offset` option.

For each partition, the returned offset is the earliest offset whose
timestamp is greater than or equal to the given timestamp for the
partition. If no such message exists (or the Kafka broker is before
0.10.0), the current end offset is returned for the partition.

The provided `kafka_time_offset` option must be a non-zero number:
Non-Negative numbers will used as is
Negative numbers will be translated to a timestamp in millis
before now

If `kafka_time_offset` has not been configured, an empty Option is
returned.
Returns start offsets for the partitions of ` topic ` and the provided ` kafka_time_offset ` option . For each partition the returned offset is the earliest offset whose timestamp is greater than or equal to the given timestamp for the partition . If no such message exists ( or the Kafka broker is before 0 . 10 . 0 ) the current end offset is returned for the partition . The provided ` kafka_time_offset ` option must be a non - zero number : Non - Negative numbers will used as is Negative numbers will be translated to a timestamp in millis before now If ` kafka_time_offset ` has not been configured an empty Option is returned .
pub async fn lookup_start_offsets(
    consumer: Arc<BaseConsumer<KafkaErrCheckContext>>,
    topic: &str,
    with_options: &BTreeMap<String, Value>,
    now: u64,
) -> Result<Option<Vec<i64>>, anyhow::Error> {
    let time_offset = with_options.get("kafka_time_offset");
    if time_offset.is_none() {
        return Ok(None);
    } else if with_options.contains_key("start_offset") {
        bail!("`start_offset` and `kafka_time_offset` cannot be set at the same time.")
    }
    // Validate and resolve `kafka_time_offset`.
    let time_offset = match time_offset.unwrap() {
        Value::Number(s) => match s.parse::<i64>() {
            // Timestamp in millis *before* now (e.g. -10 means 10 millis ago)
            Ok(ts) if ts < 0 => {
                let now: i64 = now.try_into()?;
                let ts = now - ts.abs();
                if ts <= 0 {
                    bail!("Relative `kafka_time_offset` must be smaller than current system timestamp")
                }
                ts
            }
            // Timestamp in millis (e.g. 1622659034343)
            Ok(ts) => ts,
            _ => bail!("`kafka_time_offset` must be a number"),
        },
        _ => bail!("`kafka_time_offset` must be a number"),
    };
    // Lookup offsets
    task::spawn_blocking({
        let topic = topic.to_string();
        move || {
            // There cannot be more than i32 partitions
            let num_partitions =
                get_partitions(consumer.as_ref(), &topic, Duration::from_secs(10))?.len();
            let mut tpl = TopicPartitionList::with_capacity(1);
            tpl.add_partition_range(&topic, 0, num_partitions as i32 - 1);
            tpl.set_all_offsets(Offset::Offset(time_offset))?;
            let offsets_for_times = consumer.offsets_for_times(tpl, Duration::from_secs(10))?;
            // Translate to `start_offsets`
            let start_offsets = offsets_for_times
                .elements()
                .iter()
                .map(|elem| match elem.offset() {
                    Offset::Offset(offset) => Ok(offset),
                    Offset::End => fetch_end_offset(&consumer, &topic, elem.partition()),
                    _ => bail!(
                        "Unexpected offset {:?} for partition {}",
                        elem.offset(),
                        elem.partition()
                    ),
                })
                .collect::<Result<Vec<_>, _>>()?;
            if start_offsets.len() != num_partitions {
                bail!(
                    "Expected offsets for {} partitions, but recevied {}",
                    num_partitions,
                    start_offsets.len(),
                );
            }
            Ok(Some(start_offsets))
        }
    })
    .await?
}
pub async fn lookup_start_offsets ( consumer : Arc < BaseConsumer < KafkaErrCheckContext > > , topic : & str , with_options : & BTreeMap < String , Value > , now : u64 , ) -> Result < Option < Vec < i64 > > , anyhow :: Error > { let time_offset = with_options . get ( " " ) ; if time_offset . is_none ( ) { return Ok ( None ) ; } else if with_options . contains_key ( " " ) { bail ! ( " " ) } // Validate and resolve `kafka_time_offset`. let time_offset = match time_offset . unwrap ( ) { Value :: Number ( s ) => match s . parse :: < i64 > ( ) { // Timestamp in millis *before* now (e.g. -10 means 10 millis ago) Ok ( ts ) if ts < 0 => { let now : i64 = now . try_into ( ) ? ; let ts = now - ts . abs ( ) ; if ts <= 0 { bail ! ( " " ) } ts } // Timestamp in millis (e.g. 1622659034343) Ok ( ts ) => ts , _ => bail ! ( " " ) , } , _ => bail ! ( " " ) , } ; // Lookup offsets task :: spawn_blocking ( { let topic = topic . to_string ( ) ; move | | { // There cannot be more than i32 partitions let num_partitions = get_partitions ( consumer . as_ref ( ) , & topic , Duration :: from_secs ( 10 ) ) ? . len ( ) ; let mut tpl = TopicPartitionList :: with_capacity ( 1 ) ; tpl . add_partition_range ( & topic , 0 , num_partitions as i32 - 1 ) ; tpl . set_all_offsets ( Offset :: Offset ( time_offset ) ) ? ; let offsets_for_times = consumer . offsets_for_times ( tpl , Duration :: from_secs ( 10 ) ) ? ; // Translate to `start_offsets` let start_offsets = offsets_for_times . elements ( ) . iter ( ) . map ( | elem | match elem . offset ( ) { Offset :: Offset ( offset ) => Ok ( offset ) , Offset :: End => fetch_end_offset ( & consumer , & topic , elem . partition ( ) ) , _ => bail ! ( " " elem offset ( ) elem partition ( ) ) , } ) . collect :: < Result < Vec < _ > , _ > > ( ) ? ; if start_offsets . len ( ) != num_partitions { bail ! ( " " num_partitions start_offsets len ( ) ) ; } Ok ( Some ( start_offsets ) ) } } ) . await ? }
Returns start offsets for the partitions of `topic` and the provided
`kafka_time_offset` option.
Returns start offsets for the partitions of ` topic ` and the provided ` kafka_time_offset ` option .
consumer
Arc<BaseConsumer<KafkaErrCheckContext>>
topic
&str
with_options
&BTreeMap<String, Value>
now
u64
consumer
Arc<BaseConsumer<KafkaErrCheckContext>>
topic
&str
with_options
&BTreeMap<String, Value>
now
u64