fn handle_message<S: SourceReader>(
    message: SourceMessage<S::Key, S::Value>,
    consistency_info: &mut ConsistencyInfo,
    bytes_read: &mut usize,
    cap: &Capability<Timestamp>,
    output: &mut OutputHandle<
        Timestamp,
        Result<SourceOutput<S::Key, S::Value>, String>,
        Tee<Timestamp, Result<SourceOutput<S::Key, S::Value>, String>>,
    >,
    metric_updates: &mut HashMap<PartitionId, (MzOffset, Timestamp)>,
    timer: &std::time::Instant,
    buffer: &mut Option<SourceMessage<S::Key, S::Value>>,
    timestamp_bindings: &TimestampBindingRc,
) -> (SourceStatus, MessageProcessing) {
    let partition = message.partition.clone();
    let offset = message.offset;

    // Update ingestion metrics. Guaranteed to exist as the appropriate
    // entry gets created in SourceReader or when a new partition
    // is discovered
    consistency_info
        .source_metrics
        .partition_metrics
        .get_mut(&partition)
        .unwrap_or_else(|| panic!("partition metrics do not exist for partition {}", partition))
        .offset_received
        .set(offset.offset);

    // Determine the timestamp to which we need to assign this message
    let ts = consistency_info.find_matching_timestamp(&partition, offset, timestamp_bindings);
    match ts {
        None => {
            // We have not yet decided on a timestamp for this message,
            // we need to buffer the message
            *buffer = Some(message);
            (SourceStatus::Alive, MessageProcessing::Yielded)
        }
        Some(ts) => {
            // Note: empty and null payload/keys are currently
            // treated as the same thing.
            let key = message.key;
            let out = message.value;
            // Entry for partition_metadata is guaranteed to exist as messages
            // are only processed after we have updated the partition_metadata for a
            // partition and created a partition queue for it.
            if let Some(len) = key.len() {
                *bytes_read += len;
            }
            if let Some(len) = out.len() {
                *bytes_read += len;
            }
            let ts_cap = cap.delayed(&ts);

            output.session(&ts_cap).give(Ok(SourceOutput::new(
                key,
                out,
                Some(offset.offset),
                message.upstream_time_millis,
                message.partition,
            )));

            // Update ingestion metrics
            // Entry is guaranteed to exist as it gets created when we initialise the partition
            let partition_metrics = consistency_info
                .source_metrics
                .partition_metrics
                .get_mut(&partition)
                .unwrap();
            partition_metrics.offset_ingested.set(offset.offset);
            partition_metrics.messages_ingested.inc();

            metric_updates.insert(partition, (offset, ts));

            if timer.elapsed().as_millis() > YIELD_INTERVAL_MS {
                // We didn't drain the entire queue, so indicate that we
                // should run again but yield the CPU to other operators.
                (SourceStatus::Alive, MessageProcessing::Yielded)
            } else {
                (SourceStatus::Alive, MessageProcessing::Active)
            }
        }
    }
}
/// Take `message` and assign it the appropriate timestamps and push it into the
/// dataflow layer, if possible.
///
/// TODO: This function is a bit of a mess rn but hopefully this function makes the
/// existing mess more obvious and points towards ways to improve it.
Take `message` and assign it the appropriate timestamps and push it into the
dataflow layer, if possible.

