code: def __init__(self, metrics, prefix):
      self.metrics = metrics
      self.group_name = "%s-fetch-manager-metrics" % (prefix,)
  
      self.bytes_fetched = metrics.sensor("bytes-fetched")
      self.bytes_fetched.add(
          metrics.metric_name(
              "fetch-size-avg",
              self.group_name,
              "The average number of bytes fetched per request",
          ),
          Avg(),
      )
      self.bytes_fetched.add(
          metrics.metric_name(
              "fetch-size-max",
              self.group_name,
              "The maximum number of bytes fetched per request",
          ),
          Max(),
      )
      self.bytes_fetched.add(
          metrics.metric_name(
              "bytes-consumed-rate",
              self.group_name,
              "The average number of bytes consumed per second",
          ),
          Rate(),
      )
  
      self.records_fetched = self.metrics.sensor("records-fetched")
      self.records_fetched.add(
          metrics.metric_name(
              "records-per-request-avg",
              self.group_name,
              "The average number of records in each request",
          ),
          Avg(),
      )
      self.records_fetched.add(
          metrics.metric_name(
              "records-consumed-rate",
              self.group_name,
              "The average number of records consumed per second",
          ),
          Rate(),
      )
  
      self.fetch_latency = metrics.sensor("fetch-latency")
      self.fetch_latency.add(
          metrics.metric_name(
              "fetch-latency-avg",
              self.group_name,
              "The average time taken for a fetch request.",
          ),
          Avg(),
      )
      self.fetch_latency.add(
          metrics.metric_name(
              "fetch-latency-max",
              self.group_name,
              "The max time taken for any fetch request.",
          ),
          Max(),
      )
      self.fetch_latency.add(
          metrics.metric_name(
              "fetch-rate",
              self.group_name,
              "The number of fetch requests per second.",
          ),
          Rate(sampled_stat=Count()),
      )
  
      self.records_fetch_lag = metrics.sensor("records-lag")
      self.records_fetch_lag.add(
          metrics.metric_name(
              "records-lag-max",
              self.group_name,
              "The maximum lag in terms of number of records for any partition in self window",
          ),
          Max(),
      )
  
      self.fetch_throttle_time_sensor = metrics.sensor(
          "fetch-throttle-time"
      )
      self.fetch_throttle_time_sensor.add(
          metrics.metric_name(
              "fetch-throttle-time-avg",
              self.group_name,
              "The average throttle time in ms",
          ),
          Avg(),
      )
      self.fetch_throttle_time_sensor.add(
          metrics.metric_name(
              "fetch-throttle-time-max",
              self.group_name,
              "The maximum throttle time in ms",
          ),
          Max(),
      )
  
imports: from __future__ import absolute_import
  import collections
  import copy
  import logging
  import random
  import sys
  import time
  from kafka.vendor import six
  import kafka.errors
  from kafka.future import Future
  from kafka.metrics.stats import Avg, Count, Max, Rate
  from kafka.protocol.fetch import FetchRequest
  from kafka.protocol.offset import OffsetRequest, OffsetResetStrategy, UNKNOWN_OFFSET
  from kafka.record import MemoryRecords
  from kafka.serializer import Deserializer
  from kafka.structs import TopicPartition, OffsetAndTimestamp
label_desc: Use image node_id 1 to create a new FetchManagerMetrics object from inherited base classes: object with example: obj = FetchManagerMetrics(metrics, prefix)