KAFKA-7026: Sticky Assignor Partition Assignment Improvement (KIP-341) #5291

vahidhashemian · 2018-06-26T01:52:46Z

In the current implementation of sticky assignor the leader does not cache the most recent calculated assignment. It relies on the fact that each consumer in the group sends its subscribed topics and also its current assignment when a rebalance occurs. This could lead to the issue described in KAFKA-7026, in which current assignment of a consumer is no longer valid and should be ignored. The solution implemented in this PR involves the leader caching the most recent assignment of each consumer, so the assignment reported by a consumer can be properly validated (and ignored if necessary) by the leader during a rebalance.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

steven-aerts

Hello Vahid,

I have some reservations on your solution. As I fear it will not work if the leader changes. (or even worse when an previous leader becomes leader again).

Maybe I am wrong, but I think a different kind of solution is needed. If you want I can take a stab at it.

steven-aerts · 2018-06-26T07:34:54Z

clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java

+                    }
+                }
+
+                currentAssignment.put(member, deserializedAssignment);


This caching logic might work if the leader does not change.

But I have the impression that it is possible to end up in the same problem when the leader changes. Because as far as I see, the new leader will not have build up its cache. (or even worse has an outdated cache from a previous time it was leader)

Which I think can still cause troubles.

I see two other possible solutions:

The easiest one to implement I think is keep the previous code, but change the type of the currentAssignment parameter from a Map<String, List<TopicPartition>> to a Map<TopicPartition, String>, or keep the type but do a post-processing step on this list where double TopicPartitions are removed before doing the sortPartitions.

Another possibility is putting a generation counter in the schema, and only keep assignments with the maximum generation counter. (or those where it is still missing for backwards compatibility).

The first possible solutions is easy to implement but makes it a little less sticky. (which I do not mind)
The second solution is more sound, but implementing it in a backwards compatible way is a challenge.

tedyu · 2018-06-26T07:52:38Z

I would vote for #1 above since #2 would take some time to get right.

Changing type to Map<TopicPartition, String> seems to be more efficient.

vahidhashemian · 2018-06-26T21:13:31Z

@steven-aerts thanks for the feedback. The provided solution, based on some tests I ran, should resolve the bug in the use case described in the JIRA (a leader pausing for a bit, another consumer becoming a leader, and then the old leader rejoining). Can you describe more concretely a scenario that still exposes a bug in the assignor?

steven-aerts · 2018-06-27T08:19:59Z

@vahidhashemian ,

my main problem with this solution is that you keep some state on the leader. So the other consumers do not have any state which means that if any of them becomes leader, I think theoretically it is again possible to see this issue.
I think this solution reduces the change of this problem significantly, but does not take it away.
Next to that if a consumer lost its leadership, and later on is again elected as a leader, its cache will be totally out of sync which will make its next assignment nearly random instead of sticky.

To give you a concrete scenario.

schematic representation of the bug

Let's start with a schematic representation of the bug.
G1: c3 joins group and gets assignment:

c1(leader): subscription=[1,2,3] --> assignment=[1,2]
c2        : subscription=[4,5,6] --> assignment=[4,5]
c3        : subscription=null    --> assignment=[3,6]

G2: c3 is unreachable and leaves according to c1/the coordinator the group:

c1(leader): subscription=[1,2]   --> assignment=[1,2,3]
c2        : subscription=[4,5]   --> assignment=[4,5,6]
c3(gone)  : subscription=[3,6]   -->X

G3: c3 comes back with subscription from G1 combined with the G1 subscriptions of G2 for c1 and c2 this leads to a wrong assignment:

c1(leader): subscription=[1,2,3] --> assignment=[1,2,3]
c2        : subscription=[4,5,6] --> assignment=[4,5,6]
c3(back)  : subscription=[3,6]   --> assignment=[3,6]

Now a case where your solution works:

G1: c3 joins group and gets assignment:

