code: private void moveToNextPartition() {
      if (this.currentPartitionIdx == INITIAL_PARTITION_IDX) {
        LOG.info("Pulling topic " + this.topicName);
        this.currentPartitionIdx = 0;
      } else {
        updateStatisticsForCurrentPartition();
        this.currentPartitionIdx++;
        this.currentPartitionRecordCount = 0;
        this.currentPartitionTotalSize = 0;
        this.currentPartitionDecodeRecordTime = 0;
        this.currentPartitionFetchMessageBufferTime = 0;
        this.currentPartitionReadRecordTime = 0;
        this.currentPartitionLastSuccessfulRecord = null;
      }
  
      this.messageIterator = null;
      if (this.currentPartitionIdx < this.partitions.size()) {
        LOG.info(String.format("Pulling partition %s from offset %d to %d, range=%d", this.getCurrentPartition(),
            this.nextWatermark.get(this.currentPartitionIdx), this.highWatermark.get(this.currentPartitionIdx),
            this.highWatermark.get(this.currentPartitionIdx) - this.nextWatermark.get(this.currentPartitionIdx)));
        switchMetricContextToCurrentPartition();
      }
  
      if (!allPartitionsFinished()) {
        this.startFetchEpochTime.put(this.getCurrentPartition(), System.currentTimeMillis());
      }
    }
docs: Record the avg time per record for the current partition, then increment this.currentPartitionIdx,
  and switch metric context to the new partition.