code: private void registerOffsetMetrics(
  			MetricGroup consumerMetricGroup,
  			List<KafkaTopicPartitionState<KPH>> partitionOffsetStates) {
  
  		for (KafkaTopicPartitionState<KPH> ktp : partitionOffsetStates) {
  			MetricGroup topicPartitionGroup = consumerMetricGroup
  				.addGroup(OFFSETS_BY_TOPIC_METRICS_GROUP, ktp.getTopic())
  				.addGroup(OFFSETS_BY_PARTITION_METRICS_GROUP, Integer.toString(ktp.getPartition()));
  
  			topicPartitionGroup.gauge(CURRENT_OFFSETS_METRICS_GAUGE, new OffsetGauge(ktp, OffsetGaugeType.CURRENT_OFFSET));
  			topicPartitionGroup.gauge(COMMITTED_OFFSETS_METRICS_GAUGE, new OffsetGauge(ktp, OffsetGaugeType.COMMITTED_OFFSET));
  
  			legacyCurrentOffsetsMetricGroup.gauge(getLegacyOffsetsMetricsGaugeName(ktp), new OffsetGauge(ktp, OffsetGaugeType.CURRENT_OFFSET));
  			legacyCommittedOffsetsMetricGroup.gauge(getLegacyOffsetsMetricsGaugeName(ktp), new OffsetGauge(ktp, OffsetGaugeType.COMMITTED_OFFSET));
  		}
  	}
docs: For each partition, register a new metric group to expose current offsets and committed offsets.
  Per-partition metric groups can be scoped by user variables {@link KafkaConsumerMetricConstants#OFFSETS_BY_TOPIC_METRICS_GROUP}
  and {@link KafkaConsumerMetricConstants#OFFSETS_BY_PARTITION_METRICS_GROUP}.
  
  <p>Note: this method also registers gauges for deprecated offset metrics, to maintain backwards compatibility.
  
  @param consumerMetricGroup The consumer metric group
  @param partitionOffsetStates The partition offset state holders, whose values will be used to update metrics