c1(leader): subscription=[1,2,3],cache={c1:[1,2,3], c2:[4,5,6]}     --> assignment=[1,2]
c2        : subscription=[4,5,6],cache=null                         --> assignment=[4,5]
c3        : subscription=null,cache=null                            --> assignment=[3,6]

G2: c3 is unreachable and leaves according to c1/the coordinator the group:

c1(leader): subscription=[1,2],cache={c1:[1,2], c2:[4,5], c3:[3,6]} --> assignment=[1,2,3]
c2        : subscription=[4,5],cache=null                           --> assignment=[4,5,6]
c3(gone)  : subscription=[3,6],cache=null                           -->X

G3: c3 comes back with subscription from G1 but cache drops its subscription data:

c1(leader): subscription=[1,2,3],cache={c1:[1,2,3], c2:[4,5,6]}     --> assignment=[1,2]
c2        : subscription=[4,5,6],cache=null                         --> assignment=[4,5]
c3(back)  : subscription=[3,6],cache=null                           --> assignment=[3,6]

Where I think it breaks

But if in the meanwhile another consumer becomes leader you have a problem, as G3 will then look like this as the cache will not prevent the double assignment:

c1        : subscription=[1,2,3],cache={c1:[1,2,3], c2:[4,5,6]}     --> assignment=[1,2,3]
c2(leader): subscription=[4,5,6],cache=null                         --> assignment=[4,5,6]
c3(back)  : subscription=[3,6],cache=null                           --> assignment=[3,6]

So my main problem is at the end, you are still able assign a double partition, as no doubles are filtered anywhere.

vahidhashemian · 2018-06-27T21:27:25Z

@steven-aerts Thanks for describing those scenarios in detail. Though, regarding the last one (where you think it breaks) I'm not sure about how the group can end up in that state. The scenario you described would imply that at some point c1 left the group (leader change happens only when existing leader leaves the group). Therefore, at the following rebalance after c1 leaves, c2 will build its cache, and will not have a null cache when c3 comes back above. Do you agree?

I tried to rebuild the scenario with that in mind, but couldn't reproduce the issue (see below).

G1: c3 joins group and gets assignment:

c1 (leader): subscription=[1,2,3],cache={c1:[1,2,3], c2:[4,5,6]}     --> assignment=[1,2]
c2         : subscription=[4,5,6],cache=null                         --> assignment=[4,5]
c3         : subscription=null,cache=null                            --> assignment=[3,6]

G2: c3 pauses

c1 (leader): subscription=[1,2],cache={c1:[1,2], c2:[4,5], c3:[3,6]} --> assignment=[1,2]
c2         : subscription=[4,5],cache=null                           --> assignment=[4,5]

G3: c1 pauses, c2 becomes leader

c2 (leader): subscription=[4,5],cache=null                           --> assignment=[1,2,3,4,5,6]

G4: c1 and c3 resume: c2 has built the cache and would ignore invalid subscriptions by c1 and c3

c1         : subscription=[1,2],cache={c1:[1,2], c2:[4,5], c3:[3,6]} --> assignment=[1,2]
c2 (leader): subscription=[1,2,3,4,5,6],cache={c2:[1,2,3,4,5,6]}     --> assignment=[4,5]
c3         : subscription=[3,6],cache=null                           --> assignment=[3,6]

steven-aerts · 2018-06-28T04:35:54Z

@vahidhashemian
I agree it is a lot more difficult to trigger as you need in 1 generation 3 operations (c1 disappears, c1 appears, c3 disappears).
A scenario where you need only 2 operations is if in G3, c1 disappears and c3 appears in one step.

Then G3 should look like this where partition 6 still has a double assignment:

c1(gone)  : subscription=[1,2,3],cache={c1:[1,2,3], c2:[4,5,6]}     --> X
c2(leader): subscription=[4,5,6],cache=null                         --> assignment=[1,4,5,6]
c3(back)  : subscription=[3,6],cache=null                           --> assignment=[2,3,6]

