You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: kafka.rs
    path of file: ./repos/materialize/src/storage/src/metrics/sink
    
    the code of the file until where you have to start completion: // Copyright Materialize, Inc. and contributors. All rights reserved.
//
// Use of this software is governed by the Business Source License
// included in the LICENSE file.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0.

//! Metrics for kafka sinks.

use mz_ore::metric;
use mz_ore::metrics::{DeleteOnDropGauge, GaugeVecExt, IntGaugeVec, MetricsRegistry, UIntGaugeVec};
use mz_repr::GlobalId;
use prometheus::core::{AtomicI64, AtomicU64};

/// Definitions for librdkafka produced metrics used in sinks.
#[derive(Clone, Debug)]
pub(crate) struct KafkaSinkMetricDefs {
    /// The current number of messages in producer queues.
    pub rdkafka_msg_cnt: UIntGaugeVec,
    /// The current total size of messages in producer queues.
    pub rdkafka_msg_size: UIntGaugeVec,
    /// The total number of messages transmitted (produced) to brokers.
    pub rdkafka_txmsgs: IntGaugeVec,
    /// The total number of bytes transmitted (produced) to brokers.
    pub rdkafka_txmsg_bytes: IntGaugeVec,
    /// The total number of requests sent to brokers.
    pub rdkafka_tx: IntGaugeVec,
    /// The total number of bytes transmitted to brokers.
    pub rdkafka_tx_bytes: IntGaugeVec,
    /// The number of requests awaiting transmission across all brokers.
    pub rdkafka_outbuf_cnt: IntGaugeVec,
    /// The number of messages awaiting transmission across all brokers.
    pub rdkafka_outbuf_msg_cnt: IntGaugeVec,
    /// The number of requests in-flight across all brokers that are awaiting a response.
    pub rdkafka_waitresp_cnt: IntGaugeVec,
    /// The number of messages in-flight across all brokers that are awaiting a response.
    pub rdkafka_waitresp_msg_cnt: IntGaugeVec,
    /// The total number of transmission errors across all brokers.
    pub rdkafka_txerrs: UIntGaugeVec,
    /// The total number of request retries across all brokers.
    
    pub fn new(defs: &KafkaSinkMetricDefs, sink_id: GlobalId) -> Self {
        let labels = &[sink_id.to_string()];
        Self {
            rdkafka_msg_cnt: defs
                .rdkafka_msg_cnt
                .get_delete_on_drop_gauge(labels.to_vec()),
            rdkafka_msg_size: defs
                .rdkafka_msg_size
                .get_delete_on_drop_gauge(labels.to_vec()),
            rdkafka_txmsgs: defs
                .rdkafka_txmsgs
                .get_delete_on_drop_gauge(labels.to_vec()),
            rdkafka_txmsg_bytes: defs
                .rdkafka_txmsg_bytes
                .get_delete_on_drop_gauge(labels.to_vec()),
            rdkafka_tx: defs.rdkafka_tx.get_delete_on_drop_gauge(labels.to_vec()),
            rdkafka_tx_bytes: defs
                .rdkafka_tx_bytes
                .get_delete_on_drop_gauge(labels.to_vec()),
            rdkafka_outbuf_cnt: defs
                .rdkafka_outbuf_cnt
                .get_delete_on_drop_gauge(labels.to_vec()),
            rdkafka_outbuf_msg_cnt: defs
                .rdkafka_outbuf_msg_cnt
                .get_delete_on_drop_gauge(labels.to_vec()),
            rdkafka_waitresp_cnt: defs
                .rdkafka_waitresp_cnt
                .get_delete_on_drop_gauge(labels.to_vec()),
            rdkafka_waitresp_msg_cnt: defs
                .rdkafka_waitresp_msg_cnt
                .get_delete_on_drop_gauge(labels.to_vec()),
            rdkafka_txerrs: defs
                .rdkafka_txerrs
                .get_delete_on_drop_gauge(labels.to_vec()),
            rdkafka_txretries: defs
                .rdkafka_txretries
                .get_delete_on_drop_gauge(labels.to_vec()),
            rdkafka_req_timeouts: defs
                .rdkafka_req_timeouts
                .get_delete_on_drop_gauge(labels.to_vec()),
            rdkafka_connects: defs
                .rdkafka_connects
                .get_delete_on_drop_gauge(labels.to_vec()),
            rdkafka_disconnects: defs
                .rdkafka_disconnects
                .get_delete_on_drop_gauge(labels.to_vec()),
        }
    }