Add support for using CooperativeStickyAssignor with KafkaConsumer #844
Conversation
KafkaConsumer.partitionsMapStream would drop all partition streams whenever a partition revocation was advertised, but would only create streams for newly assigned partitions. This worked fine with the default assignor, which revokes all existing assignments before new assignments are issued during a rebalance operation. The CooperativeStickyAssignor, attempts to keep assignments stable, performing rebalance operations in two steps: on a first pass a minimal set of assignments are revoked, these are then redistributed to balance the consumer cluster. In order to support this assignor, the consumer needs to keep track of the current set of assignments and, in the case of partitionsMapStream, recreate partition streams for partitions that were kept. While the underlying org.apache.kafka.clients.consumer.KafkaConsumer does this for us, this follows existing practice in KafkaConsumer.assignmentStream to keep track of the currently assigned set. Going forward, it would make sense to implement partitionedStream independently of partitionsMapStream, so as to avoid recreating streams for sticky partitions, during a rebalance operation. New tests were added to validate that assignmentStream and partitionsMapStream work with the CooperativeStickyAssignor. The former was working out of the box, as the stream kept track of the assigned set on its own. For the latter use case, a timeout was added, as revoking assigned partitions would make the test hang indefinitely.
|
Hi @biochimia! Thank you for this contribution, it's definitely a useful addition. Implementation and tests look good to me. But could you update scaladoc for methods in the |
|
I'll try to look at this at the weekend but also happy for this be to merged based on an approval from @LMnet |
|
Hello guys, would it be possible to have this back-ported to 1.x series? Thank you so very much 🙇 |
Thanks for giving this a look. I updated the documentation a bit, almost minimally, but feedback is welcome. What are your thoughts on implementing |
| @@ -540,6 +541,91 @@ final class KafkaConsumerSpec extends BaseKafkaSpec { | |||
| } | |||
| } | |||
|
|
|||
| it("should handle rebalance with CooperativeStickyAssignor") { | |||
There was a problem hiding this comment.
This is a copy of the "should handle rebalance" test, adapted to use the custom assignor, along with the resulting assert.
| val revokedFetches = revoked intersect fetches | ||
| val revokedNonFetches = revoked diff fetches | ||
| val revokedNonFetches = revoked diff revokedFetches |
There was a problem hiding this comment.
I would appreciate a comment on this change in particular.
The change makes sense to me, logically: revokedFetches and revokedNonFetches add up to revoked. But I did not dig enough into the implications to be sure this is the intention. For instance: in which combination of pending records and pending fetches it will make a difference, and what is the consequence?
I thought I had left this out of the PR, but while it's here we might as well discuss it.
| @@ -707,6 +793,65 @@ final class KafkaConsumerSpec extends BaseKafkaSpec { | |||
| } | |||
| } | |||
|
|
|||
| it("should stream assignment updates to listeners when using CooperativeStickyAssignor") { | |||
There was a problem hiding this comment.
This is a copy of the existing "should stream assignment updates to listeners" test, adapted to use the custom assignor, along with the resulting asserts.
| consumer2Updates.length == 2 | ||
| consumer2Updates.length == 2 && |
There was a problem hiding this comment.
I think this affected scalatest's ability to produce pretty failure messages.
| @@ -516,6 +516,7 @@ final class KafkaConsumerSpec extends BaseKafkaSpec { | |||
| } | |||
| } | |||
| .takeWhile(_.size < 200) | |||
| .timeout(20.seconds) | |||
There was a problem hiding this comment.
Timeout added to match test with custom assignor. This test was working without it, while the one for the custom assignor would hang indefinitely.
|
@biochimia could you please clarify for me one moment: what will a user receive after rebalancing with the
If the answer is 1, then it means that the current returning type of case class FullAssigmnentInfo[F[_], K, V](
newlyAssignedPartitions: Map[TopicPartition, Stream[F, CommittableConsumerRecord[F, K, V]],
revokedPartitions: SortedSet[TopicPartition],
retainedPartitions: SortedSet[TopicPartition],
)If the answer is 2, we have 2 variants to deal with the streams for the retained partitions:
I believe that to retain backward compatibility we should pass to the def streamDetailed: Stream[F, FullAssigmnentInfo[F, K, V]] |
As it stands, they'll receive 2. new streams for all currently assigned partitions.
As you mention, one motivation behind While this is not optimal for the Finally, these changes have minimal impact on the default use case, while making the Without breaking user expectations for the interface, I don't think that we can make My proposal here is to add an implementation of |
The current policy seems to be that we accept pull requests against 1.x but don't actively seek to maintain feature parity with 2.x. Would this be easy to retarget for 1.x @biochimia? In the longer term I think we should decide to either maintain feature parity in 1.x or stop adding any new features - any thoughts on this @LMnet @vlovgr? What do the Maven download figures for 1.x vs 2.x look like? |
Looking at the files I touched (implementation and test) and the diff between the |
|
@biochimia after your last comment I decided to test your branch locally. And it looks like even with the I think we should implement support for This change means that an old convention for the This decision also means that we will not have a method with the full assignment info after each rebalance. But I already proposed a solution for this: we need one more method with the
I don't think that this is a good idea. The current behavior of all stream methods is consistent: they all work identically. But they have a different amount of details. And that's all. I think this is an excellent thing in terms of API predictability: it doesn't matter which stream method you are using. They all have the same semantic. If different stream methods will behave differently it could be surprised for users. It may become one of these tricky minor things you have to remember to use a library. @bplommer @vlovgr I need your opinions on this design decision because we have multiple options here. About 1.x backport: I have nothing against it if someone will volunteer this backport. But overall the current strategy about 1.x is that maintainers don't try to maintain feature parity. |
|
@LMnet I agree with everything you said. 👍 |
Thank you for taking the time to give this a test run.
Yes, this was the intended behaviour for my changes.
My concerns with doing this are as follows (as noted, these concerns only affect the users of
As you note, this is a design choice for the library. I bring forward my concerns so they can be discussed (and even invalidated). If there's agreement that the best path forward is to not recreate streams at the cost of having an incomplete assignment map in About impact on current use cases using the default assignor, my only concern was about an implementation detail: currently we use a single |
|
@biochimia, all your concerns are correct. It is a trade-off, of course.
I think you can go on and update your changes. I'm pretty sure @bplommer will not be against it.
I think you could create a single |
|
I've updated the PR to reflect what we've discussed. Namely, |
The purpose of this change is to allow partition streams returned by KafkaConsumer.partitionsMapStream to be individually completed, without having to close the full assignment at once as happened before. When using the default assignor, there should be no visible change in behaviour, as partition assignments are revoked in full on a rebalance, and new assignments sent out to consumers. When using the CooperativeStickyAssignor, however, partition assignments may be individually revoked without affecting other assignments. In this case, partition streams for non-revoked partitions will remain active until the point they get revoked. Compared to behaviour observed with the default assignor, when using partitionsMapStream with CooperativeStickyAssignor the following will be observed: - maps in partitionsMapStream will include only entries for newly assigned partitions; When partition assignments are revoked, an empty map will be emitted. - partition streams for non-revoked assignments will remain active after a new (partial) assignment map is emitted.
biochimia
force-pushed
the
cooperative-assignor
branch
from
7fba4c9 to
c04aba2
Compare
|
I re-pushed my changes (updated hash, only), to trigger the CI workflow. There was an issue affecting GitHub actions before that meant some triggers did not fire. |
|
This PR's CI needs trigger from a maintainer 🙏 |
|
@biochimia I tried to run CI multiple times and it fails on the |
I can give this a look, but it seems that this test is flaky and fails on a race condition:
Edit: the unrelated PR: #859 |
|
@biochimia ok, got it. @bplommer can you take a look at this pr? Or maybe we could merge it and fix all stuff by ourselves. |
|
I had a look at the failing test, and was able to reproduce the failure on top of vanilla
When the test fails, you see the first consumer being assigned all partitions and fetching the records before the second consumer is able to join. When the second consumer joins, and assignment changes, the underlying consumer resets offsets back to 0, and records are consumed again, thus failing the test with repeated records. The race condition then seems to be about the first consumer being able to fetch and process all records before a rebalance is triggered by the second consumer joining. |
|
Awesome news that this got merged! 🎉 @biochimia how much work do you think it would take to backport this to |
KafkaConsumer.partitionsMapStream would drop all partition streams
whenever a partition revocation was advertised, but would only create
streams for newly assigned partitions. This worked fine with the default
assignor, which revokes all existing assignments before new assignments
are issued during a rebalance operation.
The CooperativeStickyAssignor, attempts to keep assignments stable,
performing rebalance operations in two steps: on a first pass a minimal
set of assignments are revoked, these are then redistributed to balance
the consumer cluster.
In order to support this assignor, the consumer needs to keep track of
the current set of assignments and, in the case of partitionsMapStream,
recreate partition streams for partitions that were kept. While the
underlying org.apache.kafka.clients.consumer.KafkaConsumer does this for
us, this follows existing practice in KafkaConsumer.assignmentStream to
keep track of the currently assigned set.
Going forward, it would make sense to implement partitionedStream
independently of partitionsMapStream, so as to avoid recreating streams
for sticky partitions, during a rebalance operation.
New tests were added to validate that assignmentStream and
partitionsMapStream work with the CooperativeStickyAssignor. The former
was working out of the box, as the stream kept track of the assigned set
on its own. For the latter use case, a timeout was added, as revoking
assigned partitions would make the test hang indefinitely.