For this to work you still two operations in one generation.
If we have somehow a guarantee that a leader in a consumer group will have operations atomically applied it might be your solution works (although I am even not convinced of that when you add more partitions and consumers to the mix).
But as far as I can see this is not guaranteed by kafka.

Reproducing it by hand by pausing debuggers might be tricky. What I would do to try to reproduce the scenario above is pause the coordinator (use a kafka cluser of 1), unpause c3, terminate c1 and then unpause the coordinator. And hopefully c2 will become leader while seeing c3s assignment.

Good luck,

Steven

vahidhashemian · 2018-06-28T20:49:30Z

@steven-aerts I understand. There are corner cases where the issue can resurface, even with this PR. I'll try to think about ways to improve this solution. Between the two options you suggested I like the second one best (using generations), as the first one could jeopardize stickiness (we should avoid randomly removing duplicates). I'll think about this some more. Thanks again for pointing out this issue.

steven-aerts · 2018-06-29T06:18:22Z

@vahidhashemian

I think there is a variant of option 1 which also ensures stickiness in (almost?) all cases and will be easier to implement then option two.

I think if you detect duplicates, you can loop over all subscriptions, find the subscription with the most duplicates, and remove from that subscription all duplicate partitions assignments, and test again.
Iff there are two assignments with the same number of duplicates, take the assignment with the least partitions.
With this logic I did not find any assignment where you loose stickiness while you do not need any extra meta data.

What I propose is something like this:

for( ; ; ) {
    Set<TopicPartition> doubleAssigned = currentAssignment.values().stream().flatMap(List::stream)
            .collect(Collectors.groupingBy(Function.identity(), Collectors.counting()))
            .entrySet().stream().filter(e -> e.getValue() > 1).map(Map.Entry::getKey)
            .collect(Collectors.toSet());
    if (doubleAssigned.isEmpty()) {
        break;
    }
    String toClean = currentAssignment.entrySet().stream().max(
            Comparator.comparing(e -> e.getValue().stream().filter(doubleAssigned::contains).count())
                    .thenComparing(e -> e.getValue().size(), Comparator.reverseOrder()))
            .get().getKey();

    currentAssignment.get(toClean).removeIf(doubleAssigned::contains);
}

This is just an example.
I did not test this at all. It also uses a lot of java 8 constructs which will make it difficult to cherry pick it to older releases, and the comparator will be very inefficient if there are a lot of consumers as the number of duplicated partitions is then calculated over and over again.

I hope this helps.

vahidhashemian · 2018-06-29T23:52:27Z

@steven-aerts If I understood your suggestion correctly, even though it will preserve stickiness in most situations, but there are still examples that it fails to do so. For example, let's assume a group of 5 consumers: c1, ..., c5. If c2, c3, c4, c5 drop briefly and then rejoin the group together, c1 will have the most duplicates, even though it has a valid (up-to-date) assignment. So invalidating c1's duplicate assignments will hurt stickiness. Please advise if I misunderstood your solution.

I think with generation markers we can preserve stickiness in the safest possible way: any consumer with a stale generation marker loses its assignment in case of a duplicate. I think the code doesn't have to change much to support that.

steven-aerts · 2018-06-30T06:44:17Z

@vahidhashemian
I agree that in your example my heuristic will alter the assignment of c1 which is the latest generation.
On the other hand, c2-c5 have to receive partitions from c1 to balance everything out so it might be better that they receive partitions they were previously assigned, instead of a random one. Which will happen when we work with generation markers. On the other hand this is really an extreme case.

Any solution is good for me, as long as it excludes a duplicate assignment and preserves stickiness in 95% of the cases. So if you see a way to introduce generation counter also OK for me.

Thanks

vahidhashemian · 2018-07-06T17:08:26Z

cc @hachikuji, please review! Thanks!

steven-aerts

Looks good.

The only place where I am doubting is the prevPartitions initialization.

Thanks

