KAFKA-15106 fix AbstractStickyAssignor isBalanced predict

[flashmouse]

https://issues.apache.org/jira/browse/KAFKA-15106

in 3.5.0 AbstractStickyAssignor may run useless loop in performReassignments because isBalanced have a trivial mistake, and result in rebalance timeout in some situation.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status

[flashmouse]

@rajinisivaram @dajac plz have a look, thx!

[divijvaidya]

Thank you for your first change to Apache Kafka @flashmouse! In the JIRA, you mention that this could be reproduced using a unit test. Can you please add the unit test here which would fail prior to this change and succeed after it?

[flashmouse]

@divijvaidya thank you for reply!
the unit test org.apache.kafka.clients.consumer.internals.AbstractStickyAssignorTest#testLargeAssignmentAndGroupWithNonEqualSubscription could use much more time than configured Timeout(90) and this could reproduce easy when partitionCount and consumerCount are large enough and not equal(ex. partitionCount= 200, consumerCount= 20), in this situation isBalanced and performReassignments may run whole loop body.

because both performReassignments and isBalanced in AbstractStickyAssignor are not so efficient(performReassignments is worse), isBalanced could abate performReassignments call times, so fix its logic could speed up this unit test when parameters are small enough.

in my test, after this fix, testLargeAssignmentAndGroupWithNonEqualSubscription could pass when partitionCount= 200, consumerCount = 20, but is still slower than the combination partitionCount= 2000, consumerCount = 2000 because in this test when the 2 parameters are equal, isBalanced satisfy the prediction min >= max - 1 thus could return true immediately.

may be I need add a unit test code with a new commit?

[kirktrue]

@ableegoldman you're pretty familiar with this code, IIUC. If so, could you take a look at this PR? Thanks!

[kirktrue]

@flashmouse I'm a little slow on the uptake, so I'm trying to come up with a scenario. Let's say the following is true:

• Topic foo has three partitions
• The consumer group has two consumers, consumer and otherConsumer

In the first case, let's say that consumer has one partition and otherConsumer has two. In that case it's as balanced as possible, right? But the code in isBalanced as written disagrees:

if (consumerPartitionCount < otherConsumerPartitionCount) {
    return false;
}

Because 1 is less than 2 😄 But it's going to run into problems because by returning false from isBalanced, I assume we then trigger a rebalance, which—at best— will cause the ownership of that partition to move from otherConsumer to consumer. When the next balance check for is run, we're in the same position 🤷‍♂️

But with your fix:

if (consumerPartitionCount + 1 < otherConsumerPartitionCount) {
    return false;
}

It's effectively stating that the consumer has to have at least two partitions difference, correct? Because 1 + 1 is not less then 2.

Do I have the above (mostly) correct?

Thanks!

[flashmouse]

@kirktrue thank you for reply!

In your case, although isBalanced return false, performReassignments still won't reassign any partition, because it would only do reassign when currentAssignment.get(consumer).size() > currentAssignment.get(otherConsumer).size() + 1, it will just loop all partitions in reassignable list and find no one match it so modified keep false then return done.

I think the problem is isBalanced and performReassignments should have the same predict logic when they judge whether should do reassign. for now performReassignments itself think should do reassign only when 2 consumers that their partition num gap should have at least 2 but isBalanced is 1.

I mean with such situation, I think if we not modify isBalanced, the reassignment logic is still right, but may very slow. performReassignments will do quite a lot unnecessary work, as I said above, it could done finally but cost a long time. you can try unit test testLargeAssignmentAndGroupWithNonEqualSubscription without this pr change, if set partitionCount= 20, consumerCount = 2, it could done with success, but if set partitionCount= 200, consumerCount = 20, it may run very long time. I have run the test and no result after 14min, and it could speed up to 47sec after I apply my change.

[rreddy-22]

Thanks for the catch! I also observed that the actual reassignments don't take place inside perform reassignments so ideally it should eventually end but only after a long period of time. I ran it for 6hrs and it finally failed with the timeout error. The change would make it faster to determine whether the assignment is balanced or not so I think we should change this irrespective. I think originally the definition of a balanced assignment was that the numbers of topic partitions assigned to consumers differ by at most one but this isn't true for assignments with different subscriptions so I agree we should change it.

[flashmouse]

@rreddy-22 that's what I mean, thank you, could any one merge this if possible or help review to verify this pr is useful?

we are now using CooperativeSticky strategy in kafka-client3.5 but meet serious problem that it may trigger rebalance with colossal times(complete one rebalance about 3sec, then keep trigger new balance within more than 10minutes, that's say, at least rebalance 200 times). for now we cannot give much more useful information and we want to verify whether that is caused by this problem after this pr is checked.

[dajac]

@flashmouse Thanks for the PR. It seems that this condition has been around for a while now. Did you observe a similar issue with version prior to 3.5? I wonder what if changes introduced in 3.5 for the rack aware assignment made this worse than it used to be. I agree that we need to get to the bottom of this.

[flashmouse]

@dajac
we haven't seen such many rebalance times before, we before use CooperativeSticky with kafka-client 3.0.0, rebalance times increase very slight.

but we are not use kafka-client 3.5 directly because CooperativeStickyAssignor in 3.5 need upgrad server to 3.5 too if need rack-aware assign logic and we couldn't upgrade server right now so we move AbstractStickyAssignor and CooperativeStickyAssignor logic to our application and move the rackInfo to userData. although we have passd all unit test in kafka-client after do this change, we still doubt whether this change caused rebalance time increase so we're keeping check.

[rreddy-22]

@flashmouse Thanks for the PR. It seems that this condition has been around for a while now. Did you observe a similar issue with version prior to 3.5? I wonder what if changes introduced in 3.5 for the rack aware assignment made this worse than it used to be. I agree that we need to get to the bottom of this.

The rack awareness changes introduce a whole iteration through all the potential consumers to check if any consumer has a matching rack and in the worst case scenario where no consumer with a matching rack is found, the whole iteration happens all over again. Previously we would only iterate through the list once, so potentially the time could've doubled now.
https://github.com/apache/kafka/blob/3.5/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java#L1389-L1400

[flashmouse]

I found another one bug.

In org.apache.kafka.clients.consumer.internals.AbstractStickyAssignor#allSubscriptionsEqual, it would check whether one partition is assigned by multiple consumers in different generation, the logic has 3 if-else body, the last one condition memberGeneration < otherMemberGeneration should add this line below:

allPreviousPartitionsToOwner.put(tp, otherConsumer);

without this line may happen error in such scenario below:
say we have 3 consumers, they own partition1 respectively in generation 1, 2, 3, and then the check sequence is gen3, gen1, gen2, then gen3 and gen2 are all added to consumerToOwnedPartitions so this partition added twice.

this may happen if one partition is owned by at least 3 generation consumer.

@rreddy-22 @kirktrue could u take a look? thx!

[rreddy-22]

So in line 161 we directly added the current consumer to the map allPreviousPartitionsToOwner. And if we choose to keep that line we should remove the put in the first two conditions in lines 175 and 180 and we can keep the change you put.
However, I feel like this would be redundant since we're replacing the otherMember and adding it back if the conditions aren't favorable. My suggestion would be to replace line 161 with get instead of put and then keep the rest of the code as it is.

[flashmouse]

updated, plz check again

[philipnee]

Hey @flashmouse - could you explain what do you mean by useless loop?

[flashmouse]

Hey @flashmouse - could you explain what do you mean by useless loop?

run while body in performReassignments but couldn't modify assignment

[philipnee]

@flashmouse thansk for the PR - I was the one who wrote the if-else if -else block. I'm a bit surprised that I actually made this mistake, so I held it off for a few days just to make sure. As you have already provided a case that breaks that existing logic, could you implement this as part of the test? Also, could you provide a test case that fails if (consumerPartitionCount < otherConsumerPartitionCount) { but succeeds in if (consumerPartitionCount + 1 < otherConsumerPartitionCount) {

This would help massively for the reviews and documentation.

[philipnee]

Does moving this line after the log.warn do anything?

[rreddy-22]

Agreed we should move it back to the original position

[flashmouse]

done

[rreddy-22]

We'd Have to add a put here too

[flashmouse]

fixed, plz take a look

[flashmouse]

@flashmouse thansk for the PR - I was the one who wrote the if-else if -else block. I'm a bit surprised that I actually made this mistake, so I held it off for a few days just to make sure. As you have already provided a case that breaks that existing logic, could you implement this as part of the test? Also, could you provide a test case that fails if (consumerPartitionCount < otherConsumerPartitionCount) { but succeeds in if (consumerPartitionCount + 1 < otherConsumerPartitionCount) {

This would help massively for the reviews and documentation.

hi @philipnee thx for reply!

In org.apache.kafka.clients.consumer.internals.AbstractStickyAssignorTest,I add:

1. testAssignmentAndGroupWithNonEqualSubscriptionNotTimeout, this test may run very long time without fix.
2. testSubscriptionNotEqualAndAssignSamePartitionWith3Generation, this test should fail without fix.

and please help review #13965 if possible, it mentions another serious bugs in org.apache.kafka.clients.consumer.internals.AbstractStickyAssignor.ConstrainedAssignmentBuilder#ConstrainedAssignmentBuilder and may cause rebalance stuck forever.

[rreddy-22]

Could you please format it in multiple lines like this, it increases readability

[rreddy-22]

    subscriptions.put(
        consumer1,
        buildSubscriptionV2Above(topics(topic1, topic2),
        partitions(tp(topic1, 0)), generationId, 0)
    );

[flashmouse]

done, but seems this is enough?

subscriptions.put(consumer1, buildSubscriptionV2Above(topics(topic),
                    partitions(tp(topic, 0), tp(topic, 2)), ints[0], 0));

[philipnee]

Hmm - you can try. But I've had similar, if not the same, failures and I haven't had a chance to examine the cause.

[ableegoldman]

@flashmouse you can decrease the consumer count. For one thing, this covers the edge case where the consumers do not all have the same topic subscription, which seems to be quite rare. For another, 2,000 consumers is fairly large to begin with. And especially large for a consumer group where the consumers are all given different topic subscriptions.

Honestly it's probably more important to test with a higher partition count than a large group of consumers like this. Are you able to get it passing within 90s with 500 consumers and 5,000 partitions? That feels like a more realistic large-scale test case. If not, fix it with just 500 consumers and maybe run the test with 4,000, then 3,000, then 2,000 partitions, etc, until you can get it to pass.

If it's still not passing with 500 consumers and 2000 partitions, that's a bit troubling. But we can cross that bridge if it comes up

[flashmouse]

thx @ableegoldman
I reduce partitions num and increase consumers num, I think all tests influenced by this pr should be PASSED now.

[guozhangwang]

@flashmouse Thanks! I echo other's thoughts that in a unit test, we could refrain ourselves from trying very large scale when trying to validate a specific logic or expose a bug. Just giving a data point here, in StreamsAssignmentScaleTest which is an integration test, given the goal is actually to validate that the assignor is efficient enough such that it can complete the assignment task within a certain time, our scale is around 1000 - 6000 partitions with 2 clients only.

I'm actually wondering if this bug could be exposed with a even small scale, like 3 consumers and 4 partitions at all?

[flashmouse]

@flashmouse Thanks! I echo other's thoughts that in a unit test, we could refrain ourselves from trying very large scale when trying to validate a specific logic or expose a bug. Just giving a data point here, in StreamsAssignmentScaleTest which is an integration test, given the goal is actually to validate that the assignor is efficient enough such that it can complete the assignment task within a certain time, our scale is around 1000 - 6000 partitions with 2 clients only.

I'm actually wondering if this bug could be exposed with a even small scale, like 3 consumers and 4 partitions at all?

@guozhangwang thx for reply!

this bug couldn't reproduce in a very small scale, it's actually not a bug, but a trivial mistake would make rebalance logic run very slow. if test scale are too small, it couldn't prove this.

I think the root cause is performReassignments in GeneralAssignmentBuilder is a O(partSize * consumerNum) functions, it is very inefficient.

[guozhangwang]

I think the root cause is performReassignments in GeneralAssignmentBuilder is a O(partSize * consumerNum) functions, it is very inefficient.

Ack, and with your changes, it's now constant and not linear to either num.partitions or num.consumers right?

[flashmouse]

I think the root cause is performReassignments in GeneralAssignmentBuilder is a O(partSize * consumerNum) functions, it is very inefficient.

Ack, and with your changes, it's now constant and not linear to either num.partitions or num.consumers right?

I hadn't changed the logic, it is still a slow function. whether execute performReassignments is checked by isBalanced, but it is not effective previously and this pr just fix this.

[guozhangwang]

I hadn't changed the logic, it is still a slow function. whether execute performReassignments is checked by isBalanced, but it is not effective previously and this pr just fix this.

I'm a bit confused now.. my understanding is that with this line of change https://github.com/apache/kafka/pull/13920/files#diff-2814643946ccefef81f480e1280e9b3e3397f4673166ca1ac47db7e3cb6fb760L1175, when num.partitions and num.consumers are not the same, we would no longer run isBalanced | performReassignments once per each combo, and hence it's no longer linear to num.partitions * num.consumers. Of course it would not return immediately (that's only the case when num.partitions and num.consumers happen to be equal), but it would still be much better than the old logic, right?

[flashmouse]

I hadn't changed the logic, it is still a slow function. whether execute performReassignments is checked by isBalanced, but it is not effective previously and this pr just fix this.

I'm a bit confused now.. my understanding is that with this line of change https://github.com/apache/kafka/pull/13920/files#diff-2814643946ccefef81f480e1280e9b3e3397f4673166ca1ac47db7e3cb6fb760L1175, when num.partitions and num.consumers are not the same, we would no longer run isBalanced | performReassignments once per each combo, and hence it's no longer linear to num.partitions * num.consumers. Of course it would not return immediately (that's only the case when num.partitions and num.consumers happen to be equal), but it would still be much better than the old logic, right?

ah this is right.

[guozhangwang]

LGTM. Merged to trunk. Thanks @flashmouse !

[guozhangwang]

Please also update the JIRA ticket status with fix version fields.

[flashmouse]

thx, already update jira status, and plz help review #13965 if could, thx.
@guozhangwang @philipnee

[showuon]

Backported to 3.5 branch.