Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

KAFKA-9987: optimize sticky assignment algorithm for same-subscription case #8668

Merged
merged 23 commits into from
Jun 1, 2020
Merged

KAFKA-9987: optimize sticky assignment algorithm for same-subscription case #8668

Show file tree
Hide file tree
Changes from 8 commits
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
dbac63d
refactor AbstractStickyAssignor and implement optimized algorithm
May 14, 2020
2b2a011
rough benchmarks
May 14, 2020
a812d34
remove areSubscriptionsIdentical when we know they are not
May 29, 2020
50486af
simplify
May 29, 2020
afbed93
renaming
May 29, 2020
52c3285
initialize movements
May 29, 2020
f7a6fa4
trim by subscription
May 29, 2020
f3a074f
fixing tests
May 29, 2020
ff28bad
remove unused imports
May 29, 2020
9a11976
improve data parallelism
May 29, 2020
63e797f
use Optional.empty
May 29, 2020
e397565
apply timeout and increase number of topics
May 29, 2020
e5eeb7d
add comment and fix checkstyle
May 29, 2020
18b13cb
remove more unused code
May 29, 2020
07267a2
simplify a bit
May 29, 2020
f6ce366
remove unnecessary check and fix test to verify valid state
May 29, 2020
d9c93d0
filter by cluster topics
May 29, 2020
e9b243a
simplify owned partition aassignemnt
May 29, 2020
ad1f63d
can reduce further
May 29, 2020
28d13ba
condition can be tighter
May 29, 2020
7ec6131
optimize by tracking reassigned partitions
May 29, 2020
6210a75
sclae to 1million
May 30, 2020
e25a323
checkstyle
May 30, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 5 additions & 7 deletions clients/src/main/java/org/apache/kafka/clients/consumer/CooperativeStickyAssignor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,22 +56,20 @@ public List<RebalanceProtocol> supportedProtocols() {

@Override
protected MemberData memberData(Subscription subscription) {
return new MemberData(subscription.ownedPartitions(), Optional.empty());
return new MemberData(subscription.ownedPartitions(), Optional.of(1));
}

@Override
public Map<String, List<TopicPartition>> assign(Map<String, Integer> partitionsPerTopic,
Map<String, Subscription> subscriptions) {

final Map<String, List<TopicPartition>> assignments = super.assign(partitionsPerTopic, subscriptions);
Map<String, List<TopicPartition>> assignments = super.assign(partitionsPerTopic, subscriptions);
adjustAssignment(subscriptions, assignments);
return assignments;
}

// Following the cooperative rebalancing protocol requires removing partitions that must first be revoked from the assignment
private void adjustAssignment(final Map<String, Subscription> subscriptions,
final Map<String, List<TopicPartition>> assignments) {

private void adjustAssignment(Map<String, Subscription> subscriptions,
Map<String, List<TopicPartition>> assignments) {
Map<TopicPartition, String> allAddedPartitions = new HashMap<>();
Set<TopicPartition> allRevokedPartitions = new HashSet<>();

Expand All @@ -87,7 +85,7 @@ private void adjustAssignment(final Map<String, Subscription> subscriptions,
allAddedPartitions.put(tp, consumer);
}

final Set<TopicPartition> revokedPartitions = new HashSet<>(ownedPartitions);
Set<TopicPartition> revokedPartitions = new HashSet<>(ownedPartitions);
revokedPartitions.removeAll(assignedPartitions);
allRevokedPartitions.addAll(revokedPartitions);
}
Expand Down
Loading