code: public static int assign(KafkaTopicPartition partition, int numParallelSubtasks) {
  		int startIndex = ((partition.getTopic().hashCode() * 31) & 0x7FFFFFFF) % numParallelSubtasks;
  
  		// here, the assumption is that the id of Kafka partitions are always ascending
  		// starting from 0, and therefore can be used directly as the offset clockwise from the start index
  		return (startIndex + partition.getPartition()) % numParallelSubtasks;
  	}
docs: Returns the index of the target subtask that a specific Kafka partition should be
  assigned to.
  
  <p>The resulting distribution of partitions of a single topic has the following contract:
  <ul>
  <li>1. Uniformly distributed across subtasks</li>
  <li>2. Partitions are round-robin distributed (strictly clockwise w.r.t. ascending
  subtask indices) by using the partition id as the offset from a starting index
  (i.e., the index of the subtask which partition 0 of the topic will be assigned to,
  determined using the topic name).</li>
  </ul>
  
  <p>The above contract is crucial and cannot be broken. Consumer subtasks rely on this
  contract to locally filter out partitions that it should not subscribe to, guaranteeing
  that all partitions of a single topic will always be assigned to some subtask in a
  uniformly distributed manner.
  
  @param partition the Kafka partition
  @param numParallelSubtasks total number of parallel subtasks
  
  @return index of the target subtask that the Kafka partition should be assigned to.