Class StickyAssignor
- java.lang.Object
-
- org.apache.kafka.clients.consumer.internals.AbstractPartitionAssignor
-
- org.apache.kafka.clients.consumer.internals.AbstractStickyAssignor
-
- org.apache.kafka.clients.consumer.StickyAssignor
-
- All Implemented Interfaces:
ConsumerPartitionAssignor
public class StickyAssignor extends org.apache.kafka.clients.consumer.internals.AbstractStickyAssignor
The sticky assignor serves two purposes. First, it guarantees an assignment that is as balanced as possible, meaning either:
- the numbers of topic partitions assigned to consumers differ by at most one; or
- each consumer that has 2+ fewer topic partitions than some other consumer cannot get any of those topic partitions transferred to it.
Starting fresh it would work by distributing the partitions over consumers as evenly as possible. Even though this may sound similar to how round robin assignor works, the second example below shows that it is not. During a reassignment it would perform the reassignment in such a way that in the new assignment
- topic partitions are still distributed as evenly as possible, and
- topic partitions stay with their previously assigned consumers as much as possible.
Example 1. Suppose there are three consumers
C0
,C1
,C2
, four topicst0,
t1
,t2
,t3
, and each topic has 2 partitions, resulting in partitionst0p0
,t0p1
,t1p0
,t1p1
,t2p0
,t2p1
,t3p0
,t3p1
. Each consumer is subscribed to all three topics. The assignment with both sticky and round robin assignors will be:C0: [t0p0, t1p1, t3p0]
C1: [t0p1, t2p0, t3p1]
C2: [t1p0, t2p1]
C1
is removed and a reassignment is about to happen. The round robin assignor would produce:C0: [t0p0, t1p0, t2p0, t3p0]
C2: [t0p1, t1p1, t2p1, t3p1]
C0 [t0p0, t1p1, t3p0, t2p0]
C2 [t1p0, t2p1, t0p1, t3p1]
Example 2. There are three consumers
C0
,C1
,C2
, and three topicst0
,t1
,t2
, with 1, 2, and 3 partitions respectively. Therefore, the partitions aret0p0
,t1p0
,t1p1
,t2p0
,t2p1
,t2p2
.C0
is subscribed tot0
;C1
is subscribed tot0
,t1
; andC2
is subscribed tot0
,t1
,t2
. The round robin assignor would come up with the following assignment:C0 [t0p0]
C1 [t1p0]
C2 [t1p1, t2p0, t2p1, t2p2]
C0 [t0p0]
C1 [t1p0, t1p1]
C2 [t2p0, t2p1, t2p2]
C0
is removed, these two assignors would produce the following assignments. Round Robin (preserves 3 partition assignments):C1 [t0p0, t1p1]
C2 [t1p0, t2p0, t2p1, t2p2]
C1 [t1p0, t1p1, t0p0]
C2 [t2p0, t2p1, t2p2]
Impact on
The sticky assignment strategy can provide some optimization to those consumers that have some partition cleanup code in theirConsumerRebalanceListener
onPartitionsRevoked()
callback listeners. The cleanup code is placed in that callback listener because the consumer has no assumption or hope of preserving any of its assigned partitions after a rebalance when it is using range or round robin assignor. The listener code would look like this:class TheOldRebalanceListener implements ConsumerRebalanceListener { void onPartitionsRevoked(Collection<TopicPartition> partitions) { for (TopicPartition partition: partitions) { commitOffsets(partition); cleanupState(partition); } } void onPartitionsAssigned(Collection<TopicPartition> partitions) { for (TopicPartition partition: partitions) { initializeState(partition); initializeOffset(partition); } } }
onPartitionsRevoked()
listener, but they can be more efficient and make a note of their partitions before and after the rebalance, and do the cleanup after the rebalance only on the partitions they have lost (which is normally not a lot). The code snippet below clarifies this point:class TheNewRebalanceListener implements ConsumerRebalanceListener { Collection<TopicPartition> lastAssignment = Collections.emptyList(); void onPartitionsRevoked(Collection<TopicPartition> partitions) { for (TopicPartition partition: partitions) commitOffsets(partition); } void onPartitionsAssigned(Collection<TopicPartition> assignment) { for (TopicPartition partition: difference(lastAssignment, assignment)) cleanupState(partition); for (TopicPartition partition: difference(assignment, lastAssignment)) initializeState(partition); for (TopicPartition partition: assignment) initializeOffset(partition); this.lastAssignment = assignment; } }
consumer.subscribe(topics, new TheNewRebalanceListener());
Note that you can leverage theCooperativeStickyAssignor
so that only partitions which are being reassigned to another consumer will be revoked. That is the preferred assignor for newer cluster. SeeConsumerPartitionAssignor.RebalanceProtocol
for a detailed explanation of cooperative rebalancing.
-
-
Nested Class Summary
-
Nested classes/interfaces inherited from class org.apache.kafka.clients.consumer.internals.AbstractStickyAssignor
org.apache.kafka.clients.consumer.internals.AbstractStickyAssignor.MemberData
-
Nested classes/interfaces inherited from class org.apache.kafka.clients.consumer.internals.AbstractPartitionAssignor
org.apache.kafka.clients.consumer.internals.AbstractPartitionAssignor.MemberInfo
-
Nested classes/interfaces inherited from interface org.apache.kafka.clients.consumer.ConsumerPartitionAssignor
ConsumerPartitionAssignor.Assignment, ConsumerPartitionAssignor.GroupAssignment, ConsumerPartitionAssignor.GroupSubscription, ConsumerPartitionAssignor.RebalanceProtocol, ConsumerPartitionAssignor.Subscription
-
-
Constructor Summary
Constructors Constructor Description StickyAssignor()
-
Method Summary
All Methods Instance Methods Concrete Methods Modifier and Type Method Description protected org.apache.kafka.clients.consumer.internals.AbstractStickyAssignor.MemberData
memberData(ConsumerPartitionAssignor.Subscription subscription)
String
name()
Unique name for this assignor (e.g.void
onAssignment(ConsumerPartitionAssignor.Assignment assignment, ConsumerGroupMetadata metadata)
Callback which is invoked when a group member receives its assignment from the leader.ByteBuffer
subscriptionUserData(Set<String> topics)
Return serialized data that will be included in theConsumerPartitionAssignor.Subscription
sent to the leader and can be leveraged inConsumerPartitionAssignor.assign(Cluster, GroupSubscription)
((e.g.-
Methods inherited from class org.apache.kafka.clients.consumer.internals.AbstractStickyAssignor
assign, isSticky
-
Methods inherited from class org.apache.kafka.clients.consumer.internals.AbstractPartitionAssignor
assign, partitions, put
-
Methods inherited from class java.lang.Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait
-
Methods inherited from interface org.apache.kafka.clients.consumer.ConsumerPartitionAssignor
supportedProtocols, version
-
-
-
-
Method Detail
-
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 inConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG
- Returns:
- non-null unique name
-
onAssignment
public void onAssignment(ConsumerPartitionAssignor.Assignment assignment, ConsumerGroupMetadata metadata)
Description copied from interface:ConsumerPartitionAssignor
Callback which is invoked when a group member receives its assignment from the leader.- Parameters:
assignment
- The local member's assignment as provided by the leader inConsumerPartitionAssignor.assign(Cluster, GroupSubscription)
metadata
- Additional metadata on the consumer (optional)
-
subscriptionUserData
public ByteBuffer subscriptionUserData(Set<String> topics)
Description copied from interface:ConsumerPartitionAssignor
Return serialized data that will be included in theConsumerPartitionAssignor.Subscription
sent to the leader and can be leveraged inConsumerPartitionAssignor.assign(Cluster, GroupSubscription)
((e.g. local host/rack information)- Parameters:
topics
- Topics subscribed to throughKafkaConsumer.subscribe(java.util.Collection)
and variants- Returns:
- nullable subscription user data
-
memberData
protected org.apache.kafka.clients.consumer.internals.AbstractStickyAssignor.MemberData memberData(ConsumerPartitionAssignor.Subscription subscription)
- Specified by:
memberData
in classorg.apache.kafka.clients.consumer.internals.AbstractStickyAssignor
-
-