Understanding Kafka Consumer Rebalancing and Partition Assignment in Scala

The article describes the internal mechanics of Kafka's high‑level consumer rebalancing, focusing on how partitions are assigned to consumer threads and how fetcher threads are managed during a rebalance.

3) PartitionAssignor – The assignor sorts available partitions and consumer threads, distributes partitions evenly across threads, and gives any remainder to the first few threads. An example with two consumers each having two threads and five partitions results in the mapping:

p0 -> C1-0, p1 -> C1-0, p2 -> C1-1, p3 -> C2-0, p4 -> C2-1

The Scala implementation of RangeAssignor shows the calculation of nPartsPerConsumer , nConsumersWithExtraPart , and the loop that assigns each partition to a specific consumerThreadId .

4) PartitionAssignment → PartitionTopicInfo – After partitionAssignor.assign returns a global assignment map, the code creates an AssignmentContext , extracts the current consumer's assignment, and builds a topicRegistry that maps topic → (partition → PartitionTopicInfo) . Offsets are fetched for each TopicAndPartition and used to construct PartitionTopicInfo objects containing the partition, topic, queue, consumed offset, fetched offset, and configuration parameters.

private def addPartitionTopicInfo(currentTopicRegistry: Pool[String, Pool[Int, PartitionTopicInfo]],
    partition: Int, topic: String, offset: Long, consumerThreadId: ConsumerThreadId) {
  val partTopicInfoMap = currentTopicRegistry.getAndMaybePut(topic)
  val queue = topicThreadIdAndQueues.get((topic, consumerThreadId))
  val consumedOffset = new AtomicLong(offset)
  val fetchedOffset = new AtomicLong(offset)
  val partTopicInfo = new PartitionTopicInfo(topic,partition,queue,consumedOffset,fetchedOffset,
    new AtomicInteger(config.fetchMessageMaxBytes), config.clientId)
  partTopicInfoMap.put(partition, partTopicInfo)
  checkpointedZkOffsets.put(TopicAndPartition(topic, partition), offset)
}

5) updateFetcher & closeFetchers – The updateFetcher method gathers all PartitionTopicInfo objects from the topicRegistry and starts fetcher connections for the current consumer. The closeFetchers method clears queues that may no longer belong to the consumer after rebalance, stops fetcher threads, clears remaining data, and commits offsets if auto‑commit is enabled.

private def closeFetchers(cluster: Cluster, messageStreams: Map[String,List[KafkaStream[_,_]]],
    relevantTopicThreadIdsMap: Map[String, Set[ConsumerThreadId]]) {
  val queuesTobeCleared = topicThreadIdAndQueues.filter(q => relevantTopicThreadIdsMap.contains(q._1._1)).map(q => q._2)
  closeFetchersForQueues(cluster, messageStreams, queuesTobeCleared)
}

ConsumerFetcherManager – Manages all fetcher threads for a consumer. When startConnections is called, it builds a partitionMap from the provided PartitionTopicInfo objects, registers the cluster, and tracks partitions without a leader. A separate LeaderFinderThread resolves leaders and updates the maps.

Summary of the high‑level rebalance flow – Each consumer registers its ID in Zookeeper, sets watches on consumer‑group and broker nodes, and triggers a rebalance via ZKRebalancerListener.rebalance . All consumers in the group rebalance simultaneously, which can cause herd effect, split‑brain scenarios, and uncontrollable adjustment results.

The article also lists references to the original blog posts and notes the author’s contact information.