/// Given the write described by this summary, and the sequencer's
/// progress for a particular kafka partition, returns the status
/// of that write in this write summary
Given the write described by this summary, and the sequencer's
progress for a particular kafka partition, returns the status
of that write in this write summary
Given the write described by this summary and the sequencer ' s progress for a particular kafka partition returns the status of that write in this write summary
pub fn write_status(
        &self,
        kafka_partition: KafkaPartition,
        progress: &SequencerProgress,
    ) -> Result<KafkaPartitionWriteStatus> {
        let sequence_numbers = self
            .sequencers
            .get(&kafka_partition)
            .context(UnknownKafkaPartitionSnafu { kafka_partition })?;
        debug!(
            ?kafka_partition,
            ?progress,
            ?sequence_numbers,
            "write_status"
        );
        if progress.is_empty() {
            return Ok(KafkaPartitionWriteStatus::KafkaPartitionUnknown);
        }
        let is_persisted = sequence_numbers
            .iter()
            .all(|sequence_number| progress.persisted(*sequence_number));
        if is_persisted {
            return Ok(KafkaPartitionWriteStatus::Persisted);
        }
        let is_readable = sequence_numbers
            .iter()
            .all(|sequence_number| progress.readable(*sequence_number));
        if is_readable {
            return Ok(KafkaPartitionWriteStatus::Readable);
        }
        Ok(KafkaPartitionWriteStatus::Durable)
    }
pub fn write_status ( & self , kafka_partition : KafkaPartition , progress : & SequencerProgress , ) -> Result < KafkaPartitionWriteStatus > { let sequence_numbers = self . sequencers . get ( & kafka_partition ) . context ( UnknownKafkaPartitionSnafu { kafka_partition } ) ? ; debug ! ( kafka_partition progress sequence_numbers " " ) ; if progress . is_empty ( ) { return Ok ( KafkaPartitionWriteStatus :: KafkaPartitionUnknown ) ; } let is_persisted = sequence_numbers . iter ( ) . all ( | sequence_number | progress . persisted ( * sequence_number ) ) ; if is_persisted { return Ok ( KafkaPartitionWriteStatus :: Persisted ) ; } let is_readable = sequence_numbers . iter ( ) . all ( | sequence_number | progress . readable ( * sequence_number ) ) ; if is_readable { return Ok ( KafkaPartitionWriteStatus :: Readable ) ; } Ok ( KafkaPartitionWriteStatus :: Durable ) }
Given the write described by this summary, and the sequencer's
progress for a particular kafka partition, returns the status
of that write in this write summary
Given the write described by this summary and the sequencer ' s progress for a particular kafka partition returns the status of that write in this write summary
&self
kafka_partition
KafkaPartition
progress
&SequencerProgress