steven-aerts · 2018-07-06T17:25:10Z

clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java

+        // note that a conflict could exists only if user data is for different generations
+
+        Map<TopicPartition, ConsumerGenerationPair> partitionConsumer = new HashMap<>();
+        int maxGeneration = 0;


Detail: just to be sure, I would initialize it to this.generation, to make sure the generation always increments.

My thinking was that this generation marker is okay to reset to 0 in case at some point in time none of the consumers in the group carry user data, which would have the same effect even if we start with this.generation here (if no consumer has user data, this.generation = 0). I made the change as you suggested.

steven-aerts · 2018-07-06T17:37:15Z

clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java

+                    if (consumerUserData.generation == existingRecord.generation) {
+                        // same partition is assigned to two consumers during the same rebalance
+                        throw new IllegalStateException("Error: Partition '" + partition + "' is assigned to multiple consumers " +
+                                "following sticky assignment generation " + consumerUserData.generation + ".");


Small remark: By throwing here an exception you are giving up.
But I am wondering if you should not try to fix it.

I do not know which consistency guarantees a kafka consumer group exactly gives. But if it is possible to get into a split brain situation (network split), where temporarily a consumer group is split in two and gets two leaders. You get in a situation where the SitckyAssignor will never recover.
If it was me, I would write a big fat error, and drop one of the two assignments.

The same for line#322.

That makes sense. I updated this to make sure it keeps running.

steven-aerts · 2018-07-06T17:52:35Z

clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java

@@ -421,10 +515,17 @@ private int getBalanceScore(Map<String, List<TopicPartition>> assignment) {
            TreeSet<String> sortedConsumers = new TreeSet<>(new SubscriptionComparator(assignments));
            sortedConsumers.addAll(assignments.keySet());

+            List<TopicPartition> prevPartitions = new ArrayList(prevAssignment.keySet());


Maybe I am wrong, but shouldn't this be moved inside the while loop, so it is initialized every time a new assignment is selected?
Because, now I think only the first consumer will be challenged against this previous partitions.

You're right. The whole list needs to be compared against each consumer's assignment. Updated.

vahidhashemian

@steven-aerts thanks for taking a look and raising those issues. I tried to address them in the new commit.

vahidhashemian · 2018-07-06T21:31:28Z

clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java

+        // note that a conflict could exists only if user data is for different generations
+
+        Map<TopicPartition, ConsumerGenerationPair> partitionConsumer = new HashMap<>();
+        int maxGeneration = 0;


My thinking was that this generation marker is okay to reset to 0 in case at some point in time none of the consumers in the group carry user data, which would have the same effect even if we start with this.generation here (if no consumer has user data, this.generation = 0). I made the change as you suggested.

vahidhashemian · 2018-07-06T21:36:22Z

clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java

+                    if (consumerUserData.generation == existingRecord.generation) {
+                        // same partition is assigned to two consumers during the same rebalance
+                        throw new IllegalStateException("Error: Partition '" + partition + "' is assigned to multiple consumers " +
+                                "following sticky assignment generation " + consumerUserData.generation + ".");


That makes sense. I updated this to make sure it keeps running.

vahidhashemian · 2018-07-06T21:43:21Z

clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java

@@ -421,10 +515,17 @@ private int getBalanceScore(Map<String, List<TopicPartition>> assignment) {
            TreeSet<String> sortedConsumers = new TreeSet<>(new SubscriptionComparator(assignments));
            sortedConsumers.addAll(assignments.keySet());

+            List<TopicPartition> prevPartitions = new ArrayList(prevAssignment.keySet());


You're right. The whole list needs to be compared against each consumer's assignment. Updated.

steven-aerts · 2018-07-07T08:07:08Z

clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java

+            else if (consumerGeneration.generation == existingPair.generation)
+                // same partition is assigned to two consumers during the same rebalance
+                throw new IllegalStateException("Error: Partition '" + partition + "' had been assigned to " +
+                        "multiple consumers following sticky assignment generation " + consumerGeneration.generation + ".");


Just wondering, if this exception stays (and did not became an error log) because it is impossible to hit and should have been prevented by the previous error log?

You're right. I missed replacing this throw statement. It's done in the new patch. Thanks!

steven-aerts

👍 cannot wait to see this merged.

I also like the fact that this is pure java 7 so it can be backported to the bugfix releases.

clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java

kscaldef · 2018-07-25T14:13:32Z

...nts/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractPartitionAssignor.java

+                                                    Map<String, Subscription> subscriptions,
+                                                    int generation) {
+        return assign(partitionsPerTopic, subscriptions);
+    }    @Override


There's a formatting error at this line.

ean5533 · 2018-07-25T18:21:50Z

clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java

@@ -421,10 +518,18 @@ private int getBalanceScore(Map<String, List<TopicPartition>> assignment) {
            TreeSet<String> sortedConsumers = new TreeSet<>(new SubscriptionComparator(assignments));
            sortedConsumers.addAll(assignments.keySet());

+            List<TopicPartition> prevAssignedPartitions = new ArrayList(prevAssignment.keySet());


This ArrayList should be generic (that is, new ArrayList<>(...)).

kscaldef · 2018-07-26T18:26:00Z

I may be missing something, but it seems to me like onAssignment() ought to be extracting the generation from the userData and updating the field value. Otherwise, won't all the non-leader members of the consumer group keep always sending generation=0 with their current assignments?

vahidhashemian · 2018-07-27T20:18:02Z

@kscaldef You are correct. Thanks for catching this issue. I'll try to fix it in the next commit.

hachikuji

Thanks, just one final api question. Once it is decided, we should update the KIP.

hachikuji · 2019-02-25T16:53:26Z

...ams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java

@@ -791,6 +792,11 @@ public Subscription subscription(final Set<String> topics) {
        return taskIdsForConsumerAssignment;
    }

+    @Override
+    public void onAssignment(final Assignment assignment, final Optional<Integer> generation) {


We should be able to get rid of this since we have the default method.

True. Removed in the recent patch.

hachikuji · 2019-02-25T17:04:07Z

clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java

        }
    }

    @Override
-    public void onAssignment(Assignment assignment) {
+    public void onAssignment(Assignment assignment, Optional<Integer> generation) {


Trying to think through whether generation needs to be an optional. The call inside ConsumerCoordinator always passes a valid generation. The only case I can imagine is if an updated assignor gets mixed with an old client. I'm not sure that it's worth the api noise to try and handle that case. What do you think?

The impact in the case you mentioned will be minimal at best. I'd also like removing the optional. I've updated the code accordingly.

mkedwards · 2019-03-26T17:15:47Z

Is this fairly close to being ready to land? We'll be wanting to use the StickyAssignor when we deploy our latest project to production, but I'm uncomfortable leaving this window open for duplicate partition assignments. Can I help rebase/test?

vahidhashemian · 2019-03-28T14:04:38Z

This is a work in progress on my part. Delayed due to vacation. Will try to resume within a week or so.

hachikuji · 2019-04-12T06:24:38Z

@vahidhashemian Do you need any help finishing this up? Would be nice to get it over the line.

vahidhashemian · 2019-04-12T15:15:04Z

@hachikuji Sorry for the delay. I'll try to spend time on this over the weekend.

In the current implementation of sticky assignor the leader does not cache the most recent calculated assignment. It relies on the fact that each consumer in the group sends its subscribed topics and also its current assignment when a rebalance occurs. This could lead to the issue described in KAFKA-7026, in which current assignment of a consumer is no loonger valid and should be ignored. The solution implemented in this PR involves the leader caching the most recent assignment of each consumer, so the assignment reported by a consumer can be properly ignored if necessary.

… assignment by the leader This helps in situations where a new leader is chosen and there are dupliacates in the reported assignments by consumers.

(instead of the locally managed generation)

… a sorted map

…complete

…ility Also, minor fixes and clarifying comments

…ument

vahidhashemian

@hachikuji I submitted another patch to address your latest comments. Let me know what you think. Thanks!

vahidhashemian · 2019-04-14T00:56:51Z

clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java

        }
    }

    @Override
-    public void onAssignment(Assignment assignment) {
+    public void onAssignment(Assignment assignment, Optional<Integer> generation) {


The impact in the case you mentioned will be minimal at best. I'd also like removing the optional. I've updated the code accordingly.

vahidhashemian · 2019-04-14T00:57:18Z

...ams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java

@@ -791,6 +792,11 @@ public Subscription subscription(final Set<String> topics) {
        return taskIdsForConsumerAssignment;
    }

+    @Override
+    public void onAssignment(final Assignment assignment, final Optional<Integer> generation) {


True. Removed in the recent patch.

vahidhashemian · 2019-04-14T01:02:34Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java

@@ -25,6 +25,7 @@
 import org.apache.kafka.common.errors.GroupMaxSizeReachedException;
 import org.apache.kafka.common.errors.IllegalGenerationException;
 import org.apache.kafka.common.errors.InterruptException;
+import org.apache.kafka.common.errors.MemberIdRequiredException;


This was missed in the rebase.

vahidhashemian · 2019-04-14T01:04:50Z

clients/src/test/java/org/apache/kafka/clients/consumer/StickyAssignorTest.java

@@ -652,7 +652,7 @@ public void testNoExceptionThrownWhenOnlySubscribedTopicDeleted() {
        Map<String, List<TopicPartition>> assignment = assignor.assign(partitionsPerTopic, subscriptions);
        subscriptions.put(consumer,
                new Subscription(topics(topic), StickyAssignor.serializeTopicPartitionAssignment(
-                        new ConsumerUserData(assignment.get(consumer), 1))));
+                        new ConsumerUserData(assignment.get(consumer), Optional.of(1)))));


This was missed in the rebase.

hachikuji

@vahidhashemian LGTM. Thanks for the iterations. Would you mind sending an update to the mail list to mention the differences from the original KIP? We can wait a day or two to make sure there are no objections before merging.

apache#5291) This patch contains the implementation of KIP-341, which adds protection in the sticky assignor from consumers which are joining with a stale assignment. More details can be found in the proposal: https://cwiki.apache.org/confluence/display/KAFKA/KIP-341%3A+Update+Sticky+Assignor%27s+User+Data+Protocol. Reviewers: Steven Aerts <[email protected]>, Jason Gustafson <[email protected]>

steven-aerts suggested changes Jun 26, 2018

View reviewed changes

vahidhashemian changed the title ~~KAFKA-7026: Sticky Assignor Partition Assignment Improvement~~ KAFKA-7026: Sticky Assignor Partition Assignment Improvement (WIP) Jun 28, 2018

vahidhashemian force-pushed the KAFKA-7026 branch from a949367 to 8714d0e Compare July 3, 2018 18:05

vahidhashemian changed the title ~~KAFKA-7026: Sticky Assignor Partition Assignment Improvement (WIP)~~ KAFKA-7026: Sticky Assignor Partition Assignment Improvement Jul 6, 2018

steven-aerts suggested changes Jul 6, 2018

View reviewed changes

vahidhashemian force-pushed the KAFKA-7026 branch from 8714d0e to 86fcf67 Compare July 6, 2018 21:48

vahidhashemian commented Jul 6, 2018

View reviewed changes

steven-aerts approved these changes Jul 7, 2018

View reviewed changes

vahidhashemian force-pushed the KAFKA-7026 branch from 86fcf67 to 23b0ff4 Compare July 9, 2018 21:03

steven-aerts approved these changes Jul 10, 2018

View reviewed changes

hachikuji reviewed Jul 11, 2018

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java Show resolved Hide resolved

vahidhashemian force-pushed the KAFKA-7026 branch 2 times, most recently from 7740773 to 14e4d82 Compare July 19, 2018 18:05

kscaldef reviewed Jul 25, 2018

View reviewed changes

ean5533 reviewed Jul 25, 2018

View reviewed changes

vahidhashemian force-pushed the KAFKA-7026 branch from 14e4d82 to 894e73f Compare July 25, 2018 22:52

vahidhashemian force-pushed the KAFKA-7026 branch 2 times, most recently from 2ed9a9c to 4cd3bd4 Compare August 1, 2018 22:37

hachikuji reviewed Feb 25, 2019

View reviewed changes

vahidhashemian added 17 commits April 13, 2019 09:51

Changed the solution to use generation markers instead of caching the…

94edbcc

… assignment by the leader This helps in situations where a new leader is chosen and there are dupliacates in the reported assignments by consumers.

Minor fixes

6ab9400

Minor update

c0926db

Fixed backward schema compatibility

bde32ed

Added a unit test to cover conflicting previous assignments

20c8253

Switch to using consumer group generation in user data

7fd5333

(instead of the locally managed generation)

Minor updates

4467d37

Fixed a couple of user data related issues

32eff14

Minor updates based on feedbacky

10564ab

Removed the generation bump in StickyAssignor

aca44c7

Simplify finding current and last consumer of each partition by using…

6ad0993

… a sorted map

Made improvements based on the feedback

e06a02b

Pass on the generation from the coordinator only after assignment is …

980d858

…complete

Use Optional<Integer> for generation to improve backward compatib…

f2be37d

…ility Also, minor fixes and clarifying comments

Added a default implementation to the interface

2e410c0

Removed the optional qualifier from onAssignment's 'generation` arg…

a03a734

…ument

vahidhashemian force-pushed the KAFKA-7026 branch from 623e7cb to d13a973 Compare April 14, 2019 01:00

vahidhashemian commented Apr 14, 2019

View reviewed changes

vahidhashemian force-pushed the KAFKA-7026 branch 2 times, most recently from ca03449 to a03a734 Compare April 14, 2019 01:16

vahidhashemian commented Apr 14, 2019

View reviewed changes

hachikuji approved these changes Apr 16, 2019

View reviewed changes

hachikuji merged commit 3e8a10e into apache:trunk Apr 18, 2019

KAFKA-7026: Sticky Assignor Partition Assignment Improvement (KIP-341) #5291

KAFKA-7026: Sticky Assignor Partition Assignment Improvement (KIP-341) #5291

Conversation

vahidhashemian commented Jun 26, 2018 • edited Loading

Committer Checklist (excluded from commit message)

steven-aerts left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tedyu commented Jun 26, 2018

vahidhashemian commented Jun 26, 2018

steven-aerts commented Jun 27, 2018

To give you a concrete scenario.

schematic representation of the bug

Now a case where your solution works:

Where I think it breaks

vahidhashemian commented Jun 27, 2018 • edited Loading

steven-aerts commented Jun 28, 2018

vahidhashemian commented Jun 28, 2018

steven-aerts commented Jun 29, 2018

vahidhashemian commented Jun 29, 2018

steven-aerts commented Jun 30, 2018

vahidhashemian commented Jul 6, 2018

steven-aerts left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vahidhashemian left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

steven-aerts left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kscaldef commented Jul 26, 2018

vahidhashemian commented Jul 27, 2018

hachikuji left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mkedwards commented Mar 26, 2019

vahidhashemian commented Mar 28, 2019 • edited Loading

hachikuji commented Apr 12, 2019

vahidhashemian commented Apr 12, 2019

vahidhashemian left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji left a comment

Choose a reason for hiding this comment

vahidhashemian commented Jun 26, 2018 •

edited

Loading

vahidhashemian commented Jun 27, 2018 •

edited

Loading

vahidhashemian commented Mar 28, 2019 •

edited

Loading