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 ( ) ) ; } }
Record the avg time per record for the current partition then increment this . currentPartitionIdx and switch metric context to the new partition .