code: @Override
  	protected TopicAndPartition createKafkaPartitionHandle(KafkaTopicPartition partition) {
  		return new TopicAndPartition(partition.getTopic(), partition.getPartition());
  	}
docs: ------------------------------------------------------------------------