Package org.apache.kafka.clients.consumer

Class RangeAssignor

java.lang.Object
org.apache.kafka.clients.consumer.internals.AbstractPartitionAssignor
org.apache.kafka.clients.consumer.RangeAssignor
All Implemented Interfaces:
ConsumerPartitionAssignor
public class RangeAssignor extends org.apache.kafka.clients.consumer.internals.AbstractPartitionAssignor

The range assignor works on a per-topic basis. For each topic, we lay out the available partitions in numeric order and the consumers in lexicographic order. We then divide the number of partitions by the total number of consumers to determine the number of partitions to assign to each consumer. If it does not evenly divide, then the first few consumers will have one extra partition.

For example, suppose there are two consumers C0 and C1, two topics t0 and t1, and each topic has 3 partitions, resulting in partitions t0p0, t0p1, t0p2, t1p0, t1p1, and t1p2.

The assignment will be:

  • C0: [t0p0, t0p1, t1p0, t1p1]
  • C1: [t0p2, t1p2]
Since the introduction of static membership, we could leverage group.instance.id to make the assignment behavior more sticky. For the above example, after one rolling bounce, group coordinator will attempt to assign new member.id towards consumers, for example C0 -> C3 C1 -> C2.

The assignment could be completely shuffled to:

  • C3 (was C0): [t0p2, t1p2] (before was [t0p0, t0p1, t1p0, t1p1])
  • C2 (was C1): [t0p0, t0p1, t1p0, t1p1] (before was [t0p2, t1p2])
The assignment change was caused by the change of member.id relative order, and can be avoided by setting the group.instance.id. Consumers will have individual instance ids I1, I2. As long as 1. Number of members remain the same across generation 2. Static members' identities persist across generation 3. Subscription pattern doesn't change for any member

The assignment will always be:

  • I0: [t0p0, t0p1, t1p0, t1p1]
  • I1: [t0p2, t1p2]

Rack-aware assignment is used if both consumer and partition replica racks are available and some partitions have replicas only on a subset of racks. We attempt to match consumer racks with partition replica racks on a best-effort basis, prioritizing balanced assignment over rack alignment. Topics with equal partition count and same set of subscribers guarantee co-partitioning by prioritizing co-partitioning over rack alignment. In this case, aligning partition replicas of these topics on the same racks will improve locality for consumers. For example, if partitions 0 of all topics have a replica on rack 'a', partition 1 on rack 'b' etc., partition 0 of all topics can be assigned to a consumer on rack 'a', partition 1 to a consumer on rack 'b' and so on.

Note that rack-aware assignment currently takes all replicas into account, including any offline replicas and replicas that are not in the ISR. This is based on the assumption that these replicas are likely to join the ISR relatively soon. Since consumers don't rebalance on ISR change, this avoids unnecessary cross-rack traffic for long durations after replicas rejoin the ISR. In the future, we may consider rebalancing when replicas are added or removed to improve consumer rack alignment.

  • Field Details

  • Constructor Details

    • RangeAssignor

      public RangeAssignor()

  • Method Details

    • name

      public String name()
      Description copied from interface: ConsumerPartitionAssignor
      Unique name for this assignor (e.g. "range" or "roundrobin" or "sticky"). Note, this is not required to be the same as the class name specified in ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG
      Returns:
      non-null unique name

    • assignPartitions

      public Map<String,List<TopicPartition>> assignPartitions(Map<String,List<PartitionInfo>> partitionsPerTopic, Map<String,ConsumerPartitionAssignor.Subscription> subscriptions)
      Performs range assignment of the specified partitions for the consumers with the provided subscriptions. If rack-awareness is enabled for one or more consumers, we perform rack-aware assignment first to assign the subset of partitions that can be aligned on racks, while retaining the same co-partitioning and per-topic balancing guarantees as non-rack-aware range assignment. The remaining partitions are assigned using standard non-rack-aware range assignment logic, which may result in mis-aligned racks.
      Overrides:
      assignPartitions in class org.apache.kafka.clients.consumer.internals.AbstractPartitionAssignor

    • assign

      public Map<String,List<TopicPartition>> assign(Map<String,Integer> partitionsPerTopic, Map<String,ConsumerPartitionAssignor.Subscription> subscriptions)
      Specified by:
      assign in class org.apache.kafka.clients.consumer.internals.AbstractPartitionAssignor