async fn store_operation(
        &self,
        sequencer_id: u32,
        operation: &DmlOperation,
    ) -> Result<DmlMeta, WriteBufferError> {
        let partition = i32::try_from(sequencer_id)?;

        // truncate milliseconds from timestamps because that's what Kafka supports
        let now = operation
            .meta()
            .producer_ts()
            .unwrap_or_else(|| self.time_provider.now());

        let timestamp_millis = now.date_time().timestamp_millis();
        let timestamp = Time::from_timestamp_millis(timestamp_millis);

        let headers = IoxHeaders::new(
            ContentType::Protobuf,
            operation.meta().span_context().cloned(),
        );

        let mut buf = Vec::new();
        crate::codec::encode_operation(&self.database_name, operation, &mut buf)?;

        // This type annotation is necessary because `FutureRecord` is generic over key type, but
        // key is optional and we're not setting a key. `String` is arbitrary.
        let record: FutureRecord<'_, String, _> = FutureRecord::to(&self.database_name)
            .payload(&buf)
            .partition(partition)
            .timestamp(timestamp_millis)
            .headers((&headers).into());
        let kafka_write_size = estimate_message_size(
            record.payload.map(|v| v.as_ref()),
            record.key.map(|s| s.as_bytes()),
            record.headers.as_ref(),
        );

        debug!(db_name=%self.database_name, partition, size=buf.len(), "writing to kafka");

        let (partition, offset) = self
            .producer
            .send(record, Timeout::Never)
            .await
            .map_err(|(e, _owned_message)| Box::new(e))?;

        debug!(db_name=%self.database_name, %offset, %partition, size=buf.len(), "wrote to kafka");

        Ok(DmlMeta::sequenced(
            Sequence::new(partition.try_into()?, offset.try_into()?),
            timestamp,
            operation.meta().span_context().cloned(),
            kafka_write_size,
        ))
    }
/// Send an `Entry` to Kafka using the sequencer ID as a partition.
Send an `Entry` to Kafka using the sequencer ID as a partition.
Send an ` Entry ` to Kafka using the sequencer ID as a partition .
async fn store_operation(
        &self,
        sequencer_id: u32,
        operation: &DmlOperation,
    ) -> Result<DmlMeta, WriteBufferError> {
        let partition = i32::try_from(sequencer_id)?;
        // truncate milliseconds from timestamps because that's what Kafka supports
        let now = operation
            .meta()
            .producer_ts()
            .unwrap_or_else(|| self.time_provider.now());
        let timestamp_millis = now.date_time().timestamp_millis();
        let timestamp = Time::from_timestamp_millis(timestamp_millis);
        let headers = IoxHeaders::new(
            ContentType::Protobuf,
            operation.meta().span_context().cloned(),
        );
        let mut buf = Vec::new();
        crate::codec::encode_operation(&self.database_name, operation, &mut buf)?;
        // This type annotation is necessary because `FutureRecord` is generic over key type, but
        // key is optional and we're not setting a key. `String` is arbitrary.
        let record: FutureRecord<'_, String, _> = FutureRecord::to(&self.database_name)
            .payload(&buf)
            .partition(partition)
            .timestamp(timestamp_millis)
            .headers((&headers).into());
        let kafka_write_size = estimate_message_size(
            record.payload.map(|v| v.as_ref()),
            record.key.map(|s| s.as_bytes()),
            record.headers.as_ref(),
        );
        debug!(db_name=%self.database_name, partition, size=buf.len(), "writing to kafka");
        let (partition, offset) = self
            .producer
            .send(record, Timeout::Never)
            .await
            .map_err(|(e, _owned_message)| Box::new(e))?;
        debug!(db_name=%self.database_name, %offset, %partition, size=buf.len(), "wrote to kafka");
        Ok(DmlMeta::sequenced(
            Sequence::new(partition.try_into()?, offset.try_into()?),
            timestamp,
            operation.meta().span_context().cloned(),
            kafka_write_size,
        ))
    }
async fn store_operation ( & self , sequencer_id : u32 , operation : & DmlOperation , ) -> Result < DmlMeta , WriteBufferError > { let partition = i32 :: try_from ( sequencer_id ) ? ; // truncate milliseconds from timestamps because that's what Kafka supports let now = operation . meta ( ) . producer_ts ( ) . unwrap_or_else ( | | self . time_provider . now ( ) ) ; let timestamp_millis = now . date_time ( ) . timestamp_millis ( ) ; let timestamp = Time :: from_timestamp_millis ( timestamp_millis ) ; let headers = IoxHeaders :: new ( ContentType :: Protobuf , operation . meta ( ) . span_context ( ) . cloned ( ) , ) ; let mut buf = Vec :: new ( ) ; crate :: codec :: encode_operation ( & self . database_name , operation , & mut buf ) ? ; // This type annotation is necessary because `FutureRecord` is generic over key type, but // key is optional and we're not setting a key. `String` is arbitrary. let record : FutureRecord < ' _ , String , _ > = FutureRecord :: to ( & self . database_name ) . payload ( & buf ) . partition ( partition ) . timestamp ( timestamp_millis ) . headers ( ( & headers ) . into ( ) ) ; let kafka_write_size = estimate_message_size ( record . payload . map ( | v | v . as_ref ( ) ) , record . key . map ( | s | s . as_bytes ( ) ) , record . headers . as_ref ( ) , ) ; debug ! ( db_name self database_name partition size buf len ( ) " " ) ; let ( partition , offset ) = self . producer . send ( record , Timeout :: Never ) . await . map_err ( | ( e , _owned_message ) | Box :: new ( e ) ) ? ; debug ! ( db_name self database_name offset partition size buf len ( ) " " ) ; Ok ( DmlMeta :: sequenced ( Sequence :: new ( partition . try_into ( ) ? , offset . try_into ( ) ? ) , timestamp , operation . meta ( ) . span_context ( ) . cloned ( ) , kafka_write_size , ) ) }
Send an `Entry` to Kafka using the sequencer ID as a partition.
Send an ` Entry ` to Kafka using the sequencer ID as a partition .
&self
sequencer_id
u32
operation
&DmlOperation
&self
sequencer_id
u32
operation
&DmlOperation