This function is a bit of a mess rn but hopefully this function makes the
existing mess more obvious and points towards ways to improve it.
Take ` message ` and assign it the appropriate timestamps and push it into the dataflow layer if possible . This function is a bit of a mess rn but hopefully this function makes the existing mess more obvious and points towards ways to improve it .
fn handle_message<S: SourceReader>(
    message: SourceMessage<S::Key, S::Value>,
    consistency_info: &mut ConsistencyInfo,
    bytes_read: &mut usize,
    cap: &Capability<Timestamp>,
    output: &mut OutputHandle<
        Timestamp,
        Result<SourceOutput<S::Key, S::Value>, String>,
        Tee<Timestamp, Result<SourceOutput<S::Key, S::Value>, String>>,
    >,
    metric_updates: &mut HashMap<PartitionId, (MzOffset, Timestamp)>,
    timer: &std::time::Instant,
    buffer: &mut Option<SourceMessage<S::Key, S::Value>>,
    timestamp_bindings: &TimestampBindingRc,
) -> (SourceStatus, MessageProcessing) {
    let partition = message.partition.clone();
    let offset = message.offset;
    // Update ingestion metrics. Guaranteed to exist as the appropriate
    // entry gets created in SourceReader or when a new partition
    // is discovered
    consistency_info
        .source_metrics
        .partition_metrics
        .get_mut(&partition)
        .unwrap_or_else(|| panic!("partition metrics do not exist for partition {}", partition))
        .offset_received
        .set(offset.offset);
    // Determine the timestamp to which we need to assign this message
    let ts = consistency_info.find_matching_timestamp(&partition, offset, timestamp_bindings);
    match ts {
        None => {
            // We have not yet decided on a timestamp for this message,
            // we need to buffer the message
            *buffer = Some(message);
            (SourceStatus::Alive, MessageProcessing::Yielded)
        }
        Some(ts) => {
            // Note: empty and null payload/keys are currently
            // treated as the same thing.
            let key = message.key;
            let out = message.value;
            // Entry for partition_metadata is guaranteed to exist as messages
            // are only processed after we have updated the partition_metadata for a
            // partition and created a partition queue for it.
            if let Some(len) = key.len() {
                *bytes_read += len;
            }
            if let Some(len) = out.len() {
                *bytes_read += len;
            }
            let ts_cap = cap.delayed(&ts);
            output.session(&ts_cap).give(Ok(SourceOutput::new(
                key,
                out,
                Some(offset.offset),
                message.upstream_time_millis,
                message.partition,
            )));
            // Update ingestion metrics
            // Entry is guaranteed to exist as it gets created when we initialise the partition
            let partition_metrics = consistency_info
                .source_metrics
                .partition_metrics
                .get_mut(&partition)
                .unwrap();
            partition_metrics.offset_ingested.set(offset.offset);
            partition_metrics.messages_ingested.inc();
            metric_updates.insert(partition, (offset, ts));
            if timer.elapsed().as_millis() > YIELD_INTERVAL_MS {
                // We didn't drain the entire queue, so indicate that we
                // should run again but yield the CPU to other operators.
                (SourceStatus::Alive, MessageProcessing::Yielded)
            } else {
                (SourceStatus::Alive, MessageProcessing::Active)
            }
        }
    }
}
fn handle_message < S : SourceReader > ( message : SourceMessage < S :: Key , S :: Value > , consistency_info : & mut ConsistencyInfo , bytes_read : & mut usize , cap : & Capability < Timestamp > , output : & mut OutputHandle < Timestamp , Result < SourceOutput < S :: Key , S :: Value > , String > , Tee < Timestamp , Result < SourceOutput < S :: Key , S :: Value > , String > > , > , metric_updates : & mut HashMap < PartitionId , ( MzOffset , Timestamp ) > , timer : & std :: time :: Instant , buffer : & mut Option < SourceMessage < S :: Key , S :: Value > > , timestamp_bindings : & TimestampBindingRc , ) -> ( SourceStatus , MessageProcessing ) { let partition = message . partition . clone ( ) ; let offset = message . offset ; // Update ingestion metrics. Guaranteed to exist as the appropriate // entry gets created in SourceReader or when a new partition // is discovered consistency_info . source_metrics . partition_metrics . get_mut ( & partition ) . unwrap_or_else ( | | panic ! ( " " partition ) ) . offset_received . set ( offset . offset ) ; // Determine the timestamp to which we need to assign this message let ts = consistency_info . find_matching_timestamp ( & partition , offset , timestamp_bindings ) ; match ts { None => { // We have not yet decided on a timestamp for this message, // we need to buffer the message * buffer = Some ( message ) ; ( SourceStatus :: Alive , MessageProcessing :: Yielded ) } Some ( ts ) => { // Note: empty and null payload/keys are currently // treated as the same thing. let key = message . key ; let out = message . value ; // Entry for partition_metadata is guaranteed to exist as messages // are only processed after we have updated the partition_metadata for a // partition and created a partition queue for it. if let Some ( len ) = key . len ( ) { * bytes_read += len ; } if let Some ( len ) = out . len ( ) { * bytes_read += len ; } let ts_cap = cap . delayed ( & ts ) ; output . session ( & ts_cap ) . give ( Ok ( SourceOutput :: new ( key , out , Some ( offset . offset ) , message . upstream_time_millis , message . partition , ) ) ) ; // Update ingestion metrics // Entry is guaranteed to exist as it gets created when we initialise the partition let partition_metrics = consistency_info . source_metrics . partition_metrics . get_mut ( & partition ) . unwrap ( ) ; partition_metrics . offset_ingested . set ( offset . offset ) ; partition_metrics . messages_ingested . inc ( ) ; metric_updates . insert ( partition , ( offset , ts ) ) ; if timer . elapsed ( ) . as_millis ( ) > YIELD_INTERVAL_MS { // We didn't drain the entire queue, so indicate that we // should run again but yield the CPU to other operators. ( SourceStatus :: Alive , MessageProcessing :: Yielded ) } else { ( SourceStatus :: Alive , MessageProcessing :: Active ) } } } }
Take `message` and assign it the appropriate timestamps and push it into the
dataflow layer, if possible.
Take ` message ` and assign it the appropriate timestamps and push it into the dataflow layer if possible .
message
SourceMessage<S::Key, S::Value>
consistency_info
&mut ConsistencyInfo
bytes_read
&mut usize
cap
&Capability<Timestamp>
output
&mut OutputHandle<
        Timestamp,
        Result<SourceOutput<S::Key, S::Value>, String>,
        Tee<Timestamp, Result<SourceOutput<S::Key, S::Value>, String>>,
    >
metric_updates
&mut HashMap<PartitionId, (MzOffset, Timestamp)>
timer
&std::time::Instant
buffer
&mut Option<SourceMessage<S::Key, S::Value>>
timestamp_bindings
&TimestampBindingRc
message
SourceMessage<S::Key, S::Value>
consistency_info
&mut ConsistencyInfo
bytes_read
&mut usize
cap
&Capability<Timestamp>
output
&mut OutputHandle<
        Timestamp,
        Result<SourceOutput<S::Key, S::Value>, String>,
        Tee<Timestamp, Result<SourceOutput<S::Key, S::Value>, String>>,
    >
metric_updates
&mut HashMap<PartitionId, (MzOffset, Timestamp)>
timer
&std::time::Instant
buffer
&mut Option<SourceMessage<S::Key, S::Value>>
timestamp_bindings
&TimestampBindingRc