pub fn new(
        inner: T,
        watermark_fetcher: F,
        kafka_topic_name: String,
        kafka_partition: KafkaPartition,
        metrics: &metric::Registry,
    ) -> Self {
        let attr = Attributes::from([
            ("kafka_partition", kafka_partition.to_string().into()),
            ("kafka_topic", kafka_topic_name.into()),
        ]);

        let write_buffer_bytes_read = metrics
            .register_metric::<U64Counter>(
                "write_buffer_read_bytes",
                "Total number of bytes read from sequencer",
            )
            .recorder(attr.clone());
        let write_buffer_last_sequence_number = metrics
            .register_metric::<U64Gauge>(
                "write_buffer_last_sequence_number",
                "Last consumed sequence number (e.g. Kafka offset)",
            )
            .recorder(attr.clone());
        let write_buffer_sequence_number_lag = metrics.register_metric::<U64Gauge>(
            "write_buffer_sequence_number_lag",
            "The difference between the the last sequence number available (e.g. Kafka offset) and (= minus) last consumed sequence number",
        ).recorder(attr.clone());
        let write_buffer_last_ingest_ts = metrics
            .register_metric::<U64Gauge>(
                "write_buffer_last_ingest_ts",
                "Last seen ingest timestamp as unix timestamp in nanoseconds",
            )
            .recorder(attr.clone());

        // The buckets for the op apply histogram
        let buckets = || {
            U64HistogramOptions::new([
                5,
                10,
                20,
                40,
                80,
                160,
                320,
                640,
                1280,
                2560,
                5120,
                10240,
                20480,
                u64::MAX,
            ])
        };

        let op_apply = metrics.register_metric_with_options::<U64Histogram, _>(
            "ingester_op_apply_duration_ms",
            "The duration of time taken to process an operation read from the sequencer",
            buckets,
        );
        let op_apply_success_ms = op_apply.recorder({
            let mut attr = attr.clone();
            attr.insert("result", "success");
            attr
        });
        let op_apply_error_ms = op_apply.recorder({
            let mut attr = attr;
            attr.insert("result", "error");
            attr
        });

        Self {
            inner,
            watermark_fetcher,
            sequencer_id: kafka_partition.get(),

            op_apply_success_ms,
            op_apply_error_ms,

            write_buffer_bytes_read,
            write_buffer_last_sequence_number,
            write_buffer_sequence_number_lag,
            write_buffer_last_ingest_ts,
            time_provider: SystemProvider::default(),
        }
    }
/// Construct a new [`SinkInstrumentation`] layer that decorates `inner`
/// with logic that records metrics from the [`DmlOperation`] instances it
/// observes.
///
/// The current high watermark is read from `watermark_fetcher` and used to
/// derive some metric values (such as lag). This impl is tolerant of
/// cached/stale watermark values being returned by `watermark_fetcher`.


The current high watermark is read from `watermark_fetcher` and used to
derive some metric values (such as lag). This impl is tolerant of
cached/stale watermark values being returned by `watermark_fetcher`.
The current high watermark is read from ` watermark_fetcher ` and used to derive some metric values ( such as lag ) . This impl is tolerant of cached / stale watermark values being returned by ` watermark_fetcher ` .
pub fn new(
        inner: T,
        watermark_fetcher: F,
        kafka_topic_name: String,
        kafka_partition: KafkaPartition,
        metrics: &metric::Registry,
    ) -> Self {
        let attr = Attributes::from([
            ("kafka_partition", kafka_partition.to_string().into()),
            ("kafka_topic", kafka_topic_name.into()),
        ]);
        let write_buffer_bytes_read = metrics
            .register_metric::<U64Counter>(
                "write_buffer_read_bytes",
                "Total number of bytes read from sequencer",
            )
            .recorder(attr.clone());
        let write_buffer_last_sequence_number = metrics
            .register_metric::<U64Gauge>(
                "write_buffer_last_sequence_number",
                "Last consumed sequence number (e.g. Kafka offset)",
            )
            .recorder(attr.clone());
        let write_buffer_sequence_number_lag = metrics.register_metric::<U64Gauge>(
            "write_buffer_sequence_number_lag",
            "The difference between the the last sequence number available (e.g. Kafka offset) and (= minus) last consumed sequence number",
        ).recorder(attr.clone());
        let write_buffer_last_ingest_ts = metrics
            .register_metric::<U64Gauge>(
                "write_buffer_last_ingest_ts",
                "Last seen ingest timestamp as unix timestamp in nanoseconds",
            )
            .recorder(attr.clone());
        // The buckets for the op apply histogram
        let buckets = || {
            U64HistogramOptions::new([
                5,
                10,
                20,
                40,
                80,
                160,
                320,
                640,
                1280,
                2560,
                5120,
                10240,
                20480,
                u64::MAX,
            ])
        };
        let op_apply = metrics.register_metric_with_options::<U64Histogram, _>(
            "ingester_op_apply_duration_ms",
            "The duration of time taken to process an operation read from the sequencer",
            buckets,
        );
        let op_apply_success_ms = op_apply.recorder({
            let mut attr = attr.clone();
            attr.insert("result", "success");
            attr
        });
        let op_apply_error_ms = op_apply.recorder({
            let mut attr = attr;
            attr.insert("result", "error");
            attr
        });
        Self {
            inner,
            watermark_fetcher,
            sequencer_id: kafka_partition.get(),
            op_apply_success_ms,
            op_apply_error_ms,
            write_buffer_bytes_read,
            write_buffer_last_sequence_number,
            write_buffer_sequence_number_lag,
            write_buffer_last_ingest_ts,
            time_provider: SystemProvider::default(),
        }
    }
pub fn new ( inner : T , watermark_fetcher : F , kafka_topic_name : String , kafka_partition : KafkaPartition , metrics : & metric :: Registry , ) -> Self { let attr = Attributes :: from ( [ ( " " , kafka_partition . to_string ( ) . into ( ) ) , ( " " , kafka_topic_name . into ( ) ) , ] ) ; let write_buffer_bytes_read = metrics . register_metric :: < U64Counter > ( " " , " " , ) . recorder ( attr . clone ( ) ) ; let write_buffer_last_sequence_number = metrics . register_metric :: < U64Gauge > ( " " , " " , ) . recorder ( attr . clone ( ) ) ; let write_buffer_sequence_number_lag = metrics . register_metric :: < U64Gauge > ( " " , " " , ) . recorder ( attr . clone ( ) ) ; let write_buffer_last_ingest_ts = metrics . register_metric :: < U64Gauge > ( " " , " " , ) . recorder ( attr . clone ( ) ) ; // The buckets for the op apply histogram let buckets = | | { U64HistogramOptions :: new ( [ 5 , 10 , 20 , 40 , 80 , 160 , 320 , 640 , 1280 , 2560 , 5120 , 10240 , 20480 , u64 :: MAX , ] ) } ; let op_apply = metrics . register_metric_with_options :: < U64Histogram , _ > ( " " , " " , buckets , ) ; let op_apply_success_ms = op_apply . recorder ( { let mut attr = attr . clone ( ) ; attr . insert ( " " , " " ) ; attr } ) ; let op_apply_error_ms = op_apply . recorder ( { let mut attr = attr ; attr . insert ( " " , " " ) ; attr } ) ; Self { inner , watermark_fetcher , sequencer_id : kafka_partition . get ( ) , op_apply_success_ms , op_apply_error_ms , write_buffer_bytes_read , write_buffer_last_sequence_number , write_buffer_sequence_number_lag , write_buffer_last_ingest_ts , time_provider : SystemProvider :: default ( ) , } }
Construct a new [`SinkInstrumentation`] layer that decorates `inner`
with logic that records metrics from the [`DmlOperation`] instances it
observes.
Construct a new [ ` SinkInstrumentation ` ] layer that decorates ` inner ` with logic that records metrics from the [ ` DmlOperation ` ] instances it observes .
inner
T
watermark_fetcher
F
kafka_topic_name
String
kafka_partition
KafkaPartition
metrics
&metric::Registry
inner
T
watermark_fetcher
F
kafka_topic_name
String
kafka_partition
KafkaPartition
metrics
&metric::Registry