Interface ConsumerPartitionAssignor
-
- All Known Implementing Classes:
org.apache.kafka.clients.consumer.internals.AbstractPartitionAssignor
,org.apache.kafka.clients.consumer.internals.AbstractStickyAssignor
,CooperativeStickyAssignor
,RangeAssignor
,RoundRobinAssignor
,StickyAssignor
public interface ConsumerPartitionAssignor
This interface is used to define custom partition assignment for use inKafkaConsumer
. Members of the consumer group subscribe to the topics they are interested in and forward their subscriptions to a Kafka broker serving as the group coordinator. The coordinator selects one member to perform the group assignment and propagates the subscriptions of all members to it. Thenassign(Cluster, GroupSubscription)
is called to perform the assignment and the results are forwarded back to each respective members In some cases, it is useful to forward additional metadata to the assignor in order to make assignment decisions. For this, you can overridesubscriptionUserData(Set)
and provide custom userData in the returned Subscription. For example, to have a rack-aware assignor, an implementation can use this user data to forward the rackId belonging to each member.
-
-
Nested Class Summary
Nested Classes Modifier and Type Interface Description static class
ConsumerPartitionAssignor.Assignment
static class
ConsumerPartitionAssignor.GroupAssignment
static class
ConsumerPartitionAssignor.GroupSubscription
static class
ConsumerPartitionAssignor.RebalanceProtocol
The rebalance protocol defines partition assignment and revocation semantics.static class
ConsumerPartitionAssignor.Subscription
-
Method Summary
All Methods Instance Methods Abstract Methods Default Methods Modifier and Type Method Description ConsumerPartitionAssignor.GroupAssignment
assign(Cluster metadata, ConsumerPartitionAssignor.GroupSubscription groupSubscription)
Perform the group assignment given the member subscriptions and current cluster metadata.java.lang.String
name()
Unique name for this assignor (e.g.default void
onAssignment(ConsumerPartitionAssignor.Assignment assignment, ConsumerGroupMetadata metadata)
Callback which is invoked when a group member receives its assignment from the leader.default java.nio.ByteBuffer
subscriptionUserData(java.util.Set<java.lang.String> topics)
Return serialized data that will be included in theConsumerPartitionAssignor.Subscription
sent to the leader and can be leveraged inassign(Cluster, GroupSubscription)
((e.g.default java.util.List<ConsumerPartitionAssignor.RebalanceProtocol>
supportedProtocols()
Indicate which rebalance protocol this assignor works with; By default it should always work withConsumerPartitionAssignor.RebalanceProtocol.EAGER
.default short
version()
Return the version of the assignor which indicates how the user metadata encodings and the assignment algorithm gets evolved.
-
-
-
Method Detail
-
subscriptionUserData
default java.nio.ByteBuffer subscriptionUserData(java.util.Set<java.lang.String> topics)
Return serialized data that will be included in theConsumerPartitionAssignor.Subscription
sent to the leader and can be leveraged inassign(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
-
assign
ConsumerPartitionAssignor.GroupAssignment assign(Cluster metadata, ConsumerPartitionAssignor.GroupSubscription groupSubscription)
Perform the group assignment given the member subscriptions and current cluster metadata.- Parameters:
metadata
- Current topic/broker metadata known by consumergroupSubscription
- Subscriptions from all members including metadata provided throughsubscriptionUserData(Set)
- Returns:
- A map from the members to their respective assignments. This should have one entry for each member in the input subscription map.
-
onAssignment
default void onAssignment(ConsumerPartitionAssignor.Assignment assignment, ConsumerGroupMetadata metadata)
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 inassign(Cluster, GroupSubscription)
metadata
- Additional metadata on the consumer (optional)
-
supportedProtocols
default java.util.List<ConsumerPartitionAssignor.RebalanceProtocol> supportedProtocols()
Indicate which rebalance protocol this assignor works with; By default it should always work withConsumerPartitionAssignor.RebalanceProtocol.EAGER
.
-
version
default short version()
Return the version of the assignor which indicates how the user metadata encodings and the assignment algorithm gets evolved.
-
name
java.lang.String name()
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
-
-