You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@flink.apache.org by Hongshun Wang <lo...@gmail.com> on 2024/12/27 06:37:21 UTC
[Discuss] Kafka Enumerator assigns splits based on the current assignment situation.
Hi devs,
I'd like to push a new FLIP xxx: Kafka Enumerator assigns splits based on
the current assignment situation. Currently, when the Kafka enumerator
finds a new split, it will choose a reader using the following algorithm:
```java
static int getSplitOwner(TopicPartition tp, int numReaders) {
int startIndex = ((tp.topic().hashCode() * 31) & 0x7FFFFFFF) %
numReaders;
// here, the assumption is that the id of Kafka partitions are
always ascending
// starting from 0, and therefore can be used directly as the
offset clockwise from the
// start index
return (startIndex + tp.partition()) % numReaders;
}
```
However, I have encountered several unevenly distributed problems in
production:
- problem 1: If I consume multiple topics from a source, the
getSplitOwner will count the owner of partitions separately for each
topic. This means that there may be overlapping assignments. For example,
if there are 3 Kafka topics with 16 partitions each, even if the Flink
task's parallelism is 48, only 16 tasks may be assigned 3 partitions in the
worst-case scenario.
- problem 2: If I use a Flink source with 16 parallelism to assumes one
topic with 16 partitions. Then, when I change the Flink parallelism to 32,
the start index changes when a new partition is discovered because the
number of readers has changed.
Each time, I only have one method to solve it: wait for the job to complete
a snapshot state, then stop and restart the job with a different
parallelism. This way, the state redistribution strategy will assign the
splits as evenly as possible. However, this is too cumbersome—why can't it
be done properly when the enumerator discovers and assigns a new split?
Therefore, I want the enumerator to select a reader based on the current
assignment situation. Each time, the enumerator should choose the reader
whose assigned split number is the smallest.
WDYT? I'm willing to hear from you.
Best,
Hongshun
Re: [Discuss] Kafka Enumerator assigns splits based on the current assignment situation.
Posted by Hongshun Wang <lo...@gmail.com>.
Hi Ahmed,
Thanks for the additional information about the current assignment strategy
problem that I had overlooked. Initially, I thought that Kafka alone was
not sufficient for the flip, but now that you've broadened the scope, I am
excited to work on it for the common Flink source interfaces.
I will consider more details and then start a FLIP document. Let's discuss
it further later.
Best
hongshun
On Mon, Dec 30, 2024 at 10:16 PM Ahmed Hamdy <ha...@gmail.com> wrote:
> Hi Hongshun,
> I personally see this as a common challenge for source connectors, it might
> be a good FLIP candidate for the common flink-source interfaces.
>
> I still see a couple of challenges that might be covered in the FLIP, for
> example currently the Kafka source doesn't wait for reader registration to
> assign splits, yet it adds them to the reader in pending mode. to guarantee
> fair distribution this would probably mean we await all reader's
> registration to get the current assignment status before assigning any
> splits, This might be okay though but needs some PoC and more details (for
> example do we only start periodic discovery after reader's registration or
> do we hold unassigned topics in state in non pending mode,...etc).
>
> Regarding the distribution strategy interface, Does this mean you want to
> expose the algorithm choice to users or do you intend to make it
> configurable by source implementers only (if we are reusing for other
> sources besides kafka)? I wouldn't vote to expose it to users but I would
> love to reuse it for different connectors if that's the case.
>
> Finally, given that restoring from state already redistributes splits
> evenly, did you consider a simpler approach, like just a round-robin
> assignment for all topic/partitions instead of hash based topic assignment
> + round-robin partition?
>
> I am sorry if the discussion is getting more detailed, happy to move it to
> a FLIP if the community agrees as well.
>
> Best Regards
> Ahmed Hamdy
>
>
> On Mon, 30 Dec 2024 at 03:12, Hongshun Wang <lo...@gmail.com>
> wrote:
>
> > Hi Ahmed,
> > I want to do two things in this flip.
> > 1. Add current assignments to ReaderRegistrationEvent. Thus, enumerator
> can
> > know the overall assignments after restarting the jon.
> > 2. Add an distribution strategy interface to the kafka enumerator, thus
> can
> > flexibly choose how to assign the split based on the current
> circumstances.
> >
> >
> > Best,
> > Hongshun
> >
> > On Mon, Dec 30, 2024 at 10:03 AM Hongshun Wang <lo...@gmail.com>
> > wrote:
> >
> > > Hi Ahmed,
> > >
> > > Thanks for your question.
> > > > Do you intend to keep the current assignment in the enumerator
> > > state or are you only going to use per session assignment status and
> keep
> > > it in memory?
> > >
> > > Not keep in state, because it's be different with the real distribution
> > > when redistribute state if the parrelism changed. When restart the
> job,
> > > each task send an event to tell enumerator its assignments. It's also
> > why I
> > > want to add a FLIP rather than jira.
> > >
> > > Best,
> > > Hongshun
> > >
> > > On Mon, Dec 30, 2024 at 12:42 AM Ahmed Hamdy <ha...@gmail.com>
> > wrote:
> > >
> > >> Hi Hongshun
> > >>
> > >> There is currently a Jira tracking this issue [1].
> > >>
> > >> >Therefore, I want the enumerator to select a reader based on the
> > current
> > >> assignment situation. Each time, the enumerator should choose the
> reader
> > >> whose assigned split number is the smallest.
> > >>
> > >> I am not sure I follow how you intend to track the "current split
> > >> assignment". Do you intend to keep the current assignment in the
> > >> enumerator
> > >> state or are you only going to use per session assignment status and
> > keep
> > >> it in memory? If you intend to do the former, then how are you going
> to
> > >> handle split redistribution upon parallelism change?
> > >>
> > >>
> > >> 1- https://issues.apache.org/jira/browse/FLINK-31762
> > >> Best Regards
> > >> Ahmed Hamdy
> > >>
> > >>
> > >> On Fri, 27 Dec 2024 at 06:38, Hongshun Wang <lo...@gmail.com>
> > >> wrote:
> > >>
> > >> > Hi devs,
> > >> >
> > >> > I'd like to push a new FLIP xxx: Kafka Enumerator assigns splits
> based
> > >> on
> > >> > the current assignment situation. Currently, when the Kafka
> enumerator
> > >> > finds a new split, it will choose a reader using the following
> > >> algorithm:
> > >> >
> > >> >
> > >> > ```java
> > >> >
> > >> > static int getSplitOwner(TopicPartition tp, int numReaders) {
> > >> > int startIndex = ((tp.topic().hashCode() * 31) &
> 0x7FFFFFFF) %
> > >> > numReaders;
> > >> >
> > >> > // here, the assumption is that the id of Kafka partitions
> are
> > >> > always ascending
> > >> > // starting from 0, and therefore can be used directly as
> the
> > >> > offset clockwise from the
> > >> > // start index
> > >> > return (startIndex + tp.partition()) % numReaders;
> > >> > }
> > >> >
> > >> > ```
> > >> >
> > >> >
> > >> > However, I have encountered several unevenly distributed problems in
> > >> > production:
> > >> >
> > >> >
> > >> > - problem 1: If I consume multiple topics from a source, the
> > >> > getSplitOwner will count the owner of partitions separately for
> > each
> > >> > topic. This means that there may be overlapping assignments. For
> > >> > example,
> > >> > if there are 3 Kafka topics with 16 partitions each, even if the
> > >> Flink
> > >> > task's parallelism is 48, only 16 tasks may be assigned 3
> > partitions
> > >> in
> > >> > the
> > >> > worst-case scenario.
> > >> > - problem 2: If I use a Flink source with 16 parallelism to
> assumes
> > >> one
> > >> > topic with 16 partitions. Then, when I change the Flink
> parallelism
> > >> to
> > >> > 32,
> > >> > the start index changes when a new partition is discovered
> because
> > >> the
> > >> > number of readers has changed.
> > >> >
> > >> >
> > >> > Each time, I only have one method to solve it: wait for the job to
> > >> complete
> > >> > a snapshot state, then stop and restart the job with a different
> > >> > parallelism. This way, the state redistribution strategy will assign
> > the
> > >> > splits as evenly as possible. However, this is too cumbersome—why
> > can't
> > >> it
> > >> > be done properly when the enumerator discovers and assigns a new
> > split?
> > >> >
> > >> >
> > >> > Therefore, I want the enumerator to select a reader based on the
> > current
> > >> > assignment situation. Each time, the enumerator should choose the
> > reader
> > >> > whose assigned split number is the smallest.
> > >> >
> > >> >
> > >> > WDYT? I'm willing to hear from you.
> > >> >
> > >> >
> > >> > Best,
> > >> >
> > >> > Hongshun
> > >> >
> > >>
> > >
> >
>
Re: [Discuss] Kafka Enumerator assigns splits based on the current assignment situation.
Posted by Ahmed Hamdy <ha...@gmail.com>.
Hi Hongshun,
I personally see this as a common challenge for source connectors, it might
be a good FLIP candidate for the common flink-source interfaces.
I still see a couple of challenges that might be covered in the FLIP, for
example currently the Kafka source doesn't wait for reader registration to
assign splits, yet it adds them to the reader in pending mode. to guarantee
fair distribution this would probably mean we await all reader's
registration to get the current assignment status before assigning any
splits, This might be okay though but needs some PoC and more details (for
example do we only start periodic discovery after reader's registration or
do we hold unassigned topics in state in non pending mode,...etc).
Regarding the distribution strategy interface, Does this mean you want to
expose the algorithm choice to users or do you intend to make it
configurable by source implementers only (if we are reusing for other
sources besides kafka)? I wouldn't vote to expose it to users but I would
love to reuse it for different connectors if that's the case.
Finally, given that restoring from state already redistributes splits
evenly, did you consider a simpler approach, like just a round-robin
assignment for all topic/partitions instead of hash based topic assignment
+ round-robin partition?
I am sorry if the discussion is getting more detailed, happy to move it to
a FLIP if the community agrees as well.
Best Regards
Ahmed Hamdy
On Mon, 30 Dec 2024 at 03:12, Hongshun Wang <lo...@gmail.com> wrote:
> Hi Ahmed,
> I want to do two things in this flip.
> 1. Add current assignments to ReaderRegistrationEvent. Thus, enumerator can
> know the overall assignments after restarting the jon.
> 2. Add an distribution strategy interface to the kafka enumerator, thus can
> flexibly choose how to assign the split based on the current circumstances.
>
>
> Best,
> Hongshun
>
> On Mon, Dec 30, 2024 at 10:03 AM Hongshun Wang <lo...@gmail.com>
> wrote:
>
> > Hi Ahmed,
> >
> > Thanks for your question.
> > > Do you intend to keep the current assignment in the enumerator
> > state or are you only going to use per session assignment status and keep
> > it in memory?
> >
> > Not keep in state, because it's be different with the real distribution
> > when redistribute state if the parrelism changed. When restart the job,
> > each task send an event to tell enumerator its assignments. It's also
> why I
> > want to add a FLIP rather than jira.
> >
> > Best,
> > Hongshun
> >
> > On Mon, Dec 30, 2024 at 12:42 AM Ahmed Hamdy <ha...@gmail.com>
> wrote:
> >
> >> Hi Hongshun
> >>
> >> There is currently a Jira tracking this issue [1].
> >>
> >> >Therefore, I want the enumerator to select a reader based on the
> current
> >> assignment situation. Each time, the enumerator should choose the reader
> >> whose assigned split number is the smallest.
> >>
> >> I am not sure I follow how you intend to track the "current split
> >> assignment". Do you intend to keep the current assignment in the
> >> enumerator
> >> state or are you only going to use per session assignment status and
> keep
> >> it in memory? If you intend to do the former, then how are you going to
> >> handle split redistribution upon parallelism change?
> >>
> >>
> >> 1- https://issues.apache.org/jira/browse/FLINK-31762
> >> Best Regards
> >> Ahmed Hamdy
> >>
> >>
> >> On Fri, 27 Dec 2024 at 06:38, Hongshun Wang <lo...@gmail.com>
> >> wrote:
> >>
> >> > Hi devs,
> >> >
> >> > I'd like to push a new FLIP xxx: Kafka Enumerator assigns splits based
> >> on
> >> > the current assignment situation. Currently, when the Kafka enumerator
> >> > finds a new split, it will choose a reader using the following
> >> algorithm:
> >> >
> >> >
> >> > ```java
> >> >
> >> > static int getSplitOwner(TopicPartition tp, int numReaders) {
> >> > int startIndex = ((tp.topic().hashCode() * 31) & 0x7FFFFFFF) %
> >> > numReaders;
> >> >
> >> > // here, the assumption is that the id of Kafka partitions are
> >> > always ascending
> >> > // starting from 0, and therefore can be used directly as the
> >> > offset clockwise from the
> >> > // start index
> >> > return (startIndex + tp.partition()) % numReaders;
> >> > }
> >> >
> >> > ```
> >> >
> >> >
> >> > However, I have encountered several unevenly distributed problems in
> >> > production:
> >> >
> >> >
> >> > - problem 1: If I consume multiple topics from a source, the
> >> > getSplitOwner will count the owner of partitions separately for
> each
> >> > topic. This means that there may be overlapping assignments. For
> >> > example,
> >> > if there are 3 Kafka topics with 16 partitions each, even if the
> >> Flink
> >> > task's parallelism is 48, only 16 tasks may be assigned 3
> partitions
> >> in
> >> > the
> >> > worst-case scenario.
> >> > - problem 2: If I use a Flink source with 16 parallelism to assumes
> >> one
> >> > topic with 16 partitions. Then, when I change the Flink parallelism
> >> to
> >> > 32,
> >> > the start index changes when a new partition is discovered because
> >> the
> >> > number of readers has changed.
> >> >
> >> >
> >> > Each time, I only have one method to solve it: wait for the job to
> >> complete
> >> > a snapshot state, then stop and restart the job with a different
> >> > parallelism. This way, the state redistribution strategy will assign
> the
> >> > splits as evenly as possible. However, this is too cumbersome—why
> can't
> >> it
> >> > be done properly when the enumerator discovers and assigns a new
> split?
> >> >
> >> >
> >> > Therefore, I want the enumerator to select a reader based on the
> current
> >> > assignment situation. Each time, the enumerator should choose the
> reader
> >> > whose assigned split number is the smallest.
> >> >
> >> >
> >> > WDYT? I'm willing to hear from you.
> >> >
> >> >
> >> > Best,
> >> >
> >> > Hongshun
> >> >
> >>
> >
>
Re: [Discuss] Kafka Enumerator assigns splits based on the current assignment situation.
Posted by Hongshun Wang <lo...@gmail.com>.
Hi Ahmed,
I want to do two things in this flip.
1. Add current assignments to ReaderRegistrationEvent. Thus, enumerator can
know the overall assignments after restarting the jon.
2. Add an distribution strategy interface to the kafka enumerator, thus can
flexibly choose how to assign the split based on the current circumstances.
Best,
Hongshun
On Mon, Dec 30, 2024 at 10:03 AM Hongshun Wang <lo...@gmail.com>
wrote:
> Hi Ahmed,
>
> Thanks for your question.
> > Do you intend to keep the current assignment in the enumerator
> state or are you only going to use per session assignment status and keep
> it in memory?
>
> Not keep in state, because it's be different with the real distribution
> when redistribute state if the parrelism changed. When restart the job,
> each task send an event to tell enumerator its assignments. It's also why I
> want to add a FLIP rather than jira.
>
> Best,
> Hongshun
>
> On Mon, Dec 30, 2024 at 12:42 AM Ahmed Hamdy <ha...@gmail.com> wrote:
>
>> Hi Hongshun
>>
>> There is currently a Jira tracking this issue [1].
>>
>> >Therefore, I want the enumerator to select a reader based on the current
>> assignment situation. Each time, the enumerator should choose the reader
>> whose assigned split number is the smallest.
>>
>> I am not sure I follow how you intend to track the "current split
>> assignment". Do you intend to keep the current assignment in the
>> enumerator
>> state or are you only going to use per session assignment status and keep
>> it in memory? If you intend to do the former, then how are you going to
>> handle split redistribution upon parallelism change?
>>
>>
>> 1- https://issues.apache.org/jira/browse/FLINK-31762
>> Best Regards
>> Ahmed Hamdy
>>
>>
>> On Fri, 27 Dec 2024 at 06:38, Hongshun Wang <lo...@gmail.com>
>> wrote:
>>
>> > Hi devs,
>> >
>> > I'd like to push a new FLIP xxx: Kafka Enumerator assigns splits based
>> on
>> > the current assignment situation. Currently, when the Kafka enumerator
>> > finds a new split, it will choose a reader using the following
>> algorithm:
>> >
>> >
>> > ```java
>> >
>> > static int getSplitOwner(TopicPartition tp, int numReaders) {
>> > int startIndex = ((tp.topic().hashCode() * 31) & 0x7FFFFFFF) %
>> > numReaders;
>> >
>> > // here, the assumption is that the id of Kafka partitions are
>> > always ascending
>> > // starting from 0, and therefore can be used directly as the
>> > offset clockwise from the
>> > // start index
>> > return (startIndex + tp.partition()) % numReaders;
>> > }
>> >
>> > ```
>> >
>> >
>> > However, I have encountered several unevenly distributed problems in
>> > production:
>> >
>> >
>> > - problem 1: If I consume multiple topics from a source, the
>> > getSplitOwner will count the owner of partitions separately for each
>> > topic. This means that there may be overlapping assignments. For
>> > example,
>> > if there are 3 Kafka topics with 16 partitions each, even if the
>> Flink
>> > task's parallelism is 48, only 16 tasks may be assigned 3 partitions
>> in
>> > the
>> > worst-case scenario.
>> > - problem 2: If I use a Flink source with 16 parallelism to assumes
>> one
>> > topic with 16 partitions. Then, when I change the Flink parallelism
>> to
>> > 32,
>> > the start index changes when a new partition is discovered because
>> the
>> > number of readers has changed.
>> >
>> >
>> > Each time, I only have one method to solve it: wait for the job to
>> complete
>> > a snapshot state, then stop and restart the job with a different
>> > parallelism. This way, the state redistribution strategy will assign the
>> > splits as evenly as possible. However, this is too cumbersome—why can't
>> it
>> > be done properly when the enumerator discovers and assigns a new split?
>> >
>> >
>> > Therefore, I want the enumerator to select a reader based on the current
>> > assignment situation. Each time, the enumerator should choose the reader
>> > whose assigned split number is the smallest.
>> >
>> >
>> > WDYT? I'm willing to hear from you.
>> >
>> >
>> > Best,
>> >
>> > Hongshun
>> >
>>
>
Re: [Discuss] Kafka Enumerator assigns splits based on the current assignment situation.
Posted by Hongshun Wang <lo...@gmail.com>.
Hi Ahmed,
Thanks for your question.
> Do you intend to keep the current assignment in the enumerator
state or are you only going to use per session assignment status and keep
it in memory?
Not keep in state, because it's be different with the real distribution
when redistribute state if the parrelism changed. When restart the job,
each task send an event to tell enumerator its assignments. It's also why I
want to add a FLIP rather than jira.
Best,
Hongshun
On Mon, Dec 30, 2024 at 12:42 AM Ahmed Hamdy <ha...@gmail.com> wrote:
> Hi Hongshun
>
> There is currently a Jira tracking this issue [1].
>
> >Therefore, I want the enumerator to select a reader based on the current
> assignment situation. Each time, the enumerator should choose the reader
> whose assigned split number is the smallest.
>
> I am not sure I follow how you intend to track the "current split
> assignment". Do you intend to keep the current assignment in the enumerator
> state or are you only going to use per session assignment status and keep
> it in memory? If you intend to do the former, then how are you going to
> handle split redistribution upon parallelism change?
>
>
> 1- https://issues.apache.org/jira/browse/FLINK-31762
> Best Regards
> Ahmed Hamdy
>
>
> On Fri, 27 Dec 2024 at 06:38, Hongshun Wang <lo...@gmail.com>
> wrote:
>
> > Hi devs,
> >
> > I'd like to push a new FLIP xxx: Kafka Enumerator assigns splits based on
> > the current assignment situation. Currently, when the Kafka enumerator
> > finds a new split, it will choose a reader using the following algorithm:
> >
> >
> > ```java
> >
> > static int getSplitOwner(TopicPartition tp, int numReaders) {
> > int startIndex = ((tp.topic().hashCode() * 31) & 0x7FFFFFFF) %
> > numReaders;
> >
> > // here, the assumption is that the id of Kafka partitions are
> > always ascending
> > // starting from 0, and therefore can be used directly as the
> > offset clockwise from the
> > // start index
> > return (startIndex + tp.partition()) % numReaders;
> > }
> >
> > ```
> >
> >
> > However, I have encountered several unevenly distributed problems in
> > production:
> >
> >
> > - problem 1: If I consume multiple topics from a source, the
> > getSplitOwner will count the owner of partitions separately for each
> > topic. This means that there may be overlapping assignments. For
> > example,
> > if there are 3 Kafka topics with 16 partitions each, even if the Flink
> > task's parallelism is 48, only 16 tasks may be assigned 3 partitions
> in
> > the
> > worst-case scenario.
> > - problem 2: If I use a Flink source with 16 parallelism to assumes
> one
> > topic with 16 partitions. Then, when I change the Flink parallelism to
> > 32,
> > the start index changes when a new partition is discovered because the
> > number of readers has changed.
> >
> >
> > Each time, I only have one method to solve it: wait for the job to
> complete
> > a snapshot state, then stop and restart the job with a different
> > parallelism. This way, the state redistribution strategy will assign the
> > splits as evenly as possible. However, this is too cumbersome—why can't
> it
> > be done properly when the enumerator discovers and assigns a new split?
> >
> >
> > Therefore, I want the enumerator to select a reader based on the current
> > assignment situation. Each time, the enumerator should choose the reader
> > whose assigned split number is the smallest.
> >
> >
> > WDYT? I'm willing to hear from you.
> >
> >
> > Best,
> >
> > Hongshun
> >
>
Re: [Discuss] Kafka Enumerator assigns splits based on the current assignment situation.
Posted by Ahmed Hamdy <ha...@gmail.com>.
Hi Hongshun
There is currently a Jira tracking this issue [1].
>Therefore, I want the enumerator to select a reader based on the current
assignment situation. Each time, the enumerator should choose the reader
whose assigned split number is the smallest.
I am not sure I follow how you intend to track the "current split
assignment". Do you intend to keep the current assignment in the enumerator
state or are you only going to use per session assignment status and keep
it in memory? If you intend to do the former, then how are you going to
handle split redistribution upon parallelism change?
1- https://issues.apache.org/jira/browse/FLINK-31762
Best Regards
Ahmed Hamdy
On Fri, 27 Dec 2024 at 06:38, Hongshun Wang <lo...@gmail.com> wrote:
> Hi devs,
>
> I'd like to push a new FLIP xxx: Kafka Enumerator assigns splits based on
> the current assignment situation. Currently, when the Kafka enumerator
> finds a new split, it will choose a reader using the following algorithm:
>
>
> ```java
>
> static int getSplitOwner(TopicPartition tp, int numReaders) {
> int startIndex = ((tp.topic().hashCode() * 31) & 0x7FFFFFFF) %
> numReaders;
>
> // here, the assumption is that the id of Kafka partitions are
> always ascending
> // starting from 0, and therefore can be used directly as the
> offset clockwise from the
> // start index
> return (startIndex + tp.partition()) % numReaders;
> }
>
> ```
>
>
> However, I have encountered several unevenly distributed problems in
> production:
>
>
> - problem 1: If I consume multiple topics from a source, the
> getSplitOwner will count the owner of partitions separately for each
> topic. This means that there may be overlapping assignments. For
> example,
> if there are 3 Kafka topics with 16 partitions each, even if the Flink
> task's parallelism is 48, only 16 tasks may be assigned 3 partitions in
> the
> worst-case scenario.
> - problem 2: If I use a Flink source with 16 parallelism to assumes one
> topic with 16 partitions. Then, when I change the Flink parallelism to
> 32,
> the start index changes when a new partition is discovered because the
> number of readers has changed.
>
>
> Each time, I only have one method to solve it: wait for the job to complete
> a snapshot state, then stop and restart the job with a different
> parallelism. This way, the state redistribution strategy will assign the
> splits as evenly as possible. However, this is too cumbersome—why can't it
> be done properly when the enumerator discovers and assigns a new split?
>
>
> Therefore, I want the enumerator to select a reader based on the current
> assignment situation. Each time, the enumerator should choose the reader
> whose assigned split number is the smallest.
>
>
> WDYT? I'm willing to hear from you.
>
>
> Best,
>
> Hongshun
>