Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Matthias J. Sax]

Thanks a lot Levani!

On 11/16/19 4:00 AM, Levani Kokhreidze wrote:
> Matthias,
> 
> Yes, I agree. KIP is updated: 
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+DSL+with+Connecting+Topic+Creation+and+Repartition+Hint
>  
> 
>  and follow-up JIRA ticket is linked in “Rejected Alternatives” section.
> 
> Thank you all for an interesting discussion.
> 
> Kind Regards,
> Levani
> 
>> On Nov 16, 2019, at 10:11 AM, Matthias J. Sax  wrote:
>>
>> Levani,
>>
>> do you agree to the current proposal? It's basically a de-scoping of the
>> already voted KIP. If you agree, could you update the KIP wiki page
>> accordingly, including the "Rejected Alternative" section (and mabye a
>> link to a follow up Jira ticket).
>>
>> Because it's a descope, and John and myself support it, there seems to
>> be no need to re-vote.
>>
>> @Sophie,John: thanks a lot for your thoughtful input!
>>
>>
>> -Matthias
>>
>> On 11/15/19 12:47 PM, John Roesler wrote:
>>> Thanks Sophie,
>>>
>>> I think your concern is valid, and also that your idea to make a
>>> ticket is a good idea.
>>>
>>> Creating a ticket has some very positive effects:
>>> * It allows us to record the thinking at this point in time so we
>>> don't have to dig through the mail archives later
>>> * It demonstrates that we did consider the use case, and do want to
>>> address it, but just don't feel confident to implement it right now.
>>> Then, if/when people do have a problem with the gap, the ticket it
>>> already there for them to consider, request, or even pick up.
>>>
>>> Since one aspect of the deferral is a desire to wait for real use
>>> experience, we should explicitly mention that in the ticket. This is
>>> just good information for people browsing the Jira looking for
>>> interesting tickets to pick up. They could still pick it up, but they
>>> can ask themselves if they really understand the real-world use cases
>>> any better than we do right now.
>>>
>>> Thanks, likewise, to you for the good discussion!
>>> -John
>>>
>>> On Fri, Nov 15, 2019 at 2:37 PM Sophie Blee-Goldman  
>>> wrote:

 While I'm concerned that "not augmenting groupBy as part of this KIP"
 really translates to "will not get around to augmenting groupBy for a long
 time if not as part of this KIP", like I said I don't want to hold up the
 new
 .repartition operator that it seems we do, at least, all agree on. It's a
 fair
 point that we can always add this in later, but undoing it is far more
 problematic.

 Anyways, I would be happy if we at least make a ticket to consider adding a
 "number of partitions" option/suggestion to groupBy, so that we don't lose
 all the thought put in to this decision so far and can avoid rehashing the
 same
 argument word for word and have something to point to when someone
 asks "why didn't we add this numPartitions option to groupBy".

 Beyond that, if the community isn't pushing for it at this moment then it
 seems very
 reasonable to shelve the idea for now so that the rest of this KIP can
 proceed.
 Without input one way or another it's hard to say what the right thing to
 do is,
 which makes the right thing to do "wait to add this feature"

 Thanks for the good discussion everyone,

 Sophie

 On Fri, Nov 15, 2019 at 12:41 PM John Roesler  wrote:

> Hi all,
>
> I think that Sophie is asking a good question, and I do think that
> such "blanket configurations" are plausible. For example, we currently
> support (and I would encourage) "I don't know if this is going to
> create a repartition topic, but if it does, then use this name instead
> of generating one".
>
> I'm not sure I'm convinced that specifying max parallelism falls into
> this category. After all, the groupByKey+aggregate will be executed
> with _some_ max parallelism. It's either the same as the inputs'
> partition count or overridden with the proposed config. It seems
> counterintuitive to override the specified option with the default
> value.
>
> I'm not sure if I can put my finger on it, but "maybe use this name"
> seems way more reasonable to me than "maybe execute with this degree
> of parallelism".
>
> I do think (and I appreciate that this is where Sophie's example is
> coming from) that Streams should strive to be absolutely as simple and
> intuitive as possible (while still maintaining correctness). Optimal
> performance can be at odds with API simplicity. For example, the
> simplest behavior is, if you ask for 5 partitions, you get 5
> partitions. Maybe a repartition is technically not necessary (if you
> didn't change the key), but at least there's no mystery to this
> behavior.
>
> Clearly, an 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Levani Kokhreidze]

Matthias,

Yes, I agree. KIP is updated: 
https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+DSL+with+Connecting+Topic+Creation+and+Repartition+Hint
 

 and follow-up JIRA ticket is linked in “Rejected Alternatives” section.

Thank you all for an interesting discussion.

Kind Regards,
Levani

> On Nov 16, 2019, at 10:11 AM, Matthias J. Sax  wrote:
> 
> Levani,
> 
> do you agree to the current proposal? It's basically a de-scoping of the
> already voted KIP. If you agree, could you update the KIP wiki page
> accordingly, including the "Rejected Alternative" section (and mabye a
> link to a follow up Jira ticket).
> 
> Because it's a descope, and John and myself support it, there seems to
> be no need to re-vote.
> 
> @Sophie,John: thanks a lot for your thoughtful input!
> 
> 
> -Matthias
> 
> On 11/15/19 12:47 PM, John Roesler wrote:
>> Thanks Sophie,
>> 
>> I think your concern is valid, and also that your idea to make a
>> ticket is a good idea.
>> 
>> Creating a ticket has some very positive effects:
>> * It allows us to record the thinking at this point in time so we
>> don't have to dig through the mail archives later
>> * It demonstrates that we did consider the use case, and do want to
>> address it, but just don't feel confident to implement it right now.
>> Then, if/when people do have a problem with the gap, the ticket it
>> already there for them to consider, request, or even pick up.
>> 
>> Since one aspect of the deferral is a desire to wait for real use
>> experience, we should explicitly mention that in the ticket. This is
>> just good information for people browsing the Jira looking for
>> interesting tickets to pick up. They could still pick it up, but they
>> can ask themselves if they really understand the real-world use cases
>> any better than we do right now.
>> 
>> Thanks, likewise, to you for the good discussion!
>> -John
>> 
>> On Fri, Nov 15, 2019 at 2:37 PM Sophie Blee-Goldman  
>> wrote:
>>> 
>>> While I'm concerned that "not augmenting groupBy as part of this KIP"
>>> really translates to "will not get around to augmenting groupBy for a long
>>> time if not as part of this KIP", like I said I don't want to hold up the
>>> new
>>> .repartition operator that it seems we do, at least, all agree on. It's a
>>> fair
>>> point that we can always add this in later, but undoing it is far more
>>> problematic.
>>> 
>>> Anyways, I would be happy if we at least make a ticket to consider adding a
>>> "number of partitions" option/suggestion to groupBy, so that we don't lose
>>> all the thought put in to this decision so far and can avoid rehashing the
>>> same
>>> argument word for word and have something to point to when someone
>>> asks "why didn't we add this numPartitions option to groupBy".
>>> 
>>> Beyond that, if the community isn't pushing for it at this moment then it
>>> seems very
>>> reasonable to shelve the idea for now so that the rest of this KIP can
>>> proceed.
>>> Without input one way or another it's hard to say what the right thing to
>>> do is,
>>> which makes the right thing to do "wait to add this feature"
>>> 
>>> Thanks for the good discussion everyone,
>>> 
>>> Sophie
>>> 
>>> On Fri, Nov 15, 2019 at 12:41 PM John Roesler  wrote:
>>> 
 Hi all,
 
 I think that Sophie is asking a good question, and I do think that
 such "blanket configurations" are plausible. For example, we currently
 support (and I would encourage) "I don't know if this is going to
 create a repartition topic, but if it does, then use this name instead
 of generating one".
 
 I'm not sure I'm convinced that specifying max parallelism falls into
 this category. After all, the groupByKey+aggregate will be executed
 with _some_ max parallelism. It's either the same as the inputs'
 partition count or overridden with the proposed config. It seems
 counterintuitive to override the specified option with the default
 value.
 
 I'm not sure if I can put my finger on it, but "maybe use this name"
 seems way more reasonable to me than "maybe execute with this degree
 of parallelism".
 
 I do think (and I appreciate that this is where Sophie's example is
 coming from) that Streams should strive to be absolutely as simple and
 intuitive as possible (while still maintaining correctness). Optimal
 performance can be at odds with API simplicity. For example, the
 simplest behavior is, if you ask for 5 partitions, you get 5
 partitions. Maybe a repartition is technically not necessary (if you
 didn't change the key), but at least there's no mystery to this
 behavior.
 
 Clearly, an (opposing) tenent of simplicity is trying to prevent
 people from making mistakes, which I think is what the example boils
 down to. Sometimes, we can prevent clear mistakes, 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Matthias J. Sax]

Levani,

do you agree to the current proposal? It's basically a de-scoping of the
already voted KIP. If you agree, could you update the KIP wiki page
accordingly, including the "Rejected Alternative" section (and mabye a
link to a follow up Jira ticket).

Because it's a descope, and John and myself support it, there seems to
be no need to re-vote.

@Sophie,John: thanks a lot for your thoughtful input!


-Matthias

On 11/15/19 12:47 PM, John Roesler wrote:
> Thanks Sophie,
> 
> I think your concern is valid, and also that your idea to make a
> ticket is a good idea.
> 
> Creating a ticket has some very positive effects:
> * It allows us to record the thinking at this point in time so we
> don't have to dig through the mail archives later
> * It demonstrates that we did consider the use case, and do want to
> address it, but just don't feel confident to implement it right now.
> Then, if/when people do have a problem with the gap, the ticket it
> already there for them to consider, request, or even pick up.
> 
> Since one aspect of the deferral is a desire to wait for real use
> experience, we should explicitly mention that in the ticket. This is
> just good information for people browsing the Jira looking for
> interesting tickets to pick up. They could still pick it up, but they
> can ask themselves if they really understand the real-world use cases
> any better than we do right now.
> 
> Thanks, likewise, to you for the good discussion!
> -John
> 
> On Fri, Nov 15, 2019 at 2:37 PM Sophie Blee-Goldman  
> wrote:
>>
>> While I'm concerned that "not augmenting groupBy as part of this KIP"
>> really translates to "will not get around to augmenting groupBy for a long
>> time if not as part of this KIP", like I said I don't want to hold up the
>> new
>> .repartition operator that it seems we do, at least, all agree on. It's a
>> fair
>> point that we can always add this in later, but undoing it is far more
>> problematic.
>>
>> Anyways, I would be happy if we at least make a ticket to consider adding a
>> "number of partitions" option/suggestion to groupBy, so that we don't lose
>> all the thought put in to this decision so far and can avoid rehashing the
>> same
>> argument word for word and have something to point to when someone
>> asks "why didn't we add this numPartitions option to groupBy".
>>
>> Beyond that, if the community isn't pushing for it at this moment then it
>> seems very
>> reasonable to shelve the idea for now so that the rest of this KIP can
>> proceed.
>> Without input one way or another it's hard to say what the right thing to
>> do is,
>> which makes the right thing to do "wait to add this feature"
>>
>> Thanks for the good discussion everyone,
>>
>> Sophie
>>
>> On Fri, Nov 15, 2019 at 12:41 PM John Roesler  wrote:
>>
>>> Hi all,
>>>
>>> I think that Sophie is asking a good question, and I do think that
>>> such "blanket configurations" are plausible. For example, we currently
>>> support (and I would encourage) "I don't know if this is going to
>>> create a repartition topic, but if it does, then use this name instead
>>> of generating one".
>>>
>>> I'm not sure I'm convinced that specifying max parallelism falls into
>>> this category. After all, the groupByKey+aggregate will be executed
>>> with _some_ max parallelism. It's either the same as the inputs'
>>> partition count or overridden with the proposed config. It seems
>>> counterintuitive to override the specified option with the default
>>> value.
>>>
>>> I'm not sure if I can put my finger on it, but "maybe use this name"
>>> seems way more reasonable to me than "maybe execute with this degree
>>> of parallelism".
>>>
>>> I do think (and I appreciate that this is where Sophie's example is
>>> coming from) that Streams should strive to be absolutely as simple and
>>> intuitive as possible (while still maintaining correctness). Optimal
>>> performance can be at odds with API simplicity. For example, the
>>> simplest behavior is, if you ask for 5 partitions, you get 5
>>> partitions. Maybe a repartition is technically not necessary (if you
>>> didn't change the key), but at least there's no mystery to this
>>> behavior.
>>>
>>> Clearly, an (opposing) tenent of simplicity is trying to prevent
>>> people from making mistakes, which I think is what the example boils
>>> down to. Sometimes, we can prevent clear mistakes, like equi-joining
>>> two topics with different partition counts. But for this case, it
>>> doesn't seem as clear-cut to be able to assume that they _said_ 5
>>> partitions, but they didn't really _want_ 5 partitions. Maybe we can
>>> just try to be clear in the documentation, and also even log a warning
>>> when we parse the topology, "hey, I've been asked to repartition this
>>> stream, but it's not necessary".
>>>
>>> If anything, this discussion really supports to me the value in just
>>> sticking with `repartition()` for now, and deferring
>>> `groupBy[Key](partitions)` to the future.
>>>
 Users should not 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[John Roesler]

Thanks Sophie,

I think your concern is valid, and also that your idea to make a
ticket is a good idea.

Creating a ticket has some very positive effects:
* It allows us to record the thinking at this point in time so we
don't have to dig through the mail archives later
* It demonstrates that we did consider the use case, and do want to
address it, but just don't feel confident to implement it right now.
Then, if/when people do have a problem with the gap, the ticket it
already there for them to consider, request, or even pick up.

Since one aspect of the deferral is a desire to wait for real use
experience, we should explicitly mention that in the ticket. This is
just good information for people browsing the Jira looking for
interesting tickets to pick up. They could still pick it up, but they
can ask themselves if they really understand the real-world use cases
any better than we do right now.

Thanks, likewise, to you for the good discussion!
-John

On Fri, Nov 15, 2019 at 2:37 PM Sophie Blee-Goldman  wrote:
>
> While I'm concerned that "not augmenting groupBy as part of this KIP"
> really translates to "will not get around to augmenting groupBy for a long
> time if not as part of this KIP", like I said I don't want to hold up the
> new
> .repartition operator that it seems we do, at least, all agree on. It's a
> fair
> point that we can always add this in later, but undoing it is far more
> problematic.
>
> Anyways, I would be happy if we at least make a ticket to consider adding a
> "number of partitions" option/suggestion to groupBy, so that we don't lose
> all the thought put in to this decision so far and can avoid rehashing the
> same
> argument word for word and have something to point to when someone
> asks "why didn't we add this numPartitions option to groupBy".
>
> Beyond that, if the community isn't pushing for it at this moment then it
> seems very
> reasonable to shelve the idea for now so that the rest of this KIP can
> proceed.
> Without input one way or another it's hard to say what the right thing to
> do is,
> which makes the right thing to do "wait to add this feature"
>
> Thanks for the good discussion everyone,
>
> Sophie
>
> On Fri, Nov 15, 2019 at 12:41 PM John Roesler  wrote:
>
> > Hi all,
> >
> > I think that Sophie is asking a good question, and I do think that
> > such "blanket configurations" are plausible. For example, we currently
> > support (and I would encourage) "I don't know if this is going to
> > create a repartition topic, but if it does, then use this name instead
> > of generating one".
> >
> > I'm not sure I'm convinced that specifying max parallelism falls into
> > this category. After all, the groupByKey+aggregate will be executed
> > with _some_ max parallelism. It's either the same as the inputs'
> > partition count or overridden with the proposed config. It seems
> > counterintuitive to override the specified option with the default
> > value.
> >
> > I'm not sure if I can put my finger on it, but "maybe use this name"
> > seems way more reasonable to me than "maybe execute with this degree
> > of parallelism".
> >
> > I do think (and I appreciate that this is where Sophie's example is
> > coming from) that Streams should strive to be absolutely as simple and
> > intuitive as possible (while still maintaining correctness). Optimal
> > performance can be at odds with API simplicity. For example, the
> > simplest behavior is, if you ask for 5 partitions, you get 5
> > partitions. Maybe a repartition is technically not necessary (if you
> > didn't change the key), but at least there's no mystery to this
> > behavior.
> >
> > Clearly, an (opposing) tenent of simplicity is trying to prevent
> > people from making mistakes, which I think is what the example boils
> > down to. Sometimes, we can prevent clear mistakes, like equi-joining
> > two topics with different partition counts. But for this case, it
> > doesn't seem as clear-cut to be able to assume that they _said_ 5
> > partitions, but they didn't really _want_ 5 partitions. Maybe we can
> > just try to be clear in the documentation, and also even log a warning
> > when we parse the topology, "hey, I've been asked to repartition this
> > stream, but it's not necessary".
> >
> > If anything, this discussion really supports to me the value in just
> > sticking with `repartition()` for now, and deferring
> > `groupBy[Key](partitions)` to the future.
> >
> > > Users should not have to choose between allowing Streams to optimize the
> > repartition placement, and allowing to specify a number of partitions.
> >
> > This is a very fair point, and it may be something that we rapidly
> > return to, but it seems safe for now to introduce the non-optimizable
> > `reparition()` only, and then consider optimization options later.
> > Skipping available optimizations will never break correctness, but
> > adding optimizations can, so it makes sense to treat them with
> > caution.
> >
> > In conclusion, I do think that a 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Sophie Blee-Goldman]

While I'm concerned that "not augmenting groupBy as part of this KIP"
really translates to "will not get around to augmenting groupBy for a long
time if not as part of this KIP", like I said I don't want to hold up the
new
.repartition operator that it seems we do, at least, all agree on. It's a
fair
point that we can always add this in later, but undoing it is far more
problematic.

Anyways, I would be happy if we at least make a ticket to consider adding a
"number of partitions" option/suggestion to groupBy, so that we don't lose
all the thought put in to this decision so far and can avoid rehashing the
same
argument word for word and have something to point to when someone
asks "why didn't we add this numPartitions option to groupBy".

Beyond that, if the community isn't pushing for it at this moment then it
seems very
reasonable to shelve the idea for now so that the rest of this KIP can
proceed.
Without input one way or another it's hard to say what the right thing to
do is,
which makes the right thing to do "wait to add this feature"

Thanks for the good discussion everyone,

Sophie

On Fri, Nov 15, 2019 at 12:41 PM John Roesler  wrote:

> Hi all,
>
> I think that Sophie is asking a good question, and I do think that
> such "blanket configurations" are plausible. For example, we currently
> support (and I would encourage) "I don't know if this is going to
> create a repartition topic, but if it does, then use this name instead
> of generating one".
>
> I'm not sure I'm convinced that specifying max parallelism falls into
> this category. After all, the groupByKey+aggregate will be executed
> with _some_ max parallelism. It's either the same as the inputs'
> partition count or overridden with the proposed config. It seems
> counterintuitive to override the specified option with the default
> value.
>
> I'm not sure if I can put my finger on it, but "maybe use this name"
> seems way more reasonable to me than "maybe execute with this degree
> of parallelism".
>
> I do think (and I appreciate that this is where Sophie's example is
> coming from) that Streams should strive to be absolutely as simple and
> intuitive as possible (while still maintaining correctness). Optimal
> performance can be at odds with API simplicity. For example, the
> simplest behavior is, if you ask for 5 partitions, you get 5
> partitions. Maybe a repartition is technically not necessary (if you
> didn't change the key), but at least there's no mystery to this
> behavior.
>
> Clearly, an (opposing) tenent of simplicity is trying to prevent
> people from making mistakes, which I think is what the example boils
> down to. Sometimes, we can prevent clear mistakes, like equi-joining
> two topics with different partition counts. But for this case, it
> doesn't seem as clear-cut to be able to assume that they _said_ 5
> partitions, but they didn't really _want_ 5 partitions. Maybe we can
> just try to be clear in the documentation, and also even log a warning
> when we parse the topology, "hey, I've been asked to repartition this
> stream, but it's not necessary".
>
> If anything, this discussion really supports to me the value in just
> sticking with `repartition()` for now, and deferring
> `groupBy[Key](partitions)` to the future.
>
> > Users should not have to choose between allowing Streams to optimize the
> repartition placement, and allowing to specify a number of partitions.
>
> This is a very fair point, and it may be something that we rapidly
> return to, but it seems safe for now to introduce the non-optimizable
> `reparition()` only, and then consider optimization options later.
> Skipping available optimizations will never break correctness, but
> adding optimizations can, so it makes sense to treat them with
> caution.
>
> In conclusion, I do think that a use _could_ want to "maybe specify"
> the partition count, but I also think we can afford to pass on
> supporting this right now.
>
> I'm open to continuing the discussion, but just to avoid ambiguity, I
> still feel we should _not_ change the groupBy[Key] operation at all,
> and we should only add `repartition()` as a non-optimizable operation.
>
> Thanks all,
> -John
>
> On Fri, Nov 15, 2019 at 11:26 AM Levani Kokhreidze
>  wrote:
> >
> > Hello,
> >
> > Just fyi, PR was updated and now it incorporates the latest suggestions
> about joins.
> > `CopartitionedTopicsEnforcer` will throw an exception if number of
> partitions aren’t the same when using `repartition` operation along with
> `join`.
> >
> > For more details please take a look at the PR:
> https://github.com/apache/kafka/pull/7170/files <
> https://github.com/apache/kafka/pull/7170/files>
> >
> > Regards,
> > Levani
> >
> >
> > > On Nov 15, 2019, at 11:01 AM, Matthias J. Sax 
> wrote:
> > >
> > > Thanks a lot for the input Sophie.
> > >
> > > Your example is quite useful, and I would use it to support my claim
> > > that a "partition hint" for `Grouped` seems "useless" and does not
> > > improve the user 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[John Roesler]

Hi all,

I think that Sophie is asking a good question, and I do think that
such "blanket configurations" are plausible. For example, we currently
support (and I would encourage) "I don't know if this is going to
create a repartition topic, but if it does, then use this name instead
of generating one".

I'm not sure I'm convinced that specifying max parallelism falls into
this category. After all, the groupByKey+aggregate will be executed
with _some_ max parallelism. It's either the same as the inputs'
partition count or overridden with the proposed config. It seems
counterintuitive to override the specified option with the default
value.

I'm not sure if I can put my finger on it, but "maybe use this name"
seems way more reasonable to me than "maybe execute with this degree
of parallelism".

I do think (and I appreciate that this is where Sophie's example is
coming from) that Streams should strive to be absolutely as simple and
intuitive as possible (while still maintaining correctness). Optimal
performance can be at odds with API simplicity. For example, the
simplest behavior is, if you ask for 5 partitions, you get 5
partitions. Maybe a repartition is technically not necessary (if you
didn't change the key), but at least there's no mystery to this
behavior.

Clearly, an (opposing) tenent of simplicity is trying to prevent
people from making mistakes, which I think is what the example boils
down to. Sometimes, we can prevent clear mistakes, like equi-joining
two topics with different partition counts. But for this case, it
doesn't seem as clear-cut to be able to assume that they _said_ 5
partitions, but they didn't really _want_ 5 partitions. Maybe we can
just try to be clear in the documentation, and also even log a warning
when we parse the topology, "hey, I've been asked to repartition this
stream, but it's not necessary".

If anything, this discussion really supports to me the value in just
sticking with `repartition()` for now, and deferring
`groupBy[Key](partitions)` to the future.

> Users should not have to choose between allowing Streams to optimize the 
> repartition placement, and allowing to specify a number of partitions.

This is a very fair point, and it may be something that we rapidly
return to, but it seems safe for now to introduce the non-optimizable
`reparition()` only, and then consider optimization options later.
Skipping available optimizations will never break correctness, but
adding optimizations can, so it makes sense to treat them with
caution.

In conclusion, I do think that a use _could_ want to "maybe specify"
the partition count, but I also think we can afford to pass on
supporting this right now.

I'm open to continuing the discussion, but just to avoid ambiguity, I
still feel we should _not_ change the groupBy[Key] operation at all,
and we should only add `repartition()` as a non-optimizable operation.

Thanks all,
-John

On Fri, Nov 15, 2019 at 11:26 AM Levani Kokhreidze
 wrote:
>
> Hello,
>
> Just fyi, PR was updated and now it incorporates the latest suggestions about 
> joins.
> `CopartitionedTopicsEnforcer` will throw an exception if number of partitions 
> aren’t the same when using `repartition` operation along with `join`.
>
> For more details please take a look at the PR: 
> https://github.com/apache/kafka/pull/7170/files 
> 
>
> Regards,
> Levani
>
>
> > On Nov 15, 2019, at 11:01 AM, Matthias J. Sax  wrote:
> >
> > Thanks a lot for the input Sophie.
> >
> > Your example is quite useful, and I would use it to support my claim
> > that a "partition hint" for `Grouped` seems "useless" and does not
> > improve the user experience.
> >
> > 1) You argue that a new user would be worries about repartitions topics
> > with too many paritions. This would imply that a user is already
> > advanced enough to understand the implication of repartitioning -- for
> > this case, I would argue that a user also understand _when_ a
> > auto-repartitioning would happen and thus the users understands where to
> > insert a `repartition()` operation.
> >
> > 2) For specifying Serdes: if a `groupByKey()` does not trigger
> > auto-repartitioning it's not required to specify the serdes and if they
> > are specified they would be ignored/unused (note, that `groupBy()` would
> > always trigger a repartitioning). Of course, if the default Serdes from
> > the config match (eg, all data types are Json anyway), a user does not
> > need to worry about specifying serdes. -- For new user that play around,
> > I would assume that they work a lot with primitive types and thus would
> > need to specify the serdes -- hence, they would learn about
> > auto-repartitioning the hard way anyhow, because each time a
> > `groupByKey()` does trigger auto-repartioning, they would need to pass
> > in the correct Serdes -- this way, they would also be educated where to
> > insert a `repartition()` operator if needed.
> >
> > 3) If a new user really just "plays 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Levani Kokhreidze]

Hello,

Just fyi, PR was updated and now it incorporates the latest suggestions about 
joins. 
`CopartitionedTopicsEnforcer` will throw an exception if number of partitions 
aren’t the same when using `repartition` operation along with `join`.

For more details please take a look at the PR: 
https://github.com/apache/kafka/pull/7170/files 


Regards,
Levani 


> On Nov 15, 2019, at 11:01 AM, Matthias J. Sax  wrote:
> 
> Thanks a lot for the input Sophie.
> 
> Your example is quite useful, and I would use it to support my claim
> that a "partition hint" for `Grouped` seems "useless" and does not
> improve the user experience.
> 
> 1) You argue that a new user would be worries about repartitions topics
> with too many paritions. This would imply that a user is already
> advanced enough to understand the implication of repartitioning -- for
> this case, I would argue that a user also understand _when_ a
> auto-repartitioning would happen and thus the users understands where to
> insert a `repartition()` operation.
> 
> 2) For specifying Serdes: if a `groupByKey()` does not trigger
> auto-repartitioning it's not required to specify the serdes and if they
> are specified they would be ignored/unused (note, that `groupBy()` would
> always trigger a repartitioning). Of course, if the default Serdes from
> the config match (eg, all data types are Json anyway), a user does not
> need to worry about specifying serdes. -- For new user that play around,
> I would assume that they work a lot with primitive types and thus would
> need to specify the serdes -- hence, they would learn about
> auto-repartitioning the hard way anyhow, because each time a
> `groupByKey()` does trigger auto-repartioning, they would need to pass
> in the correct Serdes -- this way, they would also be educated where to
> insert a `repartition()` operator if needed.
> 
> 3) If a new user really just "plays around", I don't think they use an
> input topic with 100 partitions but most likely have a local single node
> broker with most likely single partitions topics.
> 
> 
> My main argument for my current proposal is however, that---based on
> past experience---it's better to roll out a new feature more carefully
> and see how it goes. Last, as John pointed out, we can still extend the
> feature in the future. Instead of making a judgment call up-front, being
> more conservative and less fancy, and revisit the design based on
> actuall user feedback after the first version is rolled out, seems to be
> the better option. Undoing a feature is must harder than extending it.
> 
> 
> While I advocate strong for a simple first version of this feature, it's
> a community decission in the end, and I would not block this KIP if
> there is a broad preference to add `Grouped#withNumberOfPartitions()`
> either.
> 
> 
> -Matthias
> 
> On 11/14/19 11:35 PM, Sophie Blee-Goldman wrote:
>> It seems like we all agree at this point (please correct me if wrong!) that
>> we should NOT change
>> the existing repartitioning behavior, ie we should allow Streams to
>> continue to determine when and
>> where to repartition -- *unless* explicitly informed to by the use of a
>> .through or the new .repartition operator.
>> 
>> Regarding groupBy, the existing behavior we should not disrupt is
>> a) repartition *only* when required due to upstream key-changing operation
>> (ie don't force repartitioning
>> based on the presence of an optional config parameter), and
>> b) allow optimization of required repartitions, if any
>> 
>> Within the constraint of not breaking the existing behavior, this still
>> leaves open the question of whether we
>> want to improve the user experience by allowing to provide groupBy with a
>> *suggestion* for numPartitions (or to
>> put it more fairly, whether that *will* improve the experience). I agree
>> with many of the arguments outlined above but
>> let me just push back on this one issue one final time, and if we can't
>> come to a consensus then I am happy to drop
>> it for now so that the KIP can proceed.
>> 
>> Specifically, my proposal would be to simply augment Grouped with an
>> optional numPartitions, understood to
>> indicate the user's desired number of partitions *if Streams decides to
>> repartition due to that groupBy*
>> 
>>> if a user cares about the number of partition, the user wants to enforce
>> a repartitioning
>> First, I think we should take a step back and examine this claim. I agree
>> 100% that *if this is true,*
>> *then we should not give groupBy an optional numPartitions.* As far as I
>> see it, there's no argument
>> to be had there if we *presuppose that claim.* But I'm not convinced in
>> that as an axiom of the user
>> experience and think we should be examining that claim itself, not the
>> consequences of it.
>> 
>> To give a simple example, let's say some new user is trying out Streams and
>> wants to just play around
>> with it to see if it might be worth 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Matthias J. Sax]

Thanks a lot for the input Sophie.

Your example is quite useful, and I would use it to support my claim
that a "partition hint" for `Grouped` seems "useless" and does not
improve the user experience.

1) You argue that a new user would be worries about repartitions topics
with too many paritions. This would imply that a user is already
advanced enough to understand the implication of repartitioning -- for
this case, I would argue that a user also understand _when_ a
auto-repartitioning would happen and thus the users understands where to
insert a `repartition()` operation.

2) For specifying Serdes: if a `groupByKey()` does not trigger
auto-repartitioning it's not required to specify the serdes and if they
are specified they would be ignored/unused (note, that `groupBy()` would
always trigger a repartitioning). Of course, if the default Serdes from
the config match (eg, all data types are Json anyway), a user does not
need to worry about specifying serdes. -- For new user that play around,
I would assume that they work a lot with primitive types and thus would
need to specify the serdes -- hence, they would learn about
auto-repartitioning the hard way anyhow, because each time a
`groupByKey()` does trigger auto-repartioning, they would need to pass
in the correct Serdes -- this way, they would also be educated where to
insert a `repartition()` operator if needed.

3) If a new user really just "plays around", I don't think they use an
input topic with 100 partitions but most likely have a local single node
broker with most likely single partitions topics.


My main argument for my current proposal is however, that---based on
past experience---it's better to roll out a new feature more carefully
and see how it goes. Last, as John pointed out, we can still extend the
feature in the future. Instead of making a judgment call up-front, being
more conservative and less fancy, and revisit the design based on
actuall user feedback after the first version is rolled out, seems to be
the better option. Undoing a feature is must harder than extending it.


While I advocate strong for a simple first version of this feature, it's
a community decission in the end, and I would not block this KIP if
there is a broad preference to add `Grouped#withNumberOfPartitions()`
either.


-Matthias

On 11/14/19 11:35 PM, Sophie Blee-Goldman wrote:
> It seems like we all agree at this point (please correct me if wrong!) that
> we should NOT change
> the existing repartitioning behavior, ie we should allow Streams to
> continue to determine when and
> where to repartition -- *unless* explicitly informed to by the use of a
> .through or the new .repartition operator.
> 
> Regarding groupBy, the existing behavior we should not disrupt is
> a) repartition *only* when required due to upstream key-changing operation
> (ie don't force repartitioning
> based on the presence of an optional config parameter), and
> b) allow optimization of required repartitions, if any
> 
> Within the constraint of not breaking the existing behavior, this still
> leaves open the question of whether we
> want to improve the user experience by allowing to provide groupBy with a
> *suggestion* for numPartitions (or to
> put it more fairly, whether that *will* improve the experience). I agree
> with many of the arguments outlined above but
> let me just push back on this one issue one final time, and if we can't
> come to a consensus then I am happy to drop
> it for now so that the KIP can proceed.
> 
> Specifically, my proposal would be to simply augment Grouped with an
> optional numPartitions, understood to
> indicate the user's desired number of partitions *if Streams decides to
> repartition due to that groupBy*
> 
>> if a user cares about the number of partition, the user wants to enforce
> a repartitioning
> First, I think we should take a step back and examine this claim. I agree
> 100% that *if this is true,*
> *then we should not give groupBy an optional numPartitions.* As far as I
> see it, there's no argument
> to be had there if we *presuppose that claim.* But I'm not convinced in
> that as an axiom of the user
> experience and think we should be examining that claim itself, not the
> consequences of it.
> 
> To give a simple example, let's say some new user is trying out Streams and
> wants to just play around
> with it to see if it might be worth looking into. They want to just write
> up a simple app and test it out on the
> data in some existing topics they have with a large number of partitions,
> and a lot of data. They're just messing
> around, trying new topologies and don't want to go through each new one
> step by step to determine if (or where)
> a repartition might be required. They also don't want to force a
> repartition if it turns out to not be required, so they'd
> like to avoid the nice new .repartition operator they saw. But given the
> huge number of input partitions, they'd like
> to rest assured that if a repartition does end up 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Sophie Blee-Goldman]

It seems like we all agree at this point (please correct me if wrong!) that
we should NOT change
the existing repartitioning behavior, ie we should allow Streams to
continue to determine when and
where to repartition -- *unless* explicitly informed to by the use of a
.through or the new .repartition operator.

Regarding groupBy, the existing behavior we should not disrupt is
a) repartition *only* when required due to upstream key-changing operation
(ie don't force repartitioning
based on the presence of an optional config parameter), and
b) allow optimization of required repartitions, if any

Within the constraint of not breaking the existing behavior, this still
leaves open the question of whether we
want to improve the user experience by allowing to provide groupBy with a
*suggestion* for numPartitions (or to
put it more fairly, whether that *will* improve the experience). I agree
with many of the arguments outlined above but
let me just push back on this one issue one final time, and if we can't
come to a consensus then I am happy to drop
it for now so that the KIP can proceed.

Specifically, my proposal would be to simply augment Grouped with an
optional numPartitions, understood to
indicate the user's desired number of partitions *if Streams decides to
repartition due to that groupBy*

> if a user cares about the number of partition, the user wants to enforce
a repartitioning
First, I think we should take a step back and examine this claim. I agree
100% that *if this is true,*
*then we should not give groupBy an optional numPartitions.* As far as I
see it, there's no argument
to be had there if we *presuppose that claim.* But I'm not convinced in
that as an axiom of the user
experience and think we should be examining that claim itself, not the
consequences of it.

To give a simple example, let's say some new user is trying out Streams and
wants to just play around
with it to see if it might be worth looking into. They want to just write
up a simple app and test it out on the
data in some existing topics they have with a large number of partitions,
and a lot of data. They're just messing
around, trying new topologies and don't want to go through each new one
step by step to determine if (or where)
a repartition might be required. They also don't want to force a
repartition if it turns out to not be required, so they'd
like to avoid the nice new .repartition operator they saw. But given the
huge number of input partitions, they'd like
to rest assured that if a repartition does end up being required somewhere
during dev, it will not be created with
the same huge number of partitions that their input topic has -- so they
just pass groupBy a small numPartitions
suggestion.

I know that's a bit of a contrived example but I think it does highlight
how and when this might be a considerable
quality of life improvement, in particular for new users to Streams and/or
during the dev cycle. *You don't want to*
*force a repartition if it wasn't necessary, but you don't want to create a
topic with a huge partition count either.*

Also, while the optimization discussion took us down an interesting but
ultimately more distracting road, it's worth
pointing out that it is clearly a major win to have as few
repartition topics/steps as possible. Given that we
don't want to change existing behavior, the optimization framework can only
help out when the placement of
repartition steps is flexible, which means only those from .groupBy (and
not .repartition). *Users should not*
*have to choose between allowing Streams to optimize the repartition
placement, and allowing to specify a *
*number of partitions.*

Lastly, I have what may be a stupid question but for my own edification of
how groupBy works:
if you do a .groupBy and a repartition is NOT required, does it ever need
to serialize/deserialize
any of the data? In other words, if you pass a key/value serde to groupBy
and it doesn't trigger
a repartition, is the serde(s) just ignored and thus more like a suggestion
than a requirement?

So again, I don't want to hold up this KIP forever but I feel we've spent
some time getting slightly
off track (although certainly into very interesting discussions) yet never
really addressed or questioned
the basic premise: *could a user want to specify a number of partitions but
not enforce a repartition (at that*
*specific point in the topology)?*



On Fri, Nov 15, 2019 at 12:18 AM Matthias J. Sax 
wrote:

> Side remark:
>
> If the user specifies `repartition()` on both side of the join, we can
> actually throw the execption earlier, ie, when we build the topology.
>
> Current, we can do this check only after Kafka Streams was started,
> within `StreamPartitionAssignor#assign()` -- we still need to keep this
> check for the case that none or only one side has a user specified
> number of partitions though.
>
>
> -Matthias
>
> On 11/14/19 8:15 AM, John Roesler wrote:
> > Thanks, all,
> >
> > I can get behind just totally leaving out 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Matthias J. Sax]

Side remark:

If the user specifies `repartition()` on both side of the join, we can
actually throw the execption earlier, ie, when we build the topology.

Current, we can do this check only after Kafka Streams was started,
within `StreamPartitionAssignor#assign()` -- we still need to keep this
check for the case that none or only one side has a user specified
number of partitions though.


-Matthias

On 11/14/19 8:15 AM, John Roesler wrote:
> Thanks, all,
> 
> I can get behind just totally leaving out reparation-via-groupBy. If
> we only introduce `repartition()` for now, we're making the minimal
> change to gain the desired capability.
> 
> Plus, since we agree that `repartition()` should never be optimizable,
> it's a future-compatible proposal. I.e., if we were to add a
> non-optimizable groupBy(partitions) operation now, and want to make it
> optimizable in the future, we have to worry about topology
> compatibility. Better to just do non-optimizable `repartition()` now,
> and add an optimizable `groupBy(partitions)` in the future (maybe).
> 
> About joins, yes, it's a concern, and IMO we should just do the same
> thing we do now... check at runtime that the partition counts on both
> sides match and throw an exception otherwise. What this means as a
> user is that if you explicitly repartition the left side to 100
> partitions, and then join with the right side at 10 partitions, you
> get an exception, since this operation is not possible. You'd either
> have to "step down" the left side again, back to 10 partitions, or you
> could repartition the right side to 100 partitions before the join.
> The choice has to be the user's, since it depends on their desired
> execution parallelism.
> 
> Thanks,
> -John
> 
> On Thu, Nov 14, 2019 at 12:55 AM Matthias J. Sax  
> wrote:
>>
>> Thanks a lot John. I think the way you decompose the operators is super
>> helpful for this discussion.
>>
>> What you suggest with regard to using `Grouped` and enforcing
>> repartitioning if the number of partitions is specified is certainly
>> possible. However, I am not sure if we _should_ do this. My reasoning is
>> that an enforce repartitioning as introduced via `repartition()` is an
>> expensive operations, and it seems better to demand an more explicit
>> user opt-in to trigger it. Just setting an optional parameter might be
>> too subtle to trigger such a heavy "side effect".
>>
>> While I agree about "usability" in general, I would prefer a more
>> conservative appraoch to introduce this feature, see how it goes, and
>> maybe make it more advance later on. This also applies to what
>> optimzation we may or may not allow (or are able to perform at all).
>>
>> @Levani: Reflecting about my suggestion about `Repartioned extends
>> Grouped`, I agree that it might not be a good idea.
>>
>> Atm, I see an enforces repartitioning as non-optimizable and as a good
>> first step and I would suggest to not intoruce anything else for now.
>> Introducing optimizable enforce repartitioning via `groupBy(...,
>> Grouped)` is something we could add later.
>>
>>
>> Therefore, I would not change `Grouped` but only introduce
>> `repartition()`. Users that use `grouBy()` atm, and want to opt-in to
>> set the number of partitions, would need to rewrite their code to
>> `selectKey(...).repartition(...).groupByKey()`. It's less convinient but
>> also less risky from an API and optimization point of view.
>>
>>
>> @Levani: about joins -> yes, we will need to check the specified number
>> of partitions (if any) and if they don't match, throw an exception. We
>> can discuss this on the PR -- I am just trying to get the PR for KIP-466
>> merged -- your is next on the list :)
>>
>>
>> Thoughts?
>>
>>
>> -Matthias
>>
>>
>> On 11/12/19 4:51 PM, Levani Kokhreidze wrote:
>>> Thank you all for an interesting discussion. This is very enlightening.
>>>
>>> Thank you Matthias for your explanation. Your arguments are very true. It 
>>> makes sense that if user specifies number of partitions he/she really cares 
>>> that those specifications are applied to internal topics.
>>> Unfortunately, in current implementation this is not true during `join` 
>>> operation. As I’ve written in the PR comment, currently, when `Stream#join` 
>>> is used, `CopartitionedTopicsEnforcer` chooses max number of partitions 
>>> from the two source topics.
>>> I’m not really sure what would be the other way around this situation. 
>>> Maybe fail the stream altogether and inform the user to specify same number 
>>> of partitions?
>>> Or we should treat join operations in a same way as it is right now and 
>>> basically choose max number of partitions even when `repartition` operation 
>>> is specified, because Kafka Streams “knows the best” how to handle joins?
>>> You can check integration tests how it’s being handled currently. Open to 
>>> suggestions on that part.
>>>
>>> As for groupBy, I agree and John raised very interesting points. My 
>>> arguments for allowing users to 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[John Roesler]

Thanks, all,

I can get behind just totally leaving out reparation-via-groupBy. If
we only introduce `repartition()` for now, we're making the minimal
change to gain the desired capability.

Plus, since we agree that `repartition()` should never be optimizable,
it's a future-compatible proposal. I.e., if we were to add a
non-optimizable groupBy(partitions) operation now, and want to make it
optimizable in the future, we have to worry about topology
compatibility. Better to just do non-optimizable `repartition()` now,
and add an optimizable `groupBy(partitions)` in the future (maybe).

About joins, yes, it's a concern, and IMO we should just do the same
thing we do now... check at runtime that the partition counts on both
sides match and throw an exception otherwise. What this means as a
user is that if you explicitly repartition the left side to 100
partitions, and then join with the right side at 10 partitions, you
get an exception, since this operation is not possible. You'd either
have to "step down" the left side again, back to 10 partitions, or you
could repartition the right side to 100 partitions before the join.
The choice has to be the user's, since it depends on their desired
execution parallelism.

Thanks,
-John

On Thu, Nov 14, 2019 at 12:55 AM Matthias J. Sax  wrote:
>
> Thanks a lot John. I think the way you decompose the operators is super
> helpful for this discussion.
>
> What you suggest with regard to using `Grouped` and enforcing
> repartitioning if the number of partitions is specified is certainly
> possible. However, I am not sure if we _should_ do this. My reasoning is
> that an enforce repartitioning as introduced via `repartition()` is an
> expensive operations, and it seems better to demand an more explicit
> user opt-in to trigger it. Just setting an optional parameter might be
> too subtle to trigger such a heavy "side effect".
>
> While I agree about "usability" in general, I would prefer a more
> conservative appraoch to introduce this feature, see how it goes, and
> maybe make it more advance later on. This also applies to what
> optimzation we may or may not allow (or are able to perform at all).
>
> @Levani: Reflecting about my suggestion about `Repartioned extends
> Grouped`, I agree that it might not be a good idea.
>
> Atm, I see an enforces repartitioning as non-optimizable and as a good
> first step and I would suggest to not intoruce anything else for now.
> Introducing optimizable enforce repartitioning via `groupBy(...,
> Grouped)` is something we could add later.
>
>
> Therefore, I would not change `Grouped` but only introduce
> `repartition()`. Users that use `grouBy()` atm, and want to opt-in to
> set the number of partitions, would need to rewrite their code to
> `selectKey(...).repartition(...).groupByKey()`. It's less convinient but
> also less risky from an API and optimization point of view.
>
>
> @Levani: about joins -> yes, we will need to check the specified number
> of partitions (if any) and if they don't match, throw an exception. We
> can discuss this on the PR -- I am just trying to get the PR for KIP-466
> merged -- your is next on the list :)
>
>
> Thoughts?
>
>
> -Matthias
>
>
> On 11/12/19 4:51 PM, Levani Kokhreidze wrote:
> > Thank you all for an interesting discussion. This is very enlightening.
> >
> > Thank you Matthias for your explanation. Your arguments are very true. It 
> > makes sense that if user specifies number of partitions he/she really cares 
> > that those specifications are applied to internal topics.
> > Unfortunately, in current implementation this is not true during `join` 
> > operation. As I’ve written in the PR comment, currently, when `Stream#join` 
> > is used, `CopartitionedTopicsEnforcer` chooses max number of partitions 
> > from the two source topics.
> > I’m not really sure what would be the other way around this situation. 
> > Maybe fail the stream altogether and inform the user to specify same number 
> > of partitions?
> > Or we should treat join operations in a same way as it is right now and 
> > basically choose max number of partitions even when `repartition` operation 
> > is specified, because Kafka Streams “knows the best” how to handle joins?
> > You can check integration tests how it’s being handled currently. Open to 
> > suggestions on that part.
> >
> > As for groupBy, I agree and John raised very interesting points. My 
> > arguments for allowing users to specify number of partitions during groupBy 
> > operations mainly was coming from the usability perspective.
> > So building on top of what John said, maybe it makes sense to make 
> > `groupBy` operations smarter and whenever user specifies 
> > `numberOfPartitions` configuration, repartitioning will be enforced, wdyt?
> > I’m not going into optimization part yet :) I think it will be part of 
> > separate PR and task, but overall it makes sense to apply optimizations 
> > where number of partitions are the same.
> >
> > As for Repartitioned 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Matthias J. Sax]

Thanks a lot John. I think the way you decompose the operators is super
helpful for this discussion.

What you suggest with regard to using `Grouped` and enforcing
repartitioning if the number of partitions is specified is certainly
possible. However, I am not sure if we _should_ do this. My reasoning is
that an enforce repartitioning as introduced via `repartition()` is an
expensive operations, and it seems better to demand an more explicit
user opt-in to trigger it. Just setting an optional parameter might be
too subtle to trigger such a heavy "side effect".

While I agree about "usability" in general, I would prefer a more
conservative appraoch to introduce this feature, see how it goes, and
maybe make it more advance later on. This also applies to what
optimzation we may or may not allow (or are able to perform at all).

@Levani: Reflecting about my suggestion about `Repartioned extends
Grouped`, I agree that it might not be a good idea.

Atm, I see an enforces repartitioning as non-optimizable and as a good
first step and I would suggest to not intoruce anything else for now.
Introducing optimizable enforce repartitioning via `groupBy(...,
Grouped)` is something we could add later.


Therefore, I would not change `Grouped` but only introduce
`repartition()`. Users that use `grouBy()` atm, and want to opt-in to
set the number of partitions, would need to rewrite their code to
`selectKey(...).repartition(...).groupByKey()`. It's less convinient but
also less risky from an API and optimization point of view.


@Levani: about joins -> yes, we will need to check the specified number
of partitions (if any) and if they don't match, throw an exception. We
can discuss this on the PR -- I am just trying to get the PR for KIP-466
merged -- your is next on the list :)


Thoughts?


-Matthias


On 11/12/19 4:51 PM, Levani Kokhreidze wrote:
> Thank you all for an interesting discussion. This is very enlightening.
> 
> Thank you Matthias for your explanation. Your arguments are very true. It 
> makes sense that if user specifies number of partitions he/she really cares 
> that those specifications are applied to internal topics.
> Unfortunately, in current implementation this is not true during `join` 
> operation. As I’ve written in the PR comment, currently, when `Stream#join` 
> is used, `CopartitionedTopicsEnforcer` chooses max number of partitions from 
> the two source topics.
> I’m not really sure what would be the other way around this situation. Maybe 
> fail the stream altogether and inform the user to specify same number of 
> partitions? 
> Or we should treat join operations in a same way as it is right now and 
> basically choose max number of partitions even when `repartition` operation 
> is specified, because Kafka Streams “knows the best” how to handle joins?
> You can check integration tests how it’s being handled currently. Open to 
> suggestions on that part.
> 
> As for groupBy, I agree and John raised very interesting points. My arguments 
> for allowing users to specify number of partitions during groupBy operations 
> mainly was coming from the usability perspective.
> So building on top of what John said, maybe it makes sense to make `groupBy` 
> operations smarter and whenever user specifies `numberOfPartitions` 
> configuration, repartitioning will be enforced, wdyt?
> I’m not going into optimization part yet :) I think it will be part of 
> separate PR and task, but overall it makes sense to apply optimizations where 
> number of partitions are the same.
> 
> As for Repartitioned extending Grouped, I kinda feel that it won’t fit nicely 
> in current API design. 
> In addition, in the PR review, John mentioned that there were a lot of 
> troubles in the past trying to use one operation's configuration objects on 
> other operations. 
> Also it makes sense to keep them separate in terms of compatibility. 
> In that case, we don’t have to worry every time Grouped is changed, what 
> would be the implications on `repartition` operations.
> 
> Kind regards,
> Levani
> 
> 
>> On Nov 11, 2019, at 9:13 PM, John Roesler  wrote:
>>
>> Ah, thanks for the clarification. I missed your point.
>>
>> I like the framework you've presented. It does seem simpler to assume
>> that they either care about the partition count and want to
>> repartition to realize it, or they don't care about the number.
>> Returning to this discussion, it does seem unlikely that they care
>> about the number and _don't_ care if it actually gets realized.
>>
>> But then, it still seems like we can just keep the option as part of
>> Grouped. As in:
>>
>> // user does not care
>> stream.groupByKey(Grouped /*not specifying partition count*/)
>> stream.groupBy(Grouped /*not specifying partition count*/)
>>
>> // user does care
>> stream.repartition(Repartitioned)
>> stream.groupByKey(Grouped.numberOfPartitions(...))
>> stream.groupBy(Grouped.numberOfPartitions(...))
>>
>> 
>>
>> The above discussion got me thinking about 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Levani Kokhreidze]

Thank you all for an interesting discussion. This is very enlightening.

Thank you Matthias for your explanation. Your arguments are very true. It makes 
sense that if user specifies number of partitions he/she really cares that 
those specifications are applied to internal topics.
Unfortunately, in current implementation this is not true during `join` 
operation. As I’ve written in the PR comment, currently, when `Stream#join` is 
used, `CopartitionedTopicsEnforcer` chooses max number of partitions from the 
two source topics.
I’m not really sure what would be the other way around this situation. Maybe 
fail the stream altogether and inform the user to specify same number of 
partitions? 
Or we should treat join operations in a same way as it is right now and 
basically choose max number of partitions even when `repartition` operation is 
specified, because Kafka Streams “knows the best” how to handle joins?
You can check integration tests how it’s being handled currently. Open to 
suggestions on that part.

As for groupBy, I agree and John raised very interesting points. My arguments 
for allowing users to specify number of partitions during groupBy operations 
mainly was coming from the usability perspective.
So building on top of what John said, maybe it makes sense to make `groupBy` 
operations smarter and whenever user specifies `numberOfPartitions` 
configuration, repartitioning will be enforced, wdyt?
I’m not going into optimization part yet :) I think it will be part of separate 
PR and task, but overall it makes sense to apply optimizations where number of 
partitions are the same.

As for Repartitioned extending Grouped, I kinda feel that it won’t fit nicely 
in current API design. 
In addition, in the PR review, John mentioned that there were a lot of troubles 
in the past trying to use one operation's configuration objects on other 
operations. 
Also it makes sense to keep them separate in terms of compatibility. 
In that case, we don’t have to worry every time Grouped is changed, what would 
be the implications on `repartition` operations.

Kind regards,
Levani


> On Nov 11, 2019, at 9:13 PM, John Roesler  wrote:
> 
> Ah, thanks for the clarification. I missed your point.
> 
> I like the framework you've presented. It does seem simpler to assume
> that they either care about the partition count and want to
> repartition to realize it, or they don't care about the number.
> Returning to this discussion, it does seem unlikely that they care
> about the number and _don't_ care if it actually gets realized.
> 
> But then, it still seems like we can just keep the option as part of
> Grouped. As in:
> 
> // user does not care
> stream.groupByKey(Grouped /*not specifying partition count*/)
> stream.groupBy(Grouped /*not specifying partition count*/)
> 
> // user does care
> stream.repartition(Repartitioned)
> stream.groupByKey(Grouped.numberOfPartitions(...))
> stream.groupBy(Grouped.numberOfPartitions(...))
> 
> 
> 
> The above discussion got me thinking about algebra. Matthias is
> absolutely right that `groupByKey(numPartitions)` is equivalent to
> `repartition(numPartitions).groupByKey()`. I'm just not convinced that
> we should force people to apply that expansion themselves vs. having a
> more compact way to express it if they don't care where exactly the
> repartition occurs. However, thinking about these operators
> algebraically can really help *us* narrow down the number of different
> expressions we have to consider.
> 
> Let's consider some identities:
> 
> A: groupBy(mapper) + agg = mapKey(mapper) + groupByKey + agg
> B: src + ... + groupByKey + agg = src + ... + passthough + agg
> C: mapKey(mapper) + ... + groupByKey + agg
> = mapKey(mapper) + ... + repartition + groupByKey + agg
> D: repartition = sink(managed) + src
> 
> In these identities, I used one special identifier (...), which means
> any number (0+) of operations that are not src, mapKey, groupBy[Key],
> repartition, or agg.
> 
> For mental clarity, I'm just going to make up a rule that groupBy
> operations are not executable. In other words, you have to get to a
> point where you can apply B to convert a groupByKey into a passthough
> in order to execute the program. This is just a formal way of stating
> what already happens in Kafka Streams.
> 
> By applying A, we can just completely leave `groupBy` out of our
> analysis. It trivially decomposes into a mapKey followed by a
> groupByKey.
> 
> Then, we can eliminate the "repartition required" case of `groupByKey`
> by applying C followed by D to get to the "no repartition required"
> version of groupByKey, which in turn sets us up to apply B to get an
> executable topology.
> 
> Fundamentally, you can think about KIP-221 is as proposing a modified
> D identity in which you can specify the partition count of the managed
> sink topic:
> D': repartition(pc) = sink(managed w/ pc) + src
> 
> Since users _could_ apply the identities above, we don't actually have
> to add any 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[John Roesler]

Ah, thanks for the clarification. I missed your point.

I like the framework you've presented. It does seem simpler to assume
that they either care about the partition count and want to
repartition to realize it, or they don't care about the number.
Returning to this discussion, it does seem unlikely that they care
about the number and _don't_ care if it actually gets realized.

But then, it still seems like we can just keep the option as part of
Grouped. As in:

// user does not care
stream.groupByKey(Grouped /*not specifying partition count*/)
stream.groupBy(Grouped /*not specifying partition count*/)

// user does care
stream.repartition(Repartitioned)
stream.groupByKey(Grouped.numberOfPartitions(...))
stream.groupBy(Grouped.numberOfPartitions(...))



The above discussion got me thinking about algebra. Matthias is
absolutely right that `groupByKey(numPartitions)` is equivalent to
`repartition(numPartitions).groupByKey()`. I'm just not convinced that
we should force people to apply that expansion themselves vs. having a
more compact way to express it if they don't care where exactly the
repartition occurs. However, thinking about these operators
algebraically can really help *us* narrow down the number of different
expressions we have to consider.

Let's consider some identities:

A: groupBy(mapper) + agg = mapKey(mapper) + groupByKey + agg
B: src + ... + groupByKey + agg = src + ... + passthough + agg
C: mapKey(mapper) + ... + groupByKey + agg
 = mapKey(mapper) + ... + repartition + groupByKey + agg
D: repartition = sink(managed) + src

In these identities, I used one special identifier (...), which means
any number (0+) of operations that are not src, mapKey, groupBy[Key],
repartition, or agg.

For mental clarity, I'm just going to make up a rule that groupBy
operations are not executable. In other words, you have to get to a
point where you can apply B to convert a groupByKey into a passthough
in order to execute the program. This is just a formal way of stating
what already happens in Kafka Streams.

By applying A, we can just completely leave `groupBy` out of our
analysis. It trivially decomposes into a mapKey followed by a
groupByKey.

Then, we can eliminate the "repartition required" case of `groupByKey`
by applying C followed by D to get to the "no repartition required"
version of groupByKey, which in turn sets us up to apply B to get an
executable topology.

Fundamentally, you can think about KIP-221 is as proposing a modified
D identity in which you can specify the partition count of the managed
sink topic:
D': repartition(pc) = sink(managed w/ pc) + src

Since users _could_ apply the identities above, we don't actually have
to add any partition count to groupBy[Key], but we decided early on in
the KIP discussion that it's more ergonomic to add it. In that case,
we also have to modify A and C:
A': groupBy(mapper, pc) + agg
 = mapKey(mapper) + groupByKey(pc) + agg
C': mapKey(mapper) + ... + groupByKey(pc) + agg
 = mapKey(mapper) + ... + repartition(pc) + groupByKey + agg

Which sets us up still to always be able to get back to a plain
`groupByKey` operation (with no `(pc)`) and then apply D' and
ultimately B to get an executable topology.

What about the optimizer?
The optimizer applies another set of graph-algebraic identities to
minimize the number of repartition topics in a topology.

(forgive my ascii art)

E: (merging repartition nodes)
(...) -> repartition -> X
  \-> repartition -> Y
=
(... + repartition) -> X
 \-> Y
F: (reordering around repartition)
Where SVO is any non-key-changing, stateless, operation:
repartition -> SVO = SVO -> repartition

In terms of these identities, what the optimizer does is apply F
repeatedly in either direction to a topology to factor out common in
branches so that it can apply E to merge repartition nodes. This was
especially necessary before KIP-221 because you couldn't directly
express `repartition` in the DSL, only indirectly via `groupBy[Key]`,
so there was no way to do the factoring manually.

We can now state very clearly that in KIP-221, explicit
`repartition()` operators should create a "reordering barrier". So, F
cannot be applied to an explicit `repartition()`. Also, I think we
decided earlier that explicit `repartition()` operations would also be
ineligible for merging, so E can't be applied to explicit
`repartition()` operations either. I think we feel we _could_ apply E
without harm, but we want to be conservative for now.

I think the salient point from the latter discussion has been that
when you use `Grouped.numberOfPartitions`, this does _not_ constitute
an explicit `repartition()` operator, and therefore, the resulting
repartition node remains eligible for optimization.

To be clear, I agree with Matthias that the provided partition count
_must_ be used in the resulting implicit repartition. This has some
implications for E. Namely, E could only be applied to two repartition
nodes that have the same partition count. This has 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Matthias J. Sax]

> it seems like we do want to allow
>> people to optionally specify a partition count as part of this
>> operation, but we don't want that option to _force_ repartitioning

Correct, ie, that is my suggestions.

> "Use P partitions if repartitioning is necessary"

I disagree here, because my reasoning is that:

 - if a user cares about the number of partition, the user wants to
enforce a repartitioning
 - if a user does not case about the number of partitions, we don't need
to provide them a way to pass in a "hint"

Hence, it should be sufficient to support:

// user does not care

  `stream.groupByKey(Grouped)`
  `stream.grouBy(..., Grouped)`

// user does care

  `stream.repartition(Repartitioned).groupByKey()`
  `streams.groupBy(..., Repartitioned)`



-Matthias


On 11/9/19 8:10 PM, John Roesler wrote:
> Thanks for those thoughts, Matthias,
> 
> I find your reasoning about the optimization behavior compelling. The
> `through` operation is very simple and clear to reason about. It just
> passes the data exactly at the specified point in the topology exactly
> through the specified topic. Likewise, if a user invokes a
> `repartition` operator, the simplest behavior is if we just do what
> they asked for.
> 
> Stepping back to think about when optimizations are surprising and
> when they aren't, it occurs to me that we should be free to move
> around repartitions when users have asked to perform some operation
> that implies a repartition, like "change keys, then filter, then
> aggregate". This program requires a repartition, but it could be
> anywhere between the key change and the aggregation. On the other
> hand, if they say, "change keys, then filter, then repartition, then
> aggregate", it seems like they were pretty clear about their desire,
> and we should just take it at face value.
> 
> So, I'm sold on just literally doing a repartition every time they
> invoke the `repartition` operator.
> 
> 
> The "partition count" modifier for `groupBy`/`groupByKey` is more nuanced.
> 
> What you said about `groupByKey` makes sense. If they key hasn't
> actually changed, then we don't need to repartition before
> aggregating. On the other hand, `groupBy` is specifically changing the
> key as part of the grouping operation, so (as you said) we definitely
> have to do a repartition.
> 
> If I'm reading the discussion right, it seems like we do want to allow
> people to optionally specify a partition count as part of this
> operation, but we don't want that option to _force_ repartitioning if
> it's not needed. That last clause is the key. "Use P partitions if
> repartitioning is necessary" is a directive that applies cleanly and
> correctly to both `groupBy` and `groupByKey`. What if we call the
> option `numberOfPartitionsHint`, which along with the "if necessary"
> javadoc, should make it clear that the option won't force a
> repartition, and also gives us enough latitude to still employ the
> optimizer on those repartition topics?
> 
> If we like the idea of expressing it as a "hint" for grouping and a
> "command" for `repartition`, then it seems like it still makes sense
> to keep Grouped and Repartitioned separate, as they would actually
> offer different methods with distinct semantics.
> 
> WDYT?
> 
> Thanks,
> -John
> 
> On Sat, Nov 9, 2019 at 8:28 PM Matthias J. Sax  wrote:
>>
>> Sorry for late reply.
>>
>> I guess, the question boils down to the intended semantics of
>> `repartition()`. My understanding is as follows:
>>
>> - KS does auto-repartitioning for correctness reasons (using the
>> upstream topic to determine the number of partitions)
>> - KS does auto-repartitioning only for downstream DSL operators like
>> `count()` (eg, a `transform()` does never trigger an auto-repartitioning
>> even if the stream is marked as `repartitioningRequired`).
>> - KS offers `through()` to enforce a repartitioning -- however, the user
>> needs to create the topic manually (with the desired number of partitions).
>>
>> I see two main applications for `repartitioning()`:
>>
>> 1) repartition data before a `transform()` but user does not want to
>> manage the topic
>> 2) scale out a downstream subtopology
>>
>> Hence, I see `repartition()` similar to `through()`: if a users calls
>> it, a repartitining is enforced, with the difference that KS manages the
>> topic and the user does not need to create it.
>>
>> This behavior makes (1) and (2) possible.
>>
>>> I think many users would prefer to just say "if there *is* a repartition
>>> required at this point in the topology, it should
>>> have N partitions"
>>
>> Because of (2), I disagree. Either a user does not care about scaling
>> out, for which case she would not specify the number of partitions. Or a
>> user does care, and hence wants to enforce the scale out. I don't think
>> that any user would say, "maybe scale out".
>>
>> Therefore, the optimizer should never ignore the repartition operation.
>> As a "consequence" (because repartitioning is expensive) a user 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[John Roesler]

Thanks for those thoughts, Matthias,

I find your reasoning about the optimization behavior compelling. The
`through` operation is very simple and clear to reason about. It just
passes the data exactly at the specified point in the topology exactly
through the specified topic. Likewise, if a user invokes a
`repartition` operator, the simplest behavior is if we just do what
they asked for.

Stepping back to think about when optimizations are surprising and
when they aren't, it occurs to me that we should be free to move
around repartitions when users have asked to perform some operation
that implies a repartition, like "change keys, then filter, then
aggregate". This program requires a repartition, but it could be
anywhere between the key change and the aggregation. On the other
hand, if they say, "change keys, then filter, then repartition, then
aggregate", it seems like they were pretty clear about their desire,
and we should just take it at face value.

So, I'm sold on just literally doing a repartition every time they
invoke the `repartition` operator.


The "partition count" modifier for `groupBy`/`groupByKey` is more nuanced.

What you said about `groupByKey` makes sense. If they key hasn't
actually changed, then we don't need to repartition before
aggregating. On the other hand, `groupBy` is specifically changing the
key as part of the grouping operation, so (as you said) we definitely
have to do a repartition.

If I'm reading the discussion right, it seems like we do want to allow
people to optionally specify a partition count as part of this
operation, but we don't want that option to _force_ repartitioning if
it's not needed. That last clause is the key. "Use P partitions if
repartitioning is necessary" is a directive that applies cleanly and
correctly to both `groupBy` and `groupByKey`. What if we call the
option `numberOfPartitionsHint`, which along with the "if necessary"
javadoc, should make it clear that the option won't force a
repartition, and also gives us enough latitude to still employ the
optimizer on those repartition topics?

If we like the idea of expressing it as a "hint" for grouping and a
"command" for `repartition`, then it seems like it still makes sense
to keep Grouped and Repartitioned separate, as they would actually
offer different methods with distinct semantics.

WDYT?

Thanks,
-John

On Sat, Nov 9, 2019 at 8:28 PM Matthias J. Sax  wrote:
>
> Sorry for late reply.
>
> I guess, the question boils down to the intended semantics of
> `repartition()`. My understanding is as follows:
>
> - KS does auto-repartitioning for correctness reasons (using the
> upstream topic to determine the number of partitions)
> - KS does auto-repartitioning only for downstream DSL operators like
> `count()` (eg, a `transform()` does never trigger an auto-repartitioning
> even if the stream is marked as `repartitioningRequired`).
> - KS offers `through()` to enforce a repartitioning -- however, the user
> needs to create the topic manually (with the desired number of partitions).
>
> I see two main applications for `repartitioning()`:
>
> 1) repartition data before a `transform()` but user does not want to
> manage the topic
> 2) scale out a downstream subtopology
>
> Hence, I see `repartition()` similar to `through()`: if a users calls
> it, a repartitining is enforced, with the difference that KS manages the
> topic and the user does not need to create it.
>
> This behavior makes (1) and (2) possible.
>
> > I think many users would prefer to just say "if there *is* a repartition
> > required at this point in the topology, it should
> > have N partitions"
>
> Because of (2), I disagree. Either a user does not care about scaling
> out, for which case she would not specify the number of partitions. Or a
> user does care, and hence wants to enforce the scale out. I don't think
> that any user would say, "maybe scale out".
>
> Therefore, the optimizer should never ignore the repartition operation.
> As a "consequence" (because repartitioning is expensive) a user should
> make an explicit call to `repartition()` IMHO -- piggybacking an
> enforced repartitioning into `groupByKey()` seems to be "dangerous"
> because it might be too subtle and an "optional scaling out" as laid out
> above does not make sense IMHO.
>
> I am also not worried about "over repartitioning" because the result
> stream would never trigger auto-repartitioning. Only if multiple
> consecutive calls to `repartition()` are made it could be bad -- but
> that's the same with `through()`. In the end, there is always some
> responsibility on the user.
>
> Btw, for `.groupBy()` we know that repartitioning will be required,
> however, for `groupByKey()` it depends if the KStream is marked as
> `repartitioningRequired`.
>
> Hence, for `groupByKey()` it should not be possible for a user to set
> number of partitions IMHO. For `groupBy()` it's a different story,
> because calling
>
>`repartition().groupBy()`
>
> does not achieve what we 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Matthias J. Sax]

Sorry for late reply.

I guess, the question boils down to the intended semantics of
`repartition()`. My understanding is as follows:

- KS does auto-repartitioning for correctness reasons (using the
upstream topic to determine the number of partitions)
- KS does auto-repartitioning only for downstream DSL operators like
`count()` (eg, a `transform()` does never trigger an auto-repartitioning
even if the stream is marked as `repartitioningRequired`).
- KS offers `through()` to enforce a repartitioning -- however, the user
needs to create the topic manually (with the desired number of partitions).

I see two main applications for `repartitioning()`:

1) repartition data before a `transform()` but user does not want to
manage the topic
2) scale out a downstream subtopology

Hence, I see `repartition()` similar to `through()`: if a users calls
it, a repartitining is enforced, with the difference that KS manages the
topic and the user does not need to create it.

This behavior makes (1) and (2) possible.

> I think many users would prefer to just say "if there *is* a repartition
> required at this point in the topology, it should
> have N partitions"

Because of (2), I disagree. Either a user does not care about scaling
out, for which case she would not specify the number of partitions. Or a
user does care, and hence wants to enforce the scale out. I don't think
that any user would say, "maybe scale out".

Therefore, the optimizer should never ignore the repartition operation.
As a "consequence" (because repartitioning is expensive) a user should
make an explicit call to `repartition()` IMHO -- piggybacking an
enforced repartitioning into `groupByKey()` seems to be "dangerous"
because it might be too subtle and an "optional scaling out" as laid out
above does not make sense IMHO.

I am also not worried about "over repartitioning" because the result
stream would never trigger auto-repartitioning. Only if multiple
consecutive calls to `repartition()` are made it could be bad -- but
that's the same with `through()`. In the end, there is always some
responsibility on the user.

Btw, for `.groupBy()` we know that repartitioning will be required,
however, for `groupByKey()` it depends if the KStream is marked as
`repartitioningRequired`.

Hence, for `groupByKey()` it should not be possible for a user to set
number of partitions IMHO. For `groupBy()` it's a different story,
because calling

   `repartition().groupBy()`

does not achieve what we want. Hence, allowing users to pass in the
number of users partitions into `groupBy()` does actually makes sense,
because repartitioning will happen anyway and thus we can piggyback a
scaling decision.

I think that John has a fair concern about the overloads, however, I am
not convinced that using `Grouped` to specify the number of partitions
is intuitive. I double checked `Grouped` and `Repartitioned` and both
allow to specify a `name` and `keySerde/valueSerde`. Thus, I am
wondering if we could bridge the gap between both, if we would make
`Repartitioned extends Grouped`? For this case, we only need
`groupBy(Grouped)` and a user can pass in both types what seems to make
the API quite smooth:

  `stream.groupBy(..., Grouped...)`

  `stream.groupBy(..., Repartitioned...)`


Thoughts?


-Matthias



On 11/7/19 10:59 AM, Levani Kokhreidze wrote:
> Hi Sophie,
> 
> Thank you for your reply, very insightful. Looking forward hearing others 
> opinion as well on this.
> 
> Kind regards,
> Levani
> 
> 
>> On Nov 6, 2019, at 1:30 AM, Sophie Blee-Goldman  wrote:
>>
>>> Personally, I think Matthias’s concern is valid, but on the other hand
>> Kafka Streams has already
>>> optimizer in place which alters topology independently from user
>>
>> I agree (with you) and think this is a good way to put it -- we currently
>> auto-repartition for the user so
>> that they don't have to walk through their entire topology and reason about
>> when and where to place a
>> `.through` (or the new `.repartition`), so why suddenly force this onto the
>> user? How certain are we that
>> users will always get this right? It's easy to imagine that during
>> development, you write your new app with
>> correctly placed repartitions in order to use this new feature. During the
>> course of development you end up
>> tweaking the topology, but don't remember to review or move the
>> repartitioning since you're used to Streams
>> doing this for you. If you use only single-partition topics for testing,
>> you might not even notice your app is
>> spitting out incorrect results!
>>
>> Anyways, I feel pretty strongly that it would be weird to introduce a new
>> feature and say that to use it, you can't take
>> advantage of this other feature anymore. Also, is it possible our
>> optimization framework could ever include an
>> optimized repartitioning strategy that is better than what a user could
>> achieve by manually inserting repartitions?
>> Do we expect users to have a deep understanding of the best way to
>> 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Levani Kokhreidze]

Hi Sophie,

Thank you for your reply, very insightful. Looking forward hearing others 
opinion as well on this.

Kind regards,
Levani


> On Nov 6, 2019, at 1:30 AM, Sophie Blee-Goldman  wrote:
> 
>> Personally, I think Matthias’s concern is valid, but on the other hand
> Kafka Streams has already
>> optimizer in place which alters topology independently from user
> 
> I agree (with you) and think this is a good way to put it -- we currently
> auto-repartition for the user so
> that they don't have to walk through their entire topology and reason about
> when and where to place a
> `.through` (or the new `.repartition`), so why suddenly force this onto the
> user? How certain are we that
> users will always get this right? It's easy to imagine that during
> development, you write your new app with
> correctly placed repartitions in order to use this new feature. During the
> course of development you end up
> tweaking the topology, but don't remember to review or move the
> repartitioning since you're used to Streams
> doing this for you. If you use only single-partition topics for testing,
> you might not even notice your app is
> spitting out incorrect results!
> 
> Anyways, I feel pretty strongly that it would be weird to introduce a new
> feature and say that to use it, you can't take
> advantage of this other feature anymore. Also, is it possible our
> optimization framework could ever include an
> optimized repartitioning strategy that is better than what a user could
> achieve by manually inserting repartitions?
> Do we expect users to have a deep understanding of the best way to
> repartition their particular topology, or is it
> likely they will end up over-repartitioning either due to missed
> optimizations or unnecessary extra repartitions?
> I think many users would prefer to just say "if there *is* a repartition
> required at this point in the topology, it should
> have N partitions"
> 
> As to the idea of adding `numberOfPartitions` to Grouped rather than
> adding a new parameter to groupBy, that does seem more in line with the
> current syntax so +1 from me
> 
> On Tue, Nov 5, 2019 at 2:07 PM Levani Kokhreidze 
> wrote:
> 
>> Hello all,
>> 
>> While https://github.com/apache/kafka/pull/7170 <
>> https://github.com/apache/kafka/pull/7170> is under review and it’s
>> almost done, I want to resurrect discussion about this KIP to address
>> couple of concerns raised by Matthias and John.
>> 
>> As a reminder, idea of the KIP-221 was to allow DSL users control over
>> repartitioning and parallelism of sub-topologies by:
>> 1) Introducing new KStream#repartition operation which is done in
>> https://github.com/apache/kafka/pull/7170 <
>> https://github.com/apache/kafka/pull/7170>
>> 2) Add new KStream#groupBy(Repartitioned) operation, which is planned to
>> be separate PR.
>> 
>> While all agree about general implementation and idea behind
>> https://github.com/apache/kafka/pull/7170 <
>> https://github.com/apache/kafka/pull/7170> PR, introducing new
>> KStream#groupBy(Repartitioned) method overload raised some questions during
>> the review.
>> Matthias raised concern that there can be cases when user uses
>> `KStream#groupBy(Repartitioned)` operation, but actual repartitioning may
>> not required, thus configuration passed via `Repartitioned` would never be
>> applied (Matthias, please correct me if I misinterpreted your comment).
>> So instead, if user wants to control parallelism of sub-topologies, he or
>> she should always use `KStream#repartition` operation before groupBy. Full
>> comment can be seen here:
>> https://github.com/apache/kafka/pull/7170#issuecomment-519303125 <
>> https://github.com/apache/kafka/pull/7170#issuecomment-519303125>
>> 
>> On the same topic, John pointed out that, from API design perspective, we
>> shouldn’t intertwine configuration classes of different operators between
>> one another. So instead of introducing new `KStream#groupBy(Repartitioned)`
>> for specifying number of partitions for internal topic, we should update
>> existing `Grouped` class with `numberOfPartitions` field.
>> 
>> Personally, I think Matthias’s concern is valid, but on the other hand
>> Kafka Streams has already optimizer in place which alters topology
>> independently from user. So maybe it makes sense if Kafka Streams,
>> internally would optimize topology in the best way possible, even if in
>> some cases this means ignoring some operator configurations passed by the
>> user. Also, I agree with John about API design semantics. If we go through
>> with the changes for `KStream#groupBy` operation, it makes more sense to
>> add `numberOfPartitions` field to `Grouped` class instead of introducing
>> new `KStream#groupBy(Repartitioned)` method overload.
>> 
>> I would really appreciate communities feedback on this.
>> 
>> Kind regards,
>> Levani
>> 
>> 
>> 
>>> On Oct 17, 2019, at 12:57 AM, Sophie Blee-Goldman 
>> wrote:
>>> 
>>> Hey Levani,
>>> 
>>> I think people are busy with the 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Sophie Blee-Goldman]

> Personally, I think Matthias’s concern is valid, but on the other hand
Kafka Streams has already
> optimizer in place which alters topology independently from user

I agree (with you) and think this is a good way to put it -- we currently
auto-repartition for the user so
that they don't have to walk through their entire topology and reason about
when and where to place a
`.through` (or the new `.repartition`), so why suddenly force this onto the
user? How certain are we that
users will always get this right? It's easy to imagine that during
development, you write your new app with
correctly placed repartitions in order to use this new feature. During the
course of development you end up
tweaking the topology, but don't remember to review or move the
repartitioning since you're used to Streams
doing this for you. If you use only single-partition topics for testing,
you might not even notice your app is
spitting out incorrect results!

Anyways, I feel pretty strongly that it would be weird to introduce a new
feature and say that to use it, you can't take
advantage of this other feature anymore. Also, is it possible our
optimization framework could ever include an
optimized repartitioning strategy that is better than what a user could
achieve by manually inserting repartitions?
Do we expect users to have a deep understanding of the best way to
repartition their particular topology, or is it
likely they will end up over-repartitioning either due to missed
optimizations or unnecessary extra repartitions?
I think many users would prefer to just say "if there *is* a repartition
required at this point in the topology, it should
have N partitions"

As to the idea of adding `numberOfPartitions` to Grouped rather than
adding a new parameter to groupBy, that does seem more in line with the
current syntax so +1 from me

On Tue, Nov 5, 2019 at 2:07 PM Levani Kokhreidze 
wrote:

> Hello all,
>
> While https://github.com/apache/kafka/pull/7170 <
> https://github.com/apache/kafka/pull/7170> is under review and it’s
> almost done, I want to resurrect discussion about this KIP to address
> couple of concerns raised by Matthias and John.
>
> As a reminder, idea of the KIP-221 was to allow DSL users control over
> repartitioning and parallelism of sub-topologies by:
> 1) Introducing new KStream#repartition operation which is done in
> https://github.com/apache/kafka/pull/7170 <
> https://github.com/apache/kafka/pull/7170>
> 2) Add new KStream#groupBy(Repartitioned) operation, which is planned to
> be separate PR.
>
> While all agree about general implementation and idea behind
> https://github.com/apache/kafka/pull/7170 <
> https://github.com/apache/kafka/pull/7170> PR, introducing new
> KStream#groupBy(Repartitioned) method overload raised some questions during
> the review.
> Matthias raised concern that there can be cases when user uses
> `KStream#groupBy(Repartitioned)` operation, but actual repartitioning may
> not required, thus configuration passed via `Repartitioned` would never be
> applied (Matthias, please correct me if I misinterpreted your comment).
> So instead, if user wants to control parallelism of sub-topologies, he or
> she should always use `KStream#repartition` operation before groupBy. Full
> comment can be seen here:
> https://github.com/apache/kafka/pull/7170#issuecomment-519303125 <
> https://github.com/apache/kafka/pull/7170#issuecomment-519303125>
>
> On the same topic, John pointed out that, from API design perspective, we
> shouldn’t intertwine configuration classes of different operators between
> one another. So instead of introducing new `KStream#groupBy(Repartitioned)`
> for specifying number of partitions for internal topic, we should update
> existing `Grouped` class with `numberOfPartitions` field.
>
> Personally, I think Matthias’s concern is valid, but on the other hand
> Kafka Streams has already optimizer in place which alters topology
> independently from user. So maybe it makes sense if Kafka Streams,
> internally would optimize topology in the best way possible, even if in
> some cases this means ignoring some operator configurations passed by the
> user. Also, I agree with John about API design semantics. If we go through
> with the changes for `KStream#groupBy` operation, it makes more sense to
> add `numberOfPartitions` field to `Grouped` class instead of introducing
> new `KStream#groupBy(Repartitioned)` method overload.
>
> I would really appreciate communities feedback on this.
>
> Kind regards,
> Levani
>
>
>
> > On Oct 17, 2019, at 12:57 AM, Sophie Blee-Goldman 
> wrote:
> >
> > Hey Levani,
> >
> > I think people are busy with the upcoming 2.4 release, and don't have
> much
> > spare time at the
> > moment. It's kind of a difficult time to get attention on things, but
> feel
> > free to pick up something else
> > to work on in the meantime until things have calmed down a bit!
> >
> > Cheers,
> > Sophie
> >
> >
> > On Wed, Oct 16, 2019 at 11:26 AM Levani Kokhreidze <
> 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Levani Kokhreidze]

Hello all,

While https://github.com/apache/kafka/pull/7170 
 is under review and it’s almost 
done, I want to resurrect discussion about this KIP to address couple of 
concerns raised by Matthias and John.

As a reminder, idea of the KIP-221 was to allow DSL users control over 
repartitioning and parallelism of sub-topologies by:
1) Introducing new KStream#repartition operation which is done in 
https://github.com/apache/kafka/pull/7170 
 
2) Add new KStream#groupBy(Repartitioned) operation, which is planned to be 
separate PR.

While all agree about general implementation and idea behind 
https://github.com/apache/kafka/pull/7170 
 PR, introducing new 
KStream#groupBy(Repartitioned) method overload raised some questions during the 
review.
Matthias raised concern that there can be cases when user uses 
`KStream#groupBy(Repartitioned)` operation, but actual repartitioning may not 
required, thus configuration passed via `Repartitioned` would never be applied 
(Matthias, please correct me if I misinterpreted your comment). 
So instead, if user wants to control parallelism of sub-topologies, he or she 
should always use `KStream#repartition` operation before groupBy. Full comment 
can be seen here: 
https://github.com/apache/kafka/pull/7170#issuecomment-519303125 
 

On the same topic, John pointed out that, from API design perspective, we 
shouldn’t intertwine configuration classes of different operators between one 
another. So instead of introducing new `KStream#groupBy(Repartitioned)` for 
specifying number of partitions for internal topic, we should update existing 
`Grouped` class with `numberOfPartitions` field.

Personally, I think Matthias’s concern is valid, but on the other hand Kafka 
Streams has already optimizer in place which alters topology independently from 
user. So maybe it makes sense if Kafka Streams, internally would optimize 
topology in the best way possible, even if in some cases this means ignoring 
some operator configurations passed by the user. Also, I agree with John about 
API design semantics. If we go through with the changes for `KStream#groupBy` 
operation, it makes more sense to add `numberOfPartitions` field to `Grouped` 
class instead of introducing new `KStream#groupBy(Repartitioned)` method 
overload. 

I would really appreciate communities feedback on this.

Kind regards,
Levani



> On Oct 17, 2019, at 12:57 AM, Sophie Blee-Goldman  wrote:
> 
> Hey Levani,
> 
> I think people are busy with the upcoming 2.4 release, and don't have much
> spare time at the
> moment. It's kind of a difficult time to get attention on things, but feel
> free to pick up something else
> to work on in the meantime until things have calmed down a bit!
> 
> Cheers,
> Sophie
> 
> 
> On Wed, Oct 16, 2019 at 11:26 AM Levani Kokhreidze  >
> wrote:
> 
>> Hello all,
>> 
>> Sorry for bringing this thread again, but I would like to get some
>> attention on this PR: https://github.com/apache/kafka/pull/7170 
>>  <
>> https://github.com/apache/kafka/pull/7170 
>> >
>> It's been a while now and I would love to move on to other KIPs as well.
>> Please let me know if you have any concerns.
>> 
>> Regards,
>> Levani
>> 
>> 
>>> On Jul 26, 2019, at 11:25 AM, Levani Kokhreidze >> >
>> wrote:
>>> 
>>> Hi all,
>>> 
>>> Here’s voting thread for this KIP:
>> https://www.mail-archive.com/dev@kafka.apache.org/msg99680.html 
>>  <
>> https://www.mail-archive.com/dev@kafka.apache.org/msg99680.html 
>> >
>>> 
>>> Regards,
>>> Levani
>>> 
 On Jul 24, 2019, at 11:15 PM, Levani Kokhreidze >>> 
>> >> wrote:
 
 Hi Matthias,
 
 Thanks for the suggestion. I Don’t have strong opinion on that one.
 Agree that avoiding unnecessary method overloads is a good idea.
 
 Updated KIP
 
 Regards,
 Levani
 
 
> On Jul 24, 2019, at 8:50 PM, Matthias J. Sax  
>> >> wrote:
> 
> One question:
> 
> Why do we add
> 
>> Repartitioned#with(final String name, final int numberOfPartitions)
> 
> It seems that `#with(String name)`, `#numberOfPartitions(int)` in
> combination with `withName()` and `withNumberOfPartitions()` should be
> sufficient. Users can chain the method calls.
> 
> (I think it's valuable to keep the number of overload small if
>> possible.)
> 
> Otherwise LGTM.
> 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Sophie Blee-Goldman]

Hey Levani,

I think people are busy with the upcoming 2.4 release, and don't have much
spare time at the
moment. It's kind of a difficult time to get attention on things, but feel
free to pick up something else
to work on in the meantime until things have calmed down a bit!

Cheers,
Sophie


On Wed, Oct 16, 2019 at 11:26 AM Levani Kokhreidze 
wrote:

> Hello all,
>
> Sorry for bringing this thread again, but I would like to get some
> attention on this PR: https://github.com/apache/kafka/pull/7170 <
> https://github.com/apache/kafka/pull/7170>
> It's been a while now and I would love to move on to other KIPs as well.
> Please let me know if you have any concerns.
>
> Regards,
> Levani
>
>
> > On Jul 26, 2019, at 11:25 AM, Levani Kokhreidze 
> wrote:
> >
> > Hi all,
> >
> > Here’s voting thread for this KIP:
> https://www.mail-archive.com/dev@kafka.apache.org/msg99680.html <
> https://www.mail-archive.com/dev@kafka.apache.org/msg99680.html>
> >
> > Regards,
> > Levani
> >
> >> On Jul 24, 2019, at 11:15 PM, Levani Kokhreidze  > wrote:
> >>
> >> Hi Matthias,
> >>
> >> Thanks for the suggestion. I Don’t have strong opinion on that one.
> >> Agree that avoiding unnecessary method overloads is a good idea.
> >>
> >> Updated KIP
> >>
> >> Regards,
> >> Levani
> >>
> >>
> >>> On Jul 24, 2019, at 8:50 PM, Matthias J. Sax  > wrote:
> >>>
> >>> One question:
> >>>
> >>> Why do we add
> >>>
>  Repartitioned#with(final String name, final int numberOfPartitions)
> >>>
> >>> It seems that `#with(String name)`, `#numberOfPartitions(int)` in
> >>> combination with `withName()` and `withNumberOfPartitions()` should be
> >>> sufficient. Users can chain the method calls.
> >>>
> >>> (I think it's valuable to keep the number of overload small if
> possible.)
> >>>
> >>> Otherwise LGTM.
> >>>
> >>>
> >>> -Matthias
> >>>
> >>>
> >>> On 7/23/19 2:18 PM, Levani Kokhreidze wrote:
>  Hello,
> 
>  Thanks all for your feedback.
>  I started voting procedure for this KIP. If there’re any other
> concerns about this KIP, please let me know.
> 
>  Regards,
>  Levani
> 
> > On Jul 20, 2019, at 8:39 PM, Levani Kokhreidze <
> levani.co...@gmail.com > wrote:
> >
> > Hi Matthias,
> >
> > Thanks for the suggestion, makes sense.
> > I’ve updated KIP (
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> <
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint>
> <
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> <
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> >>).
> >
> > Regards,
> > Levani
> >
> >
> >> On Jul 20, 2019, at 3:53 AM, Matthias J. Sax   >> wrote:
> >>
> >> Thanks for driving the KIP.
> >>
> >> I agree that users need to be able to specify a partitioning
> strategy.
> >>
> >> Sophie raises a fair point about topic configs and producer
> configs. My
> >> take is, that consider `Repartitioned` as an "extension" to
> `Produced`,
> >> that adds topic configuration, is a good way to think about it and
> helps
> >> to keep the API "clean".
> >>
> >>
> >> With regard to method names. I would prefer to avoid abbreviations.
> Can
> >> we rename:
> >>
> >> `withNumOfPartitions` -> `withNumberOfPartitions`
> >>
> >> Furthermore, it might be good to add some more `static` methods:
> >>
> >> - Repartitioned.with(Serde, Serde)
> >> - Repartitioned.withNumberOfPartitions(int)
> >> - Repartitioned.streamPartitioner(StreamPartitioner)
> >>
> >>
> >> -Matthias
> >>
> >> On 7/19/19 3:33 PM, Levani Kokhreidze wrote:
> >>> Totally agree. I think in KStream interface it makes sense to have
> some duplicate configurations between operators in order to keep API simple
> and usable.
> >>> Also, as more surface API has, harder it is to have proper
> backward compatibility.
> >>> While initial idea of keeping topic level configs separate was
> exciting, having Repartitioned class encapsulate some producer level
> configs makes API more readable.
> >>>
> >>> Regards,
> >>> Levani
> >>>
>  On Jul 20, 2019, at 1:15 AM, Sophie Blee-Goldman <
> sop...@confluent.io   sop...@confluent.io >> wrote:
> 
>  I think that is a good point about trying to keep producer level
>  configurations and (repartition) topic level considerations
> separate.
>  Number of partitions is 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Levani Kokhreidze]

Hello all,

Sorry for bringing this thread again, but I would like to get some attention on 
this PR: https://github.com/apache/kafka/pull/7170 
 
It's been a while now and I would love to move on to other KIPs as well. Please 
let me know if you have any concerns.

Regards,
Levani


> On Jul 26, 2019, at 11:25 AM, Levani Kokhreidze  
> wrote:
> 
> Hi all,
> 
> Here’s voting thread for this KIP: 
> https://www.mail-archive.com/dev@kafka.apache.org/msg99680.html 
> 
> 
> Regards,
> Levani
> 
>> On Jul 24, 2019, at 11:15 PM, Levani Kokhreidze > > wrote:
>> 
>> Hi Matthias,
>> 
>> Thanks for the suggestion. I Don’t have strong opinion on that one.
>> Agree that avoiding unnecessary method overloads is a good idea.
>> 
>> Updated KIP
>> 
>> Regards,
>> Levani
>> 
>> 
>>> On Jul 24, 2019, at 8:50 PM, Matthias J. Sax >> > wrote:
>>> 
>>> One question:
>>> 
>>> Why do we add
>>> 
 Repartitioned#with(final String name, final int numberOfPartitions)
>>> 
>>> It seems that `#with(String name)`, `#numberOfPartitions(int)` in
>>> combination with `withName()` and `withNumberOfPartitions()` should be
>>> sufficient. Users can chain the method calls.
>>> 
>>> (I think it's valuable to keep the number of overload small if possible.)
>>> 
>>> Otherwise LGTM.
>>> 
>>> 
>>> -Matthias
>>> 
>>> 
>>> On 7/23/19 2:18 PM, Levani Kokhreidze wrote:
 Hello,
 
 Thanks all for your feedback.
 I started voting procedure for this KIP. If there’re any other concerns 
 about this KIP, please let me know.
 
 Regards,
 Levani
 
> On Jul 20, 2019, at 8:39 PM, Levani Kokhreidze  > wrote:
> 
> Hi Matthias,
> 
> Thanks for the suggestion, makes sense.
> I’ve updated KIP 
> (https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
>  
> 
>  
>   
> >).
> 
> Regards,
> Levani
> 
> 
>> On Jul 20, 2019, at 3:53 AM, Matthias J. Sax >  > >> wrote:
>> 
>> Thanks for driving the KIP.
>> 
>> I agree that users need to be able to specify a partitioning strategy.
>> 
>> Sophie raises a fair point about topic configs and producer configs. My
>> take is, that consider `Repartitioned` as an "extension" to `Produced`,
>> that adds topic configuration, is a good way to think about it and helps
>> to keep the API "clean".
>> 
>> 
>> With regard to method names. I would prefer to avoid abbreviations. Can
>> we rename:
>> 
>> `withNumOfPartitions` -> `withNumberOfPartitions`
>> 
>> Furthermore, it might be good to add some more `static` methods:
>> 
>> - Repartitioned.with(Serde, Serde)
>> - Repartitioned.withNumberOfPartitions(int)
>> - Repartitioned.streamPartitioner(StreamPartitioner)
>> 
>> 
>> -Matthias
>> 
>> On 7/19/19 3:33 PM, Levani Kokhreidze wrote:
>>> Totally agree. I think in KStream interface it makes sense to have some 
>>> duplicate configurations between operators in order to keep API simple 
>>> and usable.
>>> Also, as more surface API has, harder it is to have proper backward 
>>> compatibility.
>>> While initial idea of keeping topic level configs separate was 
>>> exciting, having Repartitioned class encapsulate some producer level 
>>> configs makes API more readable.
>>> 
>>> Regards,
>>> Levani
>>> 
 On Jul 20, 2019, at 1:15 AM, Sophie Blee-Goldman >>>  >> wrote:
 
 I think that is a good point about trying to keep producer level
 configurations and (repartition) topic level considerations separate.
 Number of partitions is definitely purely a topic level configuration. 
 But
 on some level, serdes and partitioners are just as much a topic
 configuration as a producer one. You could have two producers 
 configured
 with different serdes and/or partitioners, but if they are writing to 
 the
 same topic the result would be very difficult to part. So in a sense, 
 these
 are configurations of topics in Streams, not just 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Levani Kokhreidze]

Hi all,

Here’s voting thread for this KIP: 
https://www.mail-archive.com/dev@kafka.apache.org/msg99680.html 


Regards,
Levani

> On Jul 24, 2019, at 11:15 PM, Levani Kokhreidze  
> wrote:
> 
> Hi Matthias,
> 
> Thanks for the suggestion. I Don’t have strong opinion on that one.
> Agree that avoiding unnecessary method overloads is a good idea.
> 
> Updated KIP
> 
> Regards,
> Levani
> 
> 
>> On Jul 24, 2019, at 8:50 PM, Matthias J. Sax  wrote:
>> 
>> One question:
>> 
>> Why do we add
>> 
>>> Repartitioned#with(final String name, final int numberOfPartitions)
>> 
>> It seems that `#with(String name)`, `#numberOfPartitions(int)` in
>> combination with `withName()` and `withNumberOfPartitions()` should be
>> sufficient. Users can chain the method calls.
>> 
>> (I think it's valuable to keep the number of overload small if possible.)
>> 
>> Otherwise LGTM.
>> 
>> 
>> -Matthias
>> 
>> 
>> On 7/23/19 2:18 PM, Levani Kokhreidze wrote:
>>> Hello,
>>> 
>>> Thanks all for your feedback.
>>> I started voting procedure for this KIP. If there’re any other concerns 
>>> about this KIP, please let me know.
>>> 
>>> Regards,
>>> Levani
>>> 
 On Jul 20, 2019, at 8:39 PM, Levani Kokhreidze  
 wrote:
 
 Hi Matthias,
 
 Thanks for the suggestion, makes sense.
 I’ve updated KIP 
 (https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
  
 ).
 
 Regards,
 Levani
 
 
> On Jul 20, 2019, at 3:53 AM, Matthias J. Sax  > wrote:
> 
> Thanks for driving the KIP.
> 
> I agree that users need to be able to specify a partitioning strategy.
> 
> Sophie raises a fair point about topic configs and producer configs. My
> take is, that consider `Repartitioned` as an "extension" to `Produced`,
> that adds topic configuration, is a good way to think about it and helps
> to keep the API "clean".
> 
> 
> With regard to method names. I would prefer to avoid abbreviations. Can
> we rename:
> 
> `withNumOfPartitions` -> `withNumberOfPartitions`
> 
> Furthermore, it might be good to add some more `static` methods:
> 
> - Repartitioned.with(Serde, Serde)
> - Repartitioned.withNumberOfPartitions(int)
> - Repartitioned.streamPartitioner(StreamPartitioner)
> 
> 
> -Matthias
> 
> On 7/19/19 3:33 PM, Levani Kokhreidze wrote:
>> Totally agree. I think in KStream interface it makes sense to have some 
>> duplicate configurations between operators in order to keep API simple 
>> and usable.
>> Also, as more surface API has, harder it is to have proper backward 
>> compatibility.
>> While initial idea of keeping topic level configs separate was exciting, 
>> having Repartitioned class encapsulate some producer level configs makes 
>> API more readable.
>> 
>> Regards,
>> Levani
>> 
>>> On Jul 20, 2019, at 1:15 AM, Sophie Blee-Goldman >> > wrote:
>>> 
>>> I think that is a good point about trying to keep producer level
>>> configurations and (repartition) topic level considerations separate.
>>> Number of partitions is definitely purely a topic level configuration. 
>>> But
>>> on some level, serdes and partitioners are just as much a topic
>>> configuration as a producer one. You could have two producers configured
>>> with different serdes and/or partitioners, but if they are writing to 
>>> the
>>> same topic the result would be very difficult to part. So in a sense, 
>>> these
>>> are configurations of topics in Streams, not just producers.
>>> 
>>> Another way to think of it: while the Streams API is not always true to
>>> this, ideally all the relevant configs for an operator are wrapped into 
>>> a
>>> single object (in this case, Repartitioned). We could instead split out 
>>> the
>>> fields in common with Produced into a separate parameter to keep topic 
>>> and
>>> producer level configurations separate, but this increases the API 
>>> surface
>>> area by a lot. It's much more straightforward to just say "this is
>>> everything that this particular operator needs" without worrying about 
>>> what
>>> exactly you're specifying.
>>> 
>>> I suppose you could alternatively make Produced a field of 
>>> Repartitioned,
>>> but I don't think we do this kind of composition elsewhere in Streams at
>>> the moment
>>> 
>>> On Fri, Jul 19, 2019 at 1:45 PM Levani Kokhreidze 
>>> mailto:levani.co...@gmail.com>>
>>> wrote:
>>> 
 Hi Bill,
 
 Thanks a lot 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Levani Kokhreidze]

Hi Matthias,

Thanks for the suggestion. I Don’t have strong opinion on that one.
Agree that avoiding unnecessary method overloads is a good idea.

Updated KIP

Regards,
Levani


> On Jul 24, 2019, at 8:50 PM, Matthias J. Sax  wrote:
> 
> One question:
> 
> Why do we add
> 
>> Repartitioned#with(final String name, final int numberOfPartitions)
> 
> It seems that `#with(String name)`, `#numberOfPartitions(int)` in
> combination with `withName()` and `withNumberOfPartitions()` should be
> sufficient. Users can chain the method calls.
> 
> (I think it's valuable to keep the number of overload small if possible.)
> 
> Otherwise LGTM.
> 
> 
> -Matthias
> 
> 
> On 7/23/19 2:18 PM, Levani Kokhreidze wrote:
>> Hello,
>> 
>> Thanks all for your feedback.
>> I started voting procedure for this KIP. If there’re any other concerns 
>> about this KIP, please let me know.
>> 
>> Regards,
>> Levani
>> 
>>> On Jul 20, 2019, at 8:39 PM, Levani Kokhreidze  
>>> wrote:
>>> 
>>> Hi Matthias,
>>> 
>>> Thanks for the suggestion, makes sense.
>>> I’ve updated KIP 
>>> (https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
>>>  
>>> ).
>>> 
>>> Regards,
>>> Levani
>>> 
>>> 
 On Jul 20, 2019, at 3:53 AM, Matthias J. Sax >>> > wrote:
 
 Thanks for driving the KIP.
 
 I agree that users need to be able to specify a partitioning strategy.
 
 Sophie raises a fair point about topic configs and producer configs. My
 take is, that consider `Repartitioned` as an "extension" to `Produced`,
 that adds topic configuration, is a good way to think about it and helps
 to keep the API "clean".
 
 
 With regard to method names. I would prefer to avoid abbreviations. Can
 we rename:
 
 `withNumOfPartitions` -> `withNumberOfPartitions`
 
 Furthermore, it might be good to add some more `static` methods:
 
 - Repartitioned.with(Serde, Serde)
 - Repartitioned.withNumberOfPartitions(int)
 - Repartitioned.streamPartitioner(StreamPartitioner)
 
 
 -Matthias
 
 On 7/19/19 3:33 PM, Levani Kokhreidze wrote:
> Totally agree. I think in KStream interface it makes sense to have some 
> duplicate configurations between operators in order to keep API simple 
> and usable.
> Also, as more surface API has, harder it is to have proper backward 
> compatibility.
> While initial idea of keeping topic level configs separate was exciting, 
> having Repartitioned class encapsulate some producer level configs makes 
> API more readable.
> 
> Regards,
> Levani
> 
>> On Jul 20, 2019, at 1:15 AM, Sophie Blee-Goldman > > wrote:
>> 
>> I think that is a good point about trying to keep producer level
>> configurations and (repartition) topic level considerations separate.
>> Number of partitions is definitely purely a topic level configuration. 
>> But
>> on some level, serdes and partitioners are just as much a topic
>> configuration as a producer one. You could have two producers configured
>> with different serdes and/or partitioners, but if they are writing to the
>> same topic the result would be very difficult to part. So in a sense, 
>> these
>> are configurations of topics in Streams, not just producers.
>> 
>> Another way to think of it: while the Streams API is not always true to
>> this, ideally all the relevant configs for an operator are wrapped into a
>> single object (in this case, Repartitioned). We could instead split out 
>> the
>> fields in common with Produced into a separate parameter to keep topic 
>> and
>> producer level configurations separate, but this increases the API 
>> surface
>> area by a lot. It's much more straightforward to just say "this is
>> everything that this particular operator needs" without worrying about 
>> what
>> exactly you're specifying.
>> 
>> I suppose you could alternatively make Produced a field of Repartitioned,
>> but I don't think we do this kind of composition elsewhere in Streams at
>> the moment
>> 
>> On Fri, Jul 19, 2019 at 1:45 PM Levani Kokhreidze 
>> mailto:levani.co...@gmail.com>>
>> wrote:
>> 
>>> Hi Bill,
>>> 
>>> Thanks a lot for the feedback.
>>> Yes, that makes sense. I’ve updated KIP with `Repartitioned#partitioner`
>>> configuration.
>>> In the beginning, I wanted to introduce a class for topic level
>>> configuration and keep topic level and producer level configurations 
>>> (such
>>> as Produced) separately (see my second email in this thread).
>>> But while looking at the semantics of KStream interface, I couldn’t 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Matthias J. Sax]

One question:

Why do we add

> Repartitioned#with(final String name, final int numberOfPartitions)

It seems that `#with(String name)`, `#numberOfPartitions(int)` in
combination with `withName()` and `withNumberOfPartitions()` should be
sufficient. Users can chain the method calls.

(I think it's valuable to keep the number of overload small if possible.)

Otherwise LGTM.


-Matthias


On 7/23/19 2:18 PM, Levani Kokhreidze wrote:
> Hello,
> 
> Thanks all for your feedback.
> I started voting procedure for this KIP. If there’re any other concerns about 
> this KIP, please let me know.
> 
> Regards,
> Levani
> 
>> On Jul 20, 2019, at 8:39 PM, Levani Kokhreidze  
>> wrote:
>>
>> Hi Matthias,
>>
>> Thanks for the suggestion, makes sense.
>> I’ve updated KIP 
>> (https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
>>  
>> ).
>>
>> Regards,
>> Levani
>>
>>
>>> On Jul 20, 2019, at 3:53 AM, Matthias J. Sax >> > wrote:
>>>
>>> Thanks for driving the KIP.
>>>
>>> I agree that users need to be able to specify a partitioning strategy.
>>>
>>> Sophie raises a fair point about topic configs and producer configs. My
>>> take is, that consider `Repartitioned` as an "extension" to `Produced`,
>>> that adds topic configuration, is a good way to think about it and helps
>>> to keep the API "clean".
>>>
>>>
>>> With regard to method names. I would prefer to avoid abbreviations. Can
>>> we rename:
>>>
>>> `withNumOfPartitions` -> `withNumberOfPartitions`
>>>
>>> Furthermore, it might be good to add some more `static` methods:
>>>
>>> - Repartitioned.with(Serde, Serde)
>>> - Repartitioned.withNumberOfPartitions(int)
>>> - Repartitioned.streamPartitioner(StreamPartitioner)
>>>
>>>
>>> -Matthias
>>>
>>> On 7/19/19 3:33 PM, Levani Kokhreidze wrote:
 Totally agree. I think in KStream interface it makes sense to have some 
 duplicate configurations between operators in order to keep API simple and 
 usable.
 Also, as more surface API has, harder it is to have proper backward 
 compatibility.
 While initial idea of keeping topic level configs separate was exciting, 
 having Repartitioned class encapsulate some producer level configs makes 
 API more readable.

 Regards,
 Levani

> On Jul 20, 2019, at 1:15 AM, Sophie Blee-Goldman  > wrote:
>
> I think that is a good point about trying to keep producer level
> configurations and (repartition) topic level considerations separate.
> Number of partitions is definitely purely a topic level configuration. But
> on some level, serdes and partitioners are just as much a topic
> configuration as a producer one. You could have two producers configured
> with different serdes and/or partitioners, but if they are writing to the
> same topic the result would be very difficult to part. So in a sense, 
> these
> are configurations of topics in Streams, not just producers.
>
> Another way to think of it: while the Streams API is not always true to
> this, ideally all the relevant configs for an operator are wrapped into a
> single object (in this case, Repartitioned). We could instead split out 
> the
> fields in common with Produced into a separate parameter to keep topic and
> producer level configurations separate, but this increases the API surface
> area by a lot. It's much more straightforward to just say "this is
> everything that this particular operator needs" without worrying about 
> what
> exactly you're specifying.
>
> I suppose you could alternatively make Produced a field of Repartitioned,
> but I don't think we do this kind of composition elsewhere in Streams at
> the moment
>
> On Fri, Jul 19, 2019 at 1:45 PM Levani Kokhreidze  >
> wrote:
>
>> Hi Bill,
>>
>> Thanks a lot for the feedback.
>> Yes, that makes sense. I’ve updated KIP with `Repartitioned#partitioner`
>> configuration.
>> In the beginning, I wanted to introduce a class for topic level
>> configuration and keep topic level and producer level configurations 
>> (such
>> as Produced) separately (see my second email in this thread).
>> But while looking at the semantics of KStream interface, I couldn’t 
>> really
>> figure out good operation name for Topic level configuration class and 
>> just
>> introducing `Topic` config class was kinda breaking the semantics.
>> So I think having Repartitioned class which encapsulates topic and
>> producer level configurations for internal topics is viable thing to do.
>>
>> Regards,
>> Levani
>>
>>> On Jul 19, 2019, at 7:47 PM, Bill Bejeck 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Levani Kokhreidze]

Hello,

Thanks all for your feedback.
I started voting procedure for this KIP. If there’re any other concerns about 
this KIP, please let me know.

Regards,
Levani

> On Jul 20, 2019, at 8:39 PM, Levani Kokhreidze  wrote:
> 
> Hi Matthias,
> 
> Thanks for the suggestion, makes sense.
> I’ve updated KIP 
> (https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
>  
> ).
> 
> Regards,
> Levani
> 
> 
>> On Jul 20, 2019, at 3:53 AM, Matthias J. Sax > > wrote:
>> 
>> Thanks for driving the KIP.
>> 
>> I agree that users need to be able to specify a partitioning strategy.
>> 
>> Sophie raises a fair point about topic configs and producer configs. My
>> take is, that consider `Repartitioned` as an "extension" to `Produced`,
>> that adds topic configuration, is a good way to think about it and helps
>> to keep the API "clean".
>> 
>> 
>> With regard to method names. I would prefer to avoid abbreviations. Can
>> we rename:
>> 
>> `withNumOfPartitions` -> `withNumberOfPartitions`
>> 
>> Furthermore, it might be good to add some more `static` methods:
>> 
>> - Repartitioned.with(Serde, Serde)
>> - Repartitioned.withNumberOfPartitions(int)
>> - Repartitioned.streamPartitioner(StreamPartitioner)
>> 
>> 
>> -Matthias
>> 
>> On 7/19/19 3:33 PM, Levani Kokhreidze wrote:
>>> Totally agree. I think in KStream interface it makes sense to have some 
>>> duplicate configurations between operators in order to keep API simple and 
>>> usable.
>>> Also, as more surface API has, harder it is to have proper backward 
>>> compatibility.
>>> While initial idea of keeping topic level configs separate was exciting, 
>>> having Repartitioned class encapsulate some producer level configs makes 
>>> API more readable.
>>> 
>>> Regards,
>>> Levani
>>> 
 On Jul 20, 2019, at 1:15 AM, Sophie Blee-Goldman >>> > wrote:
 
 I think that is a good point about trying to keep producer level
 configurations and (repartition) topic level considerations separate.
 Number of partitions is definitely purely a topic level configuration. But
 on some level, serdes and partitioners are just as much a topic
 configuration as a producer one. You could have two producers configured
 with different serdes and/or partitioners, but if they are writing to the
 same topic the result would be very difficult to part. So in a sense, these
 are configurations of topics in Streams, not just producers.
 
 Another way to think of it: while the Streams API is not always true to
 this, ideally all the relevant configs for an operator are wrapped into a
 single object (in this case, Repartitioned). We could instead split out the
 fields in common with Produced into a separate parameter to keep topic and
 producer level configurations separate, but this increases the API surface
 area by a lot. It's much more straightforward to just say "this is
 everything that this particular operator needs" without worrying about what
 exactly you're specifying.
 
 I suppose you could alternatively make Produced a field of Repartitioned,
 but I don't think we do this kind of composition elsewhere in Streams at
 the moment
 
 On Fri, Jul 19, 2019 at 1:45 PM Levani Kokhreidze >>> >
 wrote:
 
> Hi Bill,
> 
> Thanks a lot for the feedback.
> Yes, that makes sense. I’ve updated KIP with `Repartitioned#partitioner`
> configuration.
> In the beginning, I wanted to introduce a class for topic level
> configuration and keep topic level and producer level configurations (such
> as Produced) separately (see my second email in this thread).
> But while looking at the semantics of KStream interface, I couldn’t really
> figure out good operation name for Topic level configuration class and 
> just
> introducing `Topic` config class was kinda breaking the semantics.
> So I think having Repartitioned class which encapsulates topic and
> producer level configurations for internal topics is viable thing to do.
> 
> Regards,
> Levani
> 
>> On Jul 19, 2019, at 7:47 PM, Bill Bejeck > > wrote:
>> 
>> Hi Lavani,
>> 
>> Thanks for resurrecting this KIP.
>> 
>> I'm also a +1 for adding a partition option.  In addition to the reason
>> provided by John, my reasoning is:
>> 
>> 1. Users may want to use something other than hash-based partitioning
>> 2. Users may wish to partition on something different than the key
>> without having to change the key.  For example:
>>1. A combination of fields in the value in conjunction with the key
>>2. Something other than 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Levani Kokhreidze]

Hi Matthias,

Thanks for the suggestion, makes sense.
I’ve updated KIP 
(https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
 
).

Regards,
Levani


> On Jul 20, 2019, at 3:53 AM, Matthias J. Sax  wrote:
> 
> Thanks for driving the KIP.
> 
> I agree that users need to be able to specify a partitioning strategy.
> 
> Sophie raises a fair point about topic configs and producer configs. My
> take is, that consider `Repartitioned` as an "extension" to `Produced`,
> that adds topic configuration, is a good way to think about it and helps
> to keep the API "clean".
> 
> 
> With regard to method names. I would prefer to avoid abbreviations. Can
> we rename:
> 
> `withNumOfPartitions` -> `withNumberOfPartitions`
> 
> Furthermore, it might be good to add some more `static` methods:
> 
> - Repartitioned.with(Serde, Serde)
> - Repartitioned.withNumberOfPartitions(int)
> - Repartitioned.streamPartitioner(StreamPartitioner)
> 
> 
> -Matthias
> 
> On 7/19/19 3:33 PM, Levani Kokhreidze wrote:
>> Totally agree. I think in KStream interface it makes sense to have some 
>> duplicate configurations between operators in order to keep API simple and 
>> usable.
>> Also, as more surface API has, harder it is to have proper backward 
>> compatibility.
>> While initial idea of keeping topic level configs separate was exciting, 
>> having Repartitioned class encapsulate some producer level configs makes API 
>> more readable.
>> 
>> Regards,
>> Levani
>> 
>>> On Jul 20, 2019, at 1:15 AM, Sophie Blee-Goldman  
>>> wrote:
>>> 
>>> I think that is a good point about trying to keep producer level
>>> configurations and (repartition) topic level considerations separate.
>>> Number of partitions is definitely purely a topic level configuration. But
>>> on some level, serdes and partitioners are just as much a topic
>>> configuration as a producer one. You could have two producers configured
>>> with different serdes and/or partitioners, but if they are writing to the
>>> same topic the result would be very difficult to part. So in a sense, these
>>> are configurations of topics in Streams, not just producers.
>>> 
>>> Another way to think of it: while the Streams API is not always true to
>>> this, ideally all the relevant configs for an operator are wrapped into a
>>> single object (in this case, Repartitioned). We could instead split out the
>>> fields in common with Produced into a separate parameter to keep topic and
>>> producer level configurations separate, but this increases the API surface
>>> area by a lot. It's much more straightforward to just say "this is
>>> everything that this particular operator needs" without worrying about what
>>> exactly you're specifying.
>>> 
>>> I suppose you could alternatively make Produced a field of Repartitioned,
>>> but I don't think we do this kind of composition elsewhere in Streams at
>>> the moment
>>> 
>>> On Fri, Jul 19, 2019 at 1:45 PM Levani Kokhreidze 
>>> wrote:
>>> 
 Hi Bill,
 
 Thanks a lot for the feedback.
 Yes, that makes sense. I’ve updated KIP with `Repartitioned#partitioner`
 configuration.
 In the beginning, I wanted to introduce a class for topic level
 configuration and keep topic level and producer level configurations (such
 as Produced) separately (see my second email in this thread).
 But while looking at the semantics of KStream interface, I couldn’t really
 figure out good operation name for Topic level configuration class and just
 introducing `Topic` config class was kinda breaking the semantics.
 So I think having Repartitioned class which encapsulates topic and
 producer level configurations for internal topics is viable thing to do.
 
 Regards,
 Levani
 
> On Jul 19, 2019, at 7:47 PM, Bill Bejeck  wrote:
> 
> Hi Lavani,
> 
> Thanks for resurrecting this KIP.
> 
> I'm also a +1 for adding a partition option.  In addition to the reason
> provided by John, my reasoning is:
> 
> 1. Users may want to use something other than hash-based partitioning
> 2. Users may wish to partition on something different than the key
> without having to change the key.  For example:
>1. A combination of fields in the value in conjunction with the key
>2. Something other than the key
> 3. We allow users to specify a partitioner on Produced hence in
> KStream.to and KStream.through, so it makes sense for API consistency.
> 
> Just my  2 cents.
> 
> Thanks,
> Bill
> 
> 
> 
> On Fri, Jul 19, 2019 at 5:46 AM Levani Kokhreidze <
 levani.co...@gmail.com>
> wrote:
> 
>> Hi John,
>> 
>> In my mind it makes sense.
>> If we add partitioner configuration to Repartitioned class, with the

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Matthias J. Sax]

Thanks for driving the KIP.

I agree that users need to be able to specify a partitioning strategy.

Sophie raises a fair point about topic configs and producer configs. My
take is, that consider `Repartitioned` as an "extension" to `Produced`,
that adds topic configuration, is a good way to think about it and helps
to keep the API "clean".


With regard to method names. I would prefer to avoid abbreviations. Can
we rename:

`withNumOfPartitions` -> `withNumberOfPartitions`

Furthermore, it might be good to add some more `static` methods:

 - Repartitioned.with(Serde, Serde)
 - Repartitioned.withNumberOfPartitions(int)
 - Repartitioned.streamPartitioner(StreamPartitioner)


-Matthias

On 7/19/19 3:33 PM, Levani Kokhreidze wrote:
> Totally agree. I think in KStream interface it makes sense to have some 
> duplicate configurations between operators in order to keep API simple and 
> usable.
> Also, as more surface API has, harder it is to have proper backward 
> compatibility.
> While initial idea of keeping topic level configs separate was exciting, 
> having Repartitioned class encapsulate some producer level configs makes API 
> more readable.
> 
> Regards,
> Levani
> 
>> On Jul 20, 2019, at 1:15 AM, Sophie Blee-Goldman  wrote:
>>
>> I think that is a good point about trying to keep producer level
>> configurations and (repartition) topic level considerations separate.
>> Number of partitions is definitely purely a topic level configuration. But
>> on some level, serdes and partitioners are just as much a topic
>> configuration as a producer one. You could have two producers configured
>> with different serdes and/or partitioners, but if they are writing to the
>> same topic the result would be very difficult to part. So in a sense, these
>> are configurations of topics in Streams, not just producers.
>>
>> Another way to think of it: while the Streams API is not always true to
>> this, ideally all the relevant configs for an operator are wrapped into a
>> single object (in this case, Repartitioned). We could instead split out the
>> fields in common with Produced into a separate parameter to keep topic and
>> producer level configurations separate, but this increases the API surface
>> area by a lot. It's much more straightforward to just say "this is
>> everything that this particular operator needs" without worrying about what
>> exactly you're specifying.
>>
>> I suppose you could alternatively make Produced a field of Repartitioned,
>> but I don't think we do this kind of composition elsewhere in Streams at
>> the moment
>>
>> On Fri, Jul 19, 2019 at 1:45 PM Levani Kokhreidze 
>> wrote:
>>
>>> Hi Bill,
>>>
>>> Thanks a lot for the feedback.
>>> Yes, that makes sense. I’ve updated KIP with `Repartitioned#partitioner`
>>> configuration.
>>> In the beginning, I wanted to introduce a class for topic level
>>> configuration and keep topic level and producer level configurations (such
>>> as Produced) separately (see my second email in this thread).
>>> But while looking at the semantics of KStream interface, I couldn’t really
>>> figure out good operation name for Topic level configuration class and just
>>> introducing `Topic` config class was kinda breaking the semantics.
>>> So I think having Repartitioned class which encapsulates topic and
>>> producer level configurations for internal topics is viable thing to do.
>>>
>>> Regards,
>>> Levani
>>>
 On Jul 19, 2019, at 7:47 PM, Bill Bejeck  wrote:

 Hi Lavani,

 Thanks for resurrecting this KIP.

 I'm also a +1 for adding a partition option.  In addition to the reason
 provided by John, my reasoning is:

  1. Users may want to use something other than hash-based partitioning
  2. Users may wish to partition on something different than the key
  without having to change the key.  For example:
 1. A combination of fields in the value in conjunction with the key
 2. Something other than the key
  3. We allow users to specify a partitioner on Produced hence in
  KStream.to and KStream.through, so it makes sense for API consistency.

 Just my  2 cents.

 Thanks,
 Bill



 On Fri, Jul 19, 2019 at 5:46 AM Levani Kokhreidze <
>>> levani.co...@gmail.com>
 wrote:

> Hi John,
>
> In my mind it makes sense.
> If we add partitioner configuration to Repartitioned class, with the
> combination of specifying number of partitions for internal topics, user
> will have opportunity to ensure co-partitioning before join operation.
> I think this can be quite powerful feature.
> Wondering what others think about this?
>
> Regards,
> Levani
>
>> On Jul 18, 2019, at 1:20 AM, John Roesler  wrote:
>>
>> Yes, I believe that's what I had in mind. Again, not totally sure it
>> makes sense, but I believe something similar is the rationale for
>> having the partitioner option in Produced.
>>
>> 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Levani Kokhreidze]

Totally agree. I think in KStream interface it makes sense to have some 
duplicate configurations between operators in order to keep API simple and 
usable.
Also, as more surface API has, harder it is to have proper backward 
compatibility.
While initial idea of keeping topic level configs separate was exciting, having 
Repartitioned class encapsulate some producer level configs makes API more 
readable.

Regards,
Levani

> On Jul 20, 2019, at 1:15 AM, Sophie Blee-Goldman  wrote:
> 
> I think that is a good point about trying to keep producer level
> configurations and (repartition) topic level considerations separate.
> Number of partitions is definitely purely a topic level configuration. But
> on some level, serdes and partitioners are just as much a topic
> configuration as a producer one. You could have two producers configured
> with different serdes and/or partitioners, but if they are writing to the
> same topic the result would be very difficult to part. So in a sense, these
> are configurations of topics in Streams, not just producers.
> 
> Another way to think of it: while the Streams API is not always true to
> this, ideally all the relevant configs for an operator are wrapped into a
> single object (in this case, Repartitioned). We could instead split out the
> fields in common with Produced into a separate parameter to keep topic and
> producer level configurations separate, but this increases the API surface
> area by a lot. It's much more straightforward to just say "this is
> everything that this particular operator needs" without worrying about what
> exactly you're specifying.
> 
> I suppose you could alternatively make Produced a field of Repartitioned,
> but I don't think we do this kind of composition elsewhere in Streams at
> the moment
> 
> On Fri, Jul 19, 2019 at 1:45 PM Levani Kokhreidze 
> wrote:
> 
>> Hi Bill,
>> 
>> Thanks a lot for the feedback.
>> Yes, that makes sense. I’ve updated KIP with `Repartitioned#partitioner`
>> configuration.
>> In the beginning, I wanted to introduce a class for topic level
>> configuration and keep topic level and producer level configurations (such
>> as Produced) separately (see my second email in this thread).
>> But while looking at the semantics of KStream interface, I couldn’t really
>> figure out good operation name for Topic level configuration class and just
>> introducing `Topic` config class was kinda breaking the semantics.
>> So I think having Repartitioned class which encapsulates topic and
>> producer level configurations for internal topics is viable thing to do.
>> 
>> Regards,
>> Levani
>> 
>>> On Jul 19, 2019, at 7:47 PM, Bill Bejeck  wrote:
>>> 
>>> Hi Lavani,
>>> 
>>> Thanks for resurrecting this KIP.
>>> 
>>> I'm also a +1 for adding a partition option.  In addition to the reason
>>> provided by John, my reasoning is:
>>> 
>>>  1. Users may want to use something other than hash-based partitioning
>>>  2. Users may wish to partition on something different than the key
>>>  without having to change the key.  For example:
>>> 1. A combination of fields in the value in conjunction with the key
>>> 2. Something other than the key
>>>  3. We allow users to specify a partitioner on Produced hence in
>>>  KStream.to and KStream.through, so it makes sense for API consistency.
>>> 
>>> Just my  2 cents.
>>> 
>>> Thanks,
>>> Bill
>>> 
>>> 
>>> 
>>> On Fri, Jul 19, 2019 at 5:46 AM Levani Kokhreidze <
>> levani.co...@gmail.com>
>>> wrote:
>>> 
 Hi John,
 
 In my mind it makes sense.
 If we add partitioner configuration to Repartitioned class, with the
 combination of specifying number of partitions for internal topics, user
 will have opportunity to ensure co-partitioning before join operation.
 I think this can be quite powerful feature.
 Wondering what others think about this?
 
 Regards,
 Levani
 
> On Jul 18, 2019, at 1:20 AM, John Roesler  wrote:
> 
> Yes, I believe that's what I had in mind. Again, not totally sure it
> makes sense, but I believe something similar is the rationale for
> having the partitioner option in Produced.
> 
> Thanks,
> -John
> 
> On Wed, Jul 17, 2019 at 3:20 PM Levani Kokhreidze
>  wrote:
>> 
>> Hey John,
>> 
>> Oh that’s interesting use-case.
>> Do I understand this correctly, in your example I would first issue
 repartition(Repartitioned) with proper partitioner that essentially
>> would
 be the same as the topic I want to join with and then do the
>> KStream#join
 with DSL?
>> 
>> Regards,
>> Levani
>> 
>>> On Jul 17, 2019, at 11:11 PM, John Roesler 
>> wrote:
>>> 
>>> Hey, all, just to chime in,
>>> 
>>> I think it might be useful to have an option to specify the
>>> partitioner. The case I have in mind is that some data may get
>>> repartitioned and then joined with an input topic. If the right-side
>>> input topic uses a 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Sophie Blee-Goldman]

I think that is a good point about trying to keep producer level
configurations and (repartition) topic level considerations separate.
Number of partitions is definitely purely a topic level configuration. But
on some level, serdes and partitioners are just as much a topic
configuration as a producer one. You could have two producers configured
with different serdes and/or partitioners, but if they are writing to the
same topic the result would be very difficult to part. So in a sense, these
are configurations of topics in Streams, not just producers.

Another way to think of it: while the Streams API is not always true to
this, ideally all the relevant configs for an operator are wrapped into a
single object (in this case, Repartitioned). We could instead split out the
fields in common with Produced into a separate parameter to keep topic and
producer level configurations separate, but this increases the API surface
area by a lot. It's much more straightforward to just say "this is
everything that this particular operator needs" without worrying about what
exactly you're specifying.

I suppose you could alternatively make Produced a field of Repartitioned,
but I don't think we do this kind of composition elsewhere in Streams at
the moment

On Fri, Jul 19, 2019 at 1:45 PM Levani Kokhreidze 
wrote:

> Hi Bill,
>
> Thanks a lot for the feedback.
> Yes, that makes sense. I’ve updated KIP with `Repartitioned#partitioner`
> configuration.
> In the beginning, I wanted to introduce a class for topic level
> configuration and keep topic level and producer level configurations (such
> as Produced) separately (see my second email in this thread).
> But while looking at the semantics of KStream interface, I couldn’t really
> figure out good operation name for Topic level configuration class and just
> introducing `Topic` config class was kinda breaking the semantics.
> So I think having Repartitioned class which encapsulates topic and
> producer level configurations for internal topics is viable thing to do.
>
> Regards,
> Levani
>
> > On Jul 19, 2019, at 7:47 PM, Bill Bejeck  wrote:
> >
> > Hi Lavani,
> >
> > Thanks for resurrecting this KIP.
> >
> > I'm also a +1 for adding a partition option.  In addition to the reason
> > provided by John, my reasoning is:
> >
> >   1. Users may want to use something other than hash-based partitioning
> >   2. Users may wish to partition on something different than the key
> >   without having to change the key.  For example:
> >  1. A combination of fields in the value in conjunction with the key
> >  2. Something other than the key
> >   3. We allow users to specify a partitioner on Produced hence in
> >   KStream.to and KStream.through, so it makes sense for API consistency.
> >
> > Just my  2 cents.
> >
> > Thanks,
> > Bill
> >
> >
> >
> > On Fri, Jul 19, 2019 at 5:46 AM Levani Kokhreidze <
> levani.co...@gmail.com>
> > wrote:
> >
> >> Hi John,
> >>
> >> In my mind it makes sense.
> >> If we add partitioner configuration to Repartitioned class, with the
> >> combination of specifying number of partitions for internal topics, user
> >> will have opportunity to ensure co-partitioning before join operation.
> >> I think this can be quite powerful feature.
> >> Wondering what others think about this?
> >>
> >> Regards,
> >> Levani
> >>
> >>> On Jul 18, 2019, at 1:20 AM, John Roesler  wrote:
> >>>
> >>> Yes, I believe that's what I had in mind. Again, not totally sure it
> >>> makes sense, but I believe something similar is the rationale for
> >>> having the partitioner option in Produced.
> >>>
> >>> Thanks,
> >>> -John
> >>>
> >>> On Wed, Jul 17, 2019 at 3:20 PM Levani Kokhreidze
> >>>  wrote:
> 
>  Hey John,
> 
>  Oh that’s interesting use-case.
>  Do I understand this correctly, in your example I would first issue
> >> repartition(Repartitioned) with proper partitioner that essentially
> would
> >> be the same as the topic I want to join with and then do the
> KStream#join
> >> with DSL?
> 
>  Regards,
>  Levani
> 
> > On Jul 17, 2019, at 11:11 PM, John Roesler 
> wrote:
> >
> > Hey, all, just to chime in,
> >
> > I think it might be useful to have an option to specify the
> > partitioner. The case I have in mind is that some data may get
> > repartitioned and then joined with an input topic. If the right-side
> > input topic uses a custom partitioning strategy, then the
> > repartitioned stream also needs to be partitioned with the same
> > strategy.
> >
> > Does that make sense, or did I maybe miss something important?
> >
> > Thanks,
> > -John
> >
> > On Wed, Jul 17, 2019 at 2:48 PM Levani Kokhreidze
> >  wrote:
> >>
> >> Yes, I was thinking about it as well. To be honest I’m not sure
> about
> >> it yet.
> >> As Kafka Streams DSL user, I don’t really think I would need control
> >> over partitioner for internal topics.
> >> As a user, I would 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Levani Kokhreidze]

Hi Bill,

Thanks a lot for the feedback.
Yes, that makes sense. I’ve updated KIP with `Repartitioned#partitioner` 
configuration.
In the beginning, I wanted to introduce a class for topic level configuration 
and keep topic level and producer level configurations (such as Produced) 
separately (see my second email in this thread).
But while looking at the semantics of KStream interface, I couldn’t really 
figure out good operation name for Topic level configuration class and just 
introducing `Topic` config class was kinda breaking the semantics.
So I think having Repartitioned class which encapsulates topic and producer 
level configurations for internal topics is viable thing to do.

Regards,
Levani

> On Jul 19, 2019, at 7:47 PM, Bill Bejeck  wrote:
> 
> Hi Lavani,
> 
> Thanks for resurrecting this KIP.
> 
> I'm also a +1 for adding a partition option.  In addition to the reason
> provided by John, my reasoning is:
> 
>   1. Users may want to use something other than hash-based partitioning
>   2. Users may wish to partition on something different than the key
>   without having to change the key.  For example:
>  1. A combination of fields in the value in conjunction with the key
>  2. Something other than the key
>   3. We allow users to specify a partitioner on Produced hence in
>   KStream.to and KStream.through, so it makes sense for API consistency.
> 
> Just my  2 cents.
> 
> Thanks,
> Bill
> 
> 
> 
> On Fri, Jul 19, 2019 at 5:46 AM Levani Kokhreidze 
> wrote:
> 
>> Hi John,
>> 
>> In my mind it makes sense.
>> If we add partitioner configuration to Repartitioned class, with the
>> combination of specifying number of partitions for internal topics, user
>> will have opportunity to ensure co-partitioning before join operation.
>> I think this can be quite powerful feature.
>> Wondering what others think about this?
>> 
>> Regards,
>> Levani
>> 
>>> On Jul 18, 2019, at 1:20 AM, John Roesler  wrote:
>>> 
>>> Yes, I believe that's what I had in mind. Again, not totally sure it
>>> makes sense, but I believe something similar is the rationale for
>>> having the partitioner option in Produced.
>>> 
>>> Thanks,
>>> -John
>>> 
>>> On Wed, Jul 17, 2019 at 3:20 PM Levani Kokhreidze
>>>  wrote:
 
 Hey John,
 
 Oh that’s interesting use-case.
 Do I understand this correctly, in your example I would first issue
>> repartition(Repartitioned) with proper partitioner that essentially would
>> be the same as the topic I want to join with and then do the KStream#join
>> with DSL?
 
 Regards,
 Levani
 
> On Jul 17, 2019, at 11:11 PM, John Roesler  wrote:
> 
> Hey, all, just to chime in,
> 
> I think it might be useful to have an option to specify the
> partitioner. The case I have in mind is that some data may get
> repartitioned and then joined with an input topic. If the right-side
> input topic uses a custom partitioning strategy, then the
> repartitioned stream also needs to be partitioned with the same
> strategy.
> 
> Does that make sense, or did I maybe miss something important?
> 
> Thanks,
> -John
> 
> On Wed, Jul 17, 2019 at 2:48 PM Levani Kokhreidze
>  wrote:
>> 
>> Yes, I was thinking about it as well. To be honest I’m not sure about
>> it yet.
>> As Kafka Streams DSL user, I don’t really think I would need control
>> over partitioner for internal topics.
>> As a user, I would assume that Kafka Streams knows best how to
>> partition data for internal topics.
>> In this KIP I wrote that Produced should be used only for topics that
>> are created by user In advance.
>> In those cases maybe it make sense to have possibility to specify the
>> partitioner.
>> I don’t have clear answer on that yet, but I guess specifying the
>> partitioner can be added as well if there’s agreement on this.
>> 
>> Regards,
>> Levani
>> 
>>> On Jul 17, 2019, at 10:42 PM, Sophie Blee-Goldman <
>> sop...@confluent.io> wrote:
>>> 
>>> Thanks for clearing that up. I agree that Repartitioned would be a
>> useful
>>> addition. I'm wondering if it might also need to have
>>> a withStreamPartitioner method/field, similar to Produced? I'm not
>> sure how
>>> widely this feature is really used, but seems it should be available
>> for
>>> repartition topics.
>>> 
>>> On Wed, Jul 17, 2019 at 11:26 AM Levani Kokhreidze <
>> levani.co...@gmail.com>
>>> wrote:
>>> 
 Hey Sophie,
 
 In both cases KStream#repartition and
>> KStream#repartition(Repartitioned)
 topic will be created and managed by Kafka Streams.
 Idea of Repartitioned is to give user more control over the topic
>> such as
 num of partitions.
 I feel like Repartitioned parameter is something that is missing in
 current DSL design.
 Essentially giving user control over parallelism by configuring num
>> of

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Bill Bejeck]

Hi Lavani,

Thanks for resurrecting this KIP.

I'm also a +1 for adding a partition option.  In addition to the reason
provided by John, my reasoning is:

   1. Users may want to use something other than hash-based partitioning
   2. Users may wish to partition on something different than the key
   without having to change the key.  For example:
  1. A combination of fields in the value in conjunction with the key
  2. Something other than the key
   3. We allow users to specify a partitioner on Produced hence in
   KStream.to and KStream.through, so it makes sense for API consistency.

Just my  2 cents.

Thanks,
Bill



On Fri, Jul 19, 2019 at 5:46 AM Levani Kokhreidze 
wrote:

> Hi John,
>
> In my mind it makes sense.
> If we add partitioner configuration to Repartitioned class, with the
> combination of specifying number of partitions for internal topics, user
> will have opportunity to ensure co-partitioning before join operation.
> I think this can be quite powerful feature.
> Wondering what others think about this?
>
> Regards,
> Levani
>
> > On Jul 18, 2019, at 1:20 AM, John Roesler  wrote:
> >
> > Yes, I believe that's what I had in mind. Again, not totally sure it
> > makes sense, but I believe something similar is the rationale for
> > having the partitioner option in Produced.
> >
> > Thanks,
> > -John
> >
> > On Wed, Jul 17, 2019 at 3:20 PM Levani Kokhreidze
> >  wrote:
> >>
> >> Hey John,
> >>
> >> Oh that’s interesting use-case.
> >> Do I understand this correctly, in your example I would first issue
> repartition(Repartitioned) with proper partitioner that essentially would
> be the same as the topic I want to join with and then do the KStream#join
> with DSL?
> >>
> >> Regards,
> >> Levani
> >>
> >>> On Jul 17, 2019, at 11:11 PM, John Roesler  wrote:
> >>>
> >>> Hey, all, just to chime in,
> >>>
> >>> I think it might be useful to have an option to specify the
> >>> partitioner. The case I have in mind is that some data may get
> >>> repartitioned and then joined with an input topic. If the right-side
> >>> input topic uses a custom partitioning strategy, then the
> >>> repartitioned stream also needs to be partitioned with the same
> >>> strategy.
> >>>
> >>> Does that make sense, or did I maybe miss something important?
> >>>
> >>> Thanks,
> >>> -John
> >>>
> >>> On Wed, Jul 17, 2019 at 2:48 PM Levani Kokhreidze
> >>>  wrote:
> 
>  Yes, I was thinking about it as well. To be honest I’m not sure about
> it yet.
>  As Kafka Streams DSL user, I don’t really think I would need control
> over partitioner for internal topics.
>  As a user, I would assume that Kafka Streams knows best how to
> partition data for internal topics.
>  In this KIP I wrote that Produced should be used only for topics that
> are created by user In advance.
>  In those cases maybe it make sense to have possibility to specify the
> partitioner.
>  I don’t have clear answer on that yet, but I guess specifying the
> partitioner can be added as well if there’s agreement on this.
> 
>  Regards,
>  Levani
> 
> > On Jul 17, 2019, at 10:42 PM, Sophie Blee-Goldman <
> sop...@confluent.io> wrote:
> >
> > Thanks for clearing that up. I agree that Repartitioned would be a
> useful
> > addition. I'm wondering if it might also need to have
> > a withStreamPartitioner method/field, similar to Produced? I'm not
> sure how
> > widely this feature is really used, but seems it should be available
> for
> > repartition topics.
> >
> > On Wed, Jul 17, 2019 at 11:26 AM Levani Kokhreidze <
> levani.co...@gmail.com>
> > wrote:
> >
> >> Hey Sophie,
> >>
> >> In both cases KStream#repartition and
> KStream#repartition(Repartitioned)
> >> topic will be created and managed by Kafka Streams.
> >> Idea of Repartitioned is to give user more control over the topic
> such as
> >> num of partitions.
> >> I feel like Repartitioned parameter is something that is missing in
> >> current DSL design.
> >> Essentially giving user control over parallelism by configuring num
> of
> >> partitions for internal topics.
> >>
> >> Hope this answers your question.
> >>
> >> Regards,
> >> Levani
> >>
> >>> On Jul 17, 2019, at 9:02 PM, Sophie Blee-Goldman <
> sop...@confluent.io>
> >> wrote:
> >>>
> >>> Hey Levani,
> >>>
> >>> Thanks for the KIP! Can you clarify one thing for me -- for the
> >>> KStream#repartition signature taking a Repartitioned, will the
> topic be
> >>> auto-created by Streams (which seems to be the case for the
> signature
> >>> without a Repartitioned) or does it have to be pre-created? The
> wording
> >> in
> >>> the KIP makes it seem like one version of the method will
> auto-create
> >>> topics while the other will not.
> >>>
> >>> Cheers,
> >>> Sophie
> >>>
> >>> On Wed, Jul 17, 2019 at 10:15 AM Levani Kokhreidze <

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Levani Kokhreidze]

Hi John,

In my mind it makes sense. 
If we add partitioner configuration to Repartitioned class, with the 
combination of specifying number of partitions for internal topics, user will 
have opportunity to ensure co-partitioning before join operation. 
I think this can be quite powerful feature.
Wondering what others think about this?

Regards,
Levani

> On Jul 18, 2019, at 1:20 AM, John Roesler  wrote:
> 
> Yes, I believe that's what I had in mind. Again, not totally sure it
> makes sense, but I believe something similar is the rationale for
> having the partitioner option in Produced.
> 
> Thanks,
> -John
> 
> On Wed, Jul 17, 2019 at 3:20 PM Levani Kokhreidze
>  wrote:
>> 
>> Hey John,
>> 
>> Oh that’s interesting use-case.
>> Do I understand this correctly, in your example I would first issue 
>> repartition(Repartitioned) with proper partitioner that essentially would be 
>> the same as the topic I want to join with and then do the KStream#join with 
>> DSL?
>> 
>> Regards,
>> Levani
>> 
>>> On Jul 17, 2019, at 11:11 PM, John Roesler  wrote:
>>> 
>>> Hey, all, just to chime in,
>>> 
>>> I think it might be useful to have an option to specify the
>>> partitioner. The case I have in mind is that some data may get
>>> repartitioned and then joined with an input topic. If the right-side
>>> input topic uses a custom partitioning strategy, then the
>>> repartitioned stream also needs to be partitioned with the same
>>> strategy.
>>> 
>>> Does that make sense, or did I maybe miss something important?
>>> 
>>> Thanks,
>>> -John
>>> 
>>> On Wed, Jul 17, 2019 at 2:48 PM Levani Kokhreidze
>>>  wrote:
 
 Yes, I was thinking about it as well. To be honest I’m not sure about it 
 yet.
 As Kafka Streams DSL user, I don’t really think I would need control over 
 partitioner for internal topics.
 As a user, I would assume that Kafka Streams knows best how to partition 
 data for internal topics.
 In this KIP I wrote that Produced should be used only for topics that are 
 created by user In advance.
 In those cases maybe it make sense to have possibility to specify the 
 partitioner.
 I don’t have clear answer on that yet, but I guess specifying the 
 partitioner can be added as well if there’s agreement on this.
 
 Regards,
 Levani
 
> On Jul 17, 2019, at 10:42 PM, Sophie Blee-Goldman  
> wrote:
> 
> Thanks for clearing that up. I agree that Repartitioned would be a useful
> addition. I'm wondering if it might also need to have
> a withStreamPartitioner method/field, similar to Produced? I'm not sure 
> how
> widely this feature is really used, but seems it should be available for
> repartition topics.
> 
> On Wed, Jul 17, 2019 at 11:26 AM Levani Kokhreidze 
> 
> wrote:
> 
>> Hey Sophie,
>> 
>> In both cases KStream#repartition and KStream#repartition(Repartitioned)
>> topic will be created and managed by Kafka Streams.
>> Idea of Repartitioned is to give user more control over the topic such as
>> num of partitions.
>> I feel like Repartitioned parameter is something that is missing in
>> current DSL design.
>> Essentially giving user control over parallelism by configuring num of
>> partitions for internal topics.
>> 
>> Hope this answers your question.
>> 
>> Regards,
>> Levani
>> 
>>> On Jul 17, 2019, at 9:02 PM, Sophie Blee-Goldman 
>> wrote:
>>> 
>>> Hey Levani,
>>> 
>>> Thanks for the KIP! Can you clarify one thing for me -- for the
>>> KStream#repartition signature taking a Repartitioned, will the topic be
>>> auto-created by Streams (which seems to be the case for the signature
>>> without a Repartitioned) or does it have to be pre-created? The wording
>> in
>>> the KIP makes it seem like one version of the method will auto-create
>>> topics while the other will not.
>>> 
>>> Cheers,
>>> Sophie
>>> 
>>> On Wed, Jul 17, 2019 at 10:15 AM Levani Kokhreidze <
>> levani.co...@gmail.com>
>>> wrote:
>>> 
 Hello,
 
 One more bump about KIP-221 (
 
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
 <
 
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
>>> )
 so it doesn’t get lost in mailing list :)
 Would love to hear communities opinions/concerns about this KIP.
 
 Regards,
 Levani
 
 
> On Jul 12, 2019, at 5:27 PM, Levani Kokhreidze >> 
 wrote:
> 
> Hello,
> 
> Kind reminder about this KIP:
 
>> 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[John Roesler]

Yes, I believe that's what I had in mind. Again, not totally sure it
makes sense, but I believe something similar is the rationale for
having the partitioner option in Produced.

Thanks,
-John

On Wed, Jul 17, 2019 at 3:20 PM Levani Kokhreidze
 wrote:
>
> Hey John,
>
> Oh that’s interesting use-case.
> Do I understand this correctly, in your example I would first issue 
> repartition(Repartitioned) with proper partitioner that essentially would be 
> the same as the topic I want to join with and then do the KStream#join with 
> DSL?
>
> Regards,
> Levani
>
> > On Jul 17, 2019, at 11:11 PM, John Roesler  wrote:
> >
> > Hey, all, just to chime in,
> >
> > I think it might be useful to have an option to specify the
> > partitioner. The case I have in mind is that some data may get
> > repartitioned and then joined with an input topic. If the right-side
> > input topic uses a custom partitioning strategy, then the
> > repartitioned stream also needs to be partitioned with the same
> > strategy.
> >
> > Does that make sense, or did I maybe miss something important?
> >
> > Thanks,
> > -John
> >
> > On Wed, Jul 17, 2019 at 2:48 PM Levani Kokhreidze
> >  wrote:
> >>
> >> Yes, I was thinking about it as well. To be honest I’m not sure about it 
> >> yet.
> >> As Kafka Streams DSL user, I don’t really think I would need control over 
> >> partitioner for internal topics.
> >> As a user, I would assume that Kafka Streams knows best how to partition 
> >> data for internal topics.
> >> In this KIP I wrote that Produced should be used only for topics that are 
> >> created by user In advance.
> >> In those cases maybe it make sense to have possibility to specify the 
> >> partitioner.
> >> I don’t have clear answer on that yet, but I guess specifying the 
> >> partitioner can be added as well if there’s agreement on this.
> >>
> >> Regards,
> >> Levani
> >>
> >>> On Jul 17, 2019, at 10:42 PM, Sophie Blee-Goldman  
> >>> wrote:
> >>>
> >>> Thanks for clearing that up. I agree that Repartitioned would be a useful
> >>> addition. I'm wondering if it might also need to have
> >>> a withStreamPartitioner method/field, similar to Produced? I'm not sure 
> >>> how
> >>> widely this feature is really used, but seems it should be available for
> >>> repartition topics.
> >>>
> >>> On Wed, Jul 17, 2019 at 11:26 AM Levani Kokhreidze 
> >>> 
> >>> wrote:
> >>>
>  Hey Sophie,
> 
>  In both cases KStream#repartition and KStream#repartition(Repartitioned)
>  topic will be created and managed by Kafka Streams.
>  Idea of Repartitioned is to give user more control over the topic such as
>  num of partitions.
>  I feel like Repartitioned parameter is something that is missing in
>  current DSL design.
>  Essentially giving user control over parallelism by configuring num of
>  partitions for internal topics.
> 
>  Hope this answers your question.
> 
>  Regards,
>  Levani
> 
> > On Jul 17, 2019, at 9:02 PM, Sophie Blee-Goldman 
>  wrote:
> >
> > Hey Levani,
> >
> > Thanks for the KIP! Can you clarify one thing for me -- for the
> > KStream#repartition signature taking a Repartitioned, will the topic be
> > auto-created by Streams (which seems to be the case for the signature
> > without a Repartitioned) or does it have to be pre-created? The wording
>  in
> > the KIP makes it seem like one version of the method will auto-create
> > topics while the other will not.
> >
> > Cheers,
> > Sophie
> >
> > On Wed, Jul 17, 2019 at 10:15 AM Levani Kokhreidze <
>  levani.co...@gmail.com>
> > wrote:
> >
> >> Hello,
> >>
> >> One more bump about KIP-221 (
> >>
>  https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> >> <
> >>
>  https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> > )
> >> so it doesn’t get lost in mailing list :)
> >> Would love to hear communities opinions/concerns about this KIP.
> >>
> >> Regards,
> >> Levani
> >>
> >>
> >>> On Jul 12, 2019, at 5:27 PM, Levani Kokhreidze  >
> >> wrote:
> >>>
> >>> Hello,
> >>>
> >>> Kind reminder about this KIP:
> >>
>  https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> >> <
> >>
>  https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> >>>
> >>>
> >>> Regards,
> >>> Levani
> >>>
>  On Jul 9, 2019, at 11:38 AM, Levani Kokhreidze <
>  levani.co...@gmail.com
> >> > wrote:
> 
>  Hello,
> 
>  In order to move this KIP forward, I’ve updated 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Levani Kokhreidze]

Hey John,

Oh that’s interesting use-case. 
Do I understand this correctly, in your example I would first issue 
repartition(Repartitioned) with proper partitioner that essentially would be 
the same as the topic I want to join with and then do the KStream#join with DSL?

Regards,
Levani

> On Jul 17, 2019, at 11:11 PM, John Roesler  wrote:
> 
> Hey, all, just to chime in,
> 
> I think it might be useful to have an option to specify the
> partitioner. The case I have in mind is that some data may get
> repartitioned and then joined with an input topic. If the right-side
> input topic uses a custom partitioning strategy, then the
> repartitioned stream also needs to be partitioned with the same
> strategy.
> 
> Does that make sense, or did I maybe miss something important?
> 
> Thanks,
> -John
> 
> On Wed, Jul 17, 2019 at 2:48 PM Levani Kokhreidze
>  wrote:
>> 
>> Yes, I was thinking about it as well. To be honest I’m not sure about it yet.
>> As Kafka Streams DSL user, I don’t really think I would need control over 
>> partitioner for internal topics.
>> As a user, I would assume that Kafka Streams knows best how to partition 
>> data for internal topics.
>> In this KIP I wrote that Produced should be used only for topics that are 
>> created by user In advance.
>> In those cases maybe it make sense to have possibility to specify the 
>> partitioner.
>> I don’t have clear answer on that yet, but I guess specifying the 
>> partitioner can be added as well if there’s agreement on this.
>> 
>> Regards,
>> Levani
>> 
>>> On Jul 17, 2019, at 10:42 PM, Sophie Blee-Goldman  
>>> wrote:
>>> 
>>> Thanks for clearing that up. I agree that Repartitioned would be a useful
>>> addition. I'm wondering if it might also need to have
>>> a withStreamPartitioner method/field, similar to Produced? I'm not sure how
>>> widely this feature is really used, but seems it should be available for
>>> repartition topics.
>>> 
>>> On Wed, Jul 17, 2019 at 11:26 AM Levani Kokhreidze 
>>> wrote:
>>> 
 Hey Sophie,
 
 In both cases KStream#repartition and KStream#repartition(Repartitioned)
 topic will be created and managed by Kafka Streams.
 Idea of Repartitioned is to give user more control over the topic such as
 num of partitions.
 I feel like Repartitioned parameter is something that is missing in
 current DSL design.
 Essentially giving user control over parallelism by configuring num of
 partitions for internal topics.
 
 Hope this answers your question.
 
 Regards,
 Levani
 
> On Jul 17, 2019, at 9:02 PM, Sophie Blee-Goldman 
 wrote:
> 
> Hey Levani,
> 
> Thanks for the KIP! Can you clarify one thing for me -- for the
> KStream#repartition signature taking a Repartitioned, will the topic be
> auto-created by Streams (which seems to be the case for the signature
> without a Repartitioned) or does it have to be pre-created? The wording
 in
> the KIP makes it seem like one version of the method will auto-create
> topics while the other will not.
> 
> Cheers,
> Sophie
> 
> On Wed, Jul 17, 2019 at 10:15 AM Levani Kokhreidze <
 levani.co...@gmail.com>
> wrote:
> 
>> Hello,
>> 
>> One more bump about KIP-221 (
>> 
 https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
>> <
>> 
 https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> )
>> so it doesn’t get lost in mailing list :)
>> Would love to hear communities opinions/concerns about this KIP.
>> 
>> Regards,
>> Levani
>> 
>> 
>>> On Jul 12, 2019, at 5:27 PM, Levani Kokhreidze  
>> wrote:
>>> 
>>> Hello,
>>> 
>>> Kind reminder about this KIP:
>> 
 https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
>> <
>> 
 https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
>>> 
>>> 
>>> Regards,
>>> Levani
>>> 
 On Jul 9, 2019, at 11:38 AM, Levani Kokhreidze <
 levani.co...@gmail.com
>> > wrote:
 
 Hello,
 
 In order to move this KIP forward, I’ve updated confluence page with
>> the new proposal
>> 
 https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
>> <
>> 
 https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
>>> 
 I’ve also filled “Rejected Alternatives” section.
 
 Looking forward to discuss this KIP :)
 
 King regards,
 Levani

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[John Roesler]

Hey, all, just to chime in,

I think it might be useful to have an option to specify the
partitioner. The case I have in mind is that some data may get
repartitioned and then joined with an input topic. If the right-side
input topic uses a custom partitioning strategy, then the
repartitioned stream also needs to be partitioned with the same
strategy.

Does that make sense, or did I maybe miss something important?

Thanks,
-John

On Wed, Jul 17, 2019 at 2:48 PM Levani Kokhreidze
 wrote:
>
> Yes, I was thinking about it as well. To be honest I’m not sure about it yet.
> As Kafka Streams DSL user, I don’t really think I would need control over 
> partitioner for internal topics.
> As a user, I would assume that Kafka Streams knows best how to partition data 
> for internal topics.
> In this KIP I wrote that Produced should be used only for topics that are 
> created by user In advance.
> In those cases maybe it make sense to have possibility to specify the 
> partitioner.
> I don’t have clear answer on that yet, but I guess specifying the partitioner 
> can be added as well if there’s agreement on this.
>
> Regards,
> Levani
>
> > On Jul 17, 2019, at 10:42 PM, Sophie Blee-Goldman  
> > wrote:
> >
> > Thanks for clearing that up. I agree that Repartitioned would be a useful
> > addition. I'm wondering if it might also need to have
> > a withStreamPartitioner method/field, similar to Produced? I'm not sure how
> > widely this feature is really used, but seems it should be available for
> > repartition topics.
> >
> > On Wed, Jul 17, 2019 at 11:26 AM Levani Kokhreidze 
> > wrote:
> >
> >> Hey Sophie,
> >>
> >> In both cases KStream#repartition and KStream#repartition(Repartitioned)
> >> topic will be created and managed by Kafka Streams.
> >> Idea of Repartitioned is to give user more control over the topic such as
> >> num of partitions.
> >> I feel like Repartitioned parameter is something that is missing in
> >> current DSL design.
> >> Essentially giving user control over parallelism by configuring num of
> >> partitions for internal topics.
> >>
> >> Hope this answers your question.
> >>
> >> Regards,
> >> Levani
> >>
> >>> On Jul 17, 2019, at 9:02 PM, Sophie Blee-Goldman 
> >> wrote:
> >>>
> >>> Hey Levani,
> >>>
> >>> Thanks for the KIP! Can you clarify one thing for me -- for the
> >>> KStream#repartition signature taking a Repartitioned, will the topic be
> >>> auto-created by Streams (which seems to be the case for the signature
> >>> without a Repartitioned) or does it have to be pre-created? The wording
> >> in
> >>> the KIP makes it seem like one version of the method will auto-create
> >>> topics while the other will not.
> >>>
> >>> Cheers,
> >>> Sophie
> >>>
> >>> On Wed, Jul 17, 2019 at 10:15 AM Levani Kokhreidze <
> >> levani.co...@gmail.com>
> >>> wrote:
> >>>
>  Hello,
> 
>  One more bump about KIP-221 (
> 
> >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
>  <
> 
> >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> >>> )
>  so it doesn’t get lost in mailing list :)
>  Would love to hear communities opinions/concerns about this KIP.
> 
>  Regards,
>  Levani
> 
> 
> > On Jul 12, 2019, at 5:27 PM, Levani Kokhreidze  >>>
>  wrote:
> >
> > Hello,
> >
> > Kind reminder about this KIP:
> 
> >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
>  <
> 
> >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> >
> >
> > Regards,
> > Levani
> >
> >> On Jul 9, 2019, at 11:38 AM, Levani Kokhreidze <
> >> levani.co...@gmail.com
>  > wrote:
> >>
> >> Hello,
> >>
> >> In order to move this KIP forward, I’ve updated confluence page with
>  the new proposal
> 
> >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
>  <
> 
> >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> >
> >> I’ve also filled “Rejected Alternatives” section.
> >>
> >> Looking forward to discuss this KIP :)
> >>
> >> King regards,
> >> Levani
> >>
> >>
> >>> On Jul 3, 2019, at 1:08 PM, Levani Kokhreidze <
> >> levani.co...@gmail.com
>  > wrote:
> >>>
> >>> Hello Matthias,
> >>>
> >>> Thanks for the feedback and ideas.
> >>> I like the idea of introducing dedicated `Topic` class for topic
>  configuration for internal operators like `groupedBy`.
> >>> Would be great to hear others opinion 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Levani Kokhreidze]

Yes, I was thinking about it as well. To be honest I’m not sure about it yet.
As Kafka Streams DSL user, I don’t really think I would need control over 
partitioner for internal topics.
As a user, I would assume that Kafka Streams knows best how to partition data 
for internal topics.
In this KIP I wrote that Produced should be used only for topics that are 
created by user In advance. 
In those cases maybe it make sense to have possibility to specify the 
partitioner.
I don’t have clear answer on that yet, but I guess specifying the partitioner 
can be added as well if there’s agreement on this.

Regards,
Levani

> On Jul 17, 2019, at 10:42 PM, Sophie Blee-Goldman  wrote:
> 
> Thanks for clearing that up. I agree that Repartitioned would be a useful
> addition. I'm wondering if it might also need to have
> a withStreamPartitioner method/field, similar to Produced? I'm not sure how
> widely this feature is really used, but seems it should be available for
> repartition topics.
> 
> On Wed, Jul 17, 2019 at 11:26 AM Levani Kokhreidze 
> wrote:
> 
>> Hey Sophie,
>> 
>> In both cases KStream#repartition and KStream#repartition(Repartitioned)
>> topic will be created and managed by Kafka Streams.
>> Idea of Repartitioned is to give user more control over the topic such as
>> num of partitions.
>> I feel like Repartitioned parameter is something that is missing in
>> current DSL design.
>> Essentially giving user control over parallelism by configuring num of
>> partitions for internal topics.
>> 
>> Hope this answers your question.
>> 
>> Regards,
>> Levani
>> 
>>> On Jul 17, 2019, at 9:02 PM, Sophie Blee-Goldman 
>> wrote:
>>> 
>>> Hey Levani,
>>> 
>>> Thanks for the KIP! Can you clarify one thing for me -- for the
>>> KStream#repartition signature taking a Repartitioned, will the topic be
>>> auto-created by Streams (which seems to be the case for the signature
>>> without a Repartitioned) or does it have to be pre-created? The wording
>> in
>>> the KIP makes it seem like one version of the method will auto-create
>>> topics while the other will not.
>>> 
>>> Cheers,
>>> Sophie
>>> 
>>> On Wed, Jul 17, 2019 at 10:15 AM Levani Kokhreidze <
>> levani.co...@gmail.com>
>>> wrote:
>>> 
 Hello,
 
 One more bump about KIP-221 (
 
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
 <
 
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
>>> )
 so it doesn’t get lost in mailing list :)
 Would love to hear communities opinions/concerns about this KIP.
 
 Regards,
 Levani
 
 
> On Jul 12, 2019, at 5:27 PM, Levani Kokhreidze >> 
 wrote:
> 
> Hello,
> 
> Kind reminder about this KIP:
 
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
 <
 
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> 
> 
> Regards,
> Levani
> 
>> On Jul 9, 2019, at 11:38 AM, Levani Kokhreidze <
>> levani.co...@gmail.com
 > wrote:
>> 
>> Hello,
>> 
>> In order to move this KIP forward, I’ve updated confluence page with
 the new proposal
 
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
 <
 
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> 
>> I’ve also filled “Rejected Alternatives” section.
>> 
>> Looking forward to discuss this KIP :)
>> 
>> King regards,
>> Levani
>> 
>> 
>>> On Jul 3, 2019, at 1:08 PM, Levani Kokhreidze <
>> levani.co...@gmail.com
 > wrote:
>>> 
>>> Hello Matthias,
>>> 
>>> Thanks for the feedback and ideas.
>>> I like the idea of introducing dedicated `Topic` class for topic
 configuration for internal operators like `groupedBy`.
>>> Would be great to hear others opinion about this as well.
>>> 
>>> Kind regards,
>>> Levani
>>> 
>>> 
 On Jul 3, 2019, at 7:00 AM, Matthias J. Sax >>> > wrote:
 
 Levani,
 
 Thanks for picking up this KIP! And thanks for summarizing
>> everything.
 Even if some points may have been discussed already (can't really
 remember), it's helpful to get a good summary to refresh the
 discussion.
 
 I think your reasoning makes sense. With regard to the distinction
 between operators that manage topics and operators that use
 user-created
 topics: Following this argument, it might indicate that leaving

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Sophie Blee-Goldman]

Thanks for clearing that up. I agree that Repartitioned would be a useful
addition. I'm wondering if it might also need to have
a withStreamPartitioner method/field, similar to Produced? I'm not sure how
widely this feature is really used, but seems it should be available for
repartition topics.

On Wed, Jul 17, 2019 at 11:26 AM Levani Kokhreidze 
wrote:

> Hey Sophie,
>
> In both cases KStream#repartition and KStream#repartition(Repartitioned)
> topic will be created and managed by Kafka Streams.
> Idea of Repartitioned is to give user more control over the topic such as
> num of partitions.
> I feel like Repartitioned parameter is something that is missing in
> current DSL design.
> Essentially giving user control over parallelism by configuring num of
> partitions for internal topics.
>
> Hope this answers your question.
>
> Regards,
> Levani
>
> > On Jul 17, 2019, at 9:02 PM, Sophie Blee-Goldman 
> wrote:
> >
> > Hey Levani,
> >
> > Thanks for the KIP! Can you clarify one thing for me -- for the
> > KStream#repartition signature taking a Repartitioned, will the topic be
> > auto-created by Streams (which seems to be the case for the signature
> > without a Repartitioned) or does it have to be pre-created? The wording
> in
> > the KIP makes it seem like one version of the method will auto-create
> > topics while the other will not.
> >
> > Cheers,
> > Sophie
> >
> > On Wed, Jul 17, 2019 at 10:15 AM Levani Kokhreidze <
> levani.co...@gmail.com>
> > wrote:
> >
> >> Hello,
> >>
> >> One more bump about KIP-221 (
> >>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> >> <
> >>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> >)
> >> so it doesn’t get lost in mailing list :)
> >> Would love to hear communities opinions/concerns about this KIP.
> >>
> >> Regards,
> >> Levani
> >>
> >>
> >>> On Jul 12, 2019, at 5:27 PM, Levani Kokhreidze  >
> >> wrote:
> >>>
> >>> Hello,
> >>>
> >>> Kind reminder about this KIP:
> >>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> >> <
> >>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> >>>
> >>>
> >>> Regards,
> >>> Levani
> >>>
>  On Jul 9, 2019, at 11:38 AM, Levani Kokhreidze <
> levani.co...@gmail.com
> >> > wrote:
> 
>  Hello,
> 
>  In order to move this KIP forward, I’ve updated confluence page with
> >> the new proposal
> >>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> >> <
> >>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> >>>
>  I’ve also filled “Rejected Alternatives” section.
> 
>  Looking forward to discuss this KIP :)
> 
>  King regards,
>  Levani
> 
> 
> > On Jul 3, 2019, at 1:08 PM, Levani Kokhreidze <
> levani.co...@gmail.com
> >> > wrote:
> >
> > Hello Matthias,
> >
> > Thanks for the feedback and ideas.
> > I like the idea of introducing dedicated `Topic` class for topic
> >> configuration for internal operators like `groupedBy`.
> > Would be great to hear others opinion about this as well.
> >
> > Kind regards,
> > Levani
> >
> >
> >> On Jul 3, 2019, at 7:00 AM, Matthias J. Sax  >> > wrote:
> >>
> >> Levani,
> >>
> >> Thanks for picking up this KIP! And thanks for summarizing
> everything.
> >> Even if some points may have been discussed already (can't really
> >> remember), it's helpful to get a good summary to refresh the
> >> discussion.
> >>
> >> I think your reasoning makes sense. With regard to the distinction
> >> between operators that manage topics and operators that use
> >> user-created
> >> topics: Following this argument, it might indicate that leaving
> >> `through()` as-is (as an operator that uses use-defined topics) and
> >> introducing a new `repartition()` operator (an operator that manages
> >> topics itself) might be good. Otherwise, there is one operator
> >> `through()` that sometimes manages topics but sometimes not; a
> >> different
> >> name, ie, new operator would make the distinction clearer.
> >>
> >> About adding `numOfPartitions` to `Grouped`. I am wondering if the
> >> same
> >> argument as for `Produced` does apply and adding it is semantically
> >> questionable? Might be good to get opinions of others on this, too.
> I
> >> am
> >> not sure myself what solution I prefer atm.
> >>
> >> So far, KS uses configuration objects that allow to configure a

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Levani Kokhreidze]

Hey Sophie,

In both cases KStream#repartition and KStream#repartition(Repartitioned) topic 
will be created and managed by Kafka Streams. 
Idea of Repartitioned is to give user more control over the topic such as num 
of partitions. 
I feel like Repartitioned parameter is something that is missing in current DSL 
design. 
Essentially giving user control over parallelism by configuring num of 
partitions for internal topics.

Hope this answers your question.

Regards,
Levani

> On Jul 17, 2019, at 9:02 PM, Sophie Blee-Goldman  wrote:
> 
> Hey Levani,
> 
> Thanks for the KIP! Can you clarify one thing for me -- for the
> KStream#repartition signature taking a Repartitioned, will the topic be
> auto-created by Streams (which seems to be the case for the signature
> without a Repartitioned) or does it have to be pre-created? The wording in
> the KIP makes it seem like one version of the method will auto-create
> topics while the other will not.
> 
> Cheers,
> Sophie
> 
> On Wed, Jul 17, 2019 at 10:15 AM Levani Kokhreidze 
> wrote:
> 
>> Hello,
>> 
>> One more bump about KIP-221 (
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
>> <
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint>)
>> so it doesn’t get lost in mailing list :)
>> Would love to hear communities opinions/concerns about this KIP.
>> 
>> Regards,
>> Levani
>> 
>> 
>>> On Jul 12, 2019, at 5:27 PM, Levani Kokhreidze 
>> wrote:
>>> 
>>> Hello,
>>> 
>>> Kind reminder about this KIP:
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
>> <
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
>>> 
>>> 
>>> Regards,
>>> Levani
>>> 
 On Jul 9, 2019, at 11:38 AM, Levani Kokhreidze > > wrote:
 
 Hello,
 
 In order to move this KIP forward, I’ve updated confluence page with
>> the new proposal
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
>> <
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
>>> 
 I’ve also filled “Rejected Alternatives” section.
 
 Looking forward to discuss this KIP :)
 
 King regards,
 Levani
 
 
> On Jul 3, 2019, at 1:08 PM, Levani Kokhreidze > > wrote:
> 
> Hello Matthias,
> 
> Thanks for the feedback and ideas.
> I like the idea of introducing dedicated `Topic` class for topic
>> configuration for internal operators like `groupedBy`.
> Would be great to hear others opinion about this as well.
> 
> Kind regards,
> Levani
> 
> 
>> On Jul 3, 2019, at 7:00 AM, Matthias J. Sax > > wrote:
>> 
>> Levani,
>> 
>> Thanks for picking up this KIP! And thanks for summarizing everything.
>> Even if some points may have been discussed already (can't really
>> remember), it's helpful to get a good summary to refresh the
>> discussion.
>> 
>> I think your reasoning makes sense. With regard to the distinction
>> between operators that manage topics and operators that use
>> user-created
>> topics: Following this argument, it might indicate that leaving
>> `through()` as-is (as an operator that uses use-defined topics) and
>> introducing a new `repartition()` operator (an operator that manages
>> topics itself) might be good. Otherwise, there is one operator
>> `through()` that sometimes manages topics but sometimes not; a
>> different
>> name, ie, new operator would make the distinction clearer.
>> 
>> About adding `numOfPartitions` to `Grouped`. I am wondering if the
>> same
>> argument as for `Produced` does apply and adding it is semantically
>> questionable? Might be good to get opinions of others on this, too. I
>> am
>> not sure myself what solution I prefer atm.
>> 
>> So far, KS uses configuration objects that allow to configure a
>> certain
>> "entity" like a consumer, producer, store. If we assume that a topic
>> is
>> a similar entity, I am wonder if we should have a
>> `Topic#withNumberOfPartitions()` class and method instead of a plain
>> integer? This would allow us to add other configs, like replication
>> factor, retention-time etc, easily, without the need to change the
>> "main
>> API".
>> 
>> Just want to give some ideas. Not sure if I like them myself. :)
>> 
>> 
>> -Matthias
>> 
>> 
>> 
>> On 7/1/19 1:04 AM, Levani Kokhreidze wrote:
>>> Actually, giving it more though - maybe enhancing Produced with num
>> of 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Sophie Blee-Goldman]

Hey Levani,

Thanks for the KIP! Can you clarify one thing for me -- for the
KStream#repartition signature taking a Repartitioned, will the topic be
auto-created by Streams (which seems to be the case for the signature
without a Repartitioned) or does it have to be pre-created? The wording in
the KIP makes it seem like one version of the method will auto-create
topics while the other will not.

Cheers,
Sophie

On Wed, Jul 17, 2019 at 10:15 AM Levani Kokhreidze 
wrote:

> Hello,
>
> One more bump about KIP-221 (
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> <
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint>)
> so it doesn’t get lost in mailing list :)
> Would love to hear communities opinions/concerns about this KIP.
>
> Regards,
> Levani
>
>
> > On Jul 12, 2019, at 5:27 PM, Levani Kokhreidze 
> wrote:
> >
> > Hello,
> >
> > Kind reminder about this KIP:
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> <
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> >
> >
> > Regards,
> > Levani
> >
> >> On Jul 9, 2019, at 11:38 AM, Levani Kokhreidze  > wrote:
> >>
> >> Hello,
> >>
> >> In order to move this KIP forward, I’ve updated confluence page with
> the new proposal
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> <
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221:+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
> >
> >> I’ve also filled “Rejected Alternatives” section.
> >>
> >> Looking forward to discuss this KIP :)
> >>
> >> King regards,
> >> Levani
> >>
> >>
> >>> On Jul 3, 2019, at 1:08 PM, Levani Kokhreidze  > wrote:
> >>>
> >>> Hello Matthias,
> >>>
> >>> Thanks for the feedback and ideas.
> >>> I like the idea of introducing dedicated `Topic` class for topic
> configuration for internal operators like `groupedBy`.
> >>> Would be great to hear others opinion about this as well.
> >>>
> >>> Kind regards,
> >>> Levani
> >>>
> >>>
>  On Jul 3, 2019, at 7:00 AM, Matthias J. Sax  > wrote:
> 
>  Levani,
> 
>  Thanks for picking up this KIP! And thanks for summarizing everything.
>  Even if some points may have been discussed already (can't really
>  remember), it's helpful to get a good summary to refresh the
> discussion.
> 
>  I think your reasoning makes sense. With regard to the distinction
>  between operators that manage topics and operators that use
> user-created
>  topics: Following this argument, it might indicate that leaving
>  `through()` as-is (as an operator that uses use-defined topics) and
>  introducing a new `repartition()` operator (an operator that manages
>  topics itself) might be good. Otherwise, there is one operator
>  `through()` that sometimes manages topics but sometimes not; a
> different
>  name, ie, new operator would make the distinction clearer.
> 
>  About adding `numOfPartitions` to `Grouped`. I am wondering if the
> same
>  argument as for `Produced` does apply and adding it is semantically
>  questionable? Might be good to get opinions of others on this, too. I
> am
>  not sure myself what solution I prefer atm.
> 
>  So far, KS uses configuration objects that allow to configure a
> certain
>  "entity" like a consumer, producer, store. If we assume that a topic
> is
>  a similar entity, I am wonder if we should have a
>  `Topic#withNumberOfPartitions()` class and method instead of a plain
>  integer? This would allow us to add other configs, like replication
>  factor, retention-time etc, easily, without the need to change the
> "main
>  API".
> 
>  Just want to give some ideas. Not sure if I like them myself. :)
> 
> 
>  -Matthias
> 
> 
> 
>  On 7/1/19 1:04 AM, Levani Kokhreidze wrote:
> > Actually, giving it more though - maybe enhancing Produced with num
> of partitions configuration is not the best approach. Let me explain why:
> >
> > 1) If we enhance Produced class with this configuration, this will
> also affect KStream#to operation. Since KStream#to is the final sink of the
> topology, for me, it seems to be reasonable assumption that user needs to
> manually create sink topic in advance. And in that case, having num of
> partitions configuration doesn’t make much sense.
> >
> > 2) Looking at Produced class, based on API contract, seems like
> Produced is designed to be something that is explicitly for producer (key
> serializer, value serializer, partitioner those 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Levani Kokhreidze]

Hello,

One more bump about KIP-221 
(https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
 
)
 so it doesn’t get lost in mailing list :)
Would love to hear communities opinions/concerns about this KIP.

Regards,
Levani


> On Jul 12, 2019, at 5:27 PM, Levani Kokhreidze  wrote:
> 
> Hello,
> 
> Kind reminder about this KIP: 
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
>  
> 
> 
> Regards,
> Levani
> 
>> On Jul 9, 2019, at 11:38 AM, Levani Kokhreidze > > wrote:
>> 
>> Hello,
>> 
>> In order to move this KIP forward, I’ve updated confluence page with the new 
>> proposal 
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
>>  
>> 
>> I’ve also filled “Rejected Alternatives” section. 
>> 
>> Looking forward to discuss this KIP :)
>> 
>> King regards,
>> Levani
>> 
>> 
>>> On Jul 3, 2019, at 1:08 PM, Levani Kokhreidze >> > wrote:
>>> 
>>> Hello Matthias,
>>> 
>>> Thanks for the feedback and ideas. 
>>> I like the idea of introducing dedicated `Topic` class for topic 
>>> configuration for internal operators like `groupedBy`.
>>> Would be great to hear others opinion about this as well.
>>> 
>>> Kind regards,
>>> Levani 
>>> 
>>> 
 On Jul 3, 2019, at 7:00 AM, Matthias J. Sax >>> > wrote:
 
 Levani,
 
 Thanks for picking up this KIP! And thanks for summarizing everything.
 Even if some points may have been discussed already (can't really
 remember), it's helpful to get a good summary to refresh the discussion.
 
 I think your reasoning makes sense. With regard to the distinction
 between operators that manage topics and operators that use user-created
 topics: Following this argument, it might indicate that leaving
 `through()` as-is (as an operator that uses use-defined topics) and
 introducing a new `repartition()` operator (an operator that manages
 topics itself) might be good. Otherwise, there is one operator
 `through()` that sometimes manages topics but sometimes not; a different
 name, ie, new operator would make the distinction clearer.
 
 About adding `numOfPartitions` to `Grouped`. I am wondering if the same
 argument as for `Produced` does apply and adding it is semantically
 questionable? Might be good to get opinions of others on this, too. I am
 not sure myself what solution I prefer atm.
 
 So far, KS uses configuration objects that allow to configure a certain
 "entity" like a consumer, producer, store. If we assume that a topic is
 a similar entity, I am wonder if we should have a
 `Topic#withNumberOfPartitions()` class and method instead of a plain
 integer? This would allow us to add other configs, like replication
 factor, retention-time etc, easily, without the need to change the "main
 API".
 
 Just want to give some ideas. Not sure if I like them myself. :)
 
 
 -Matthias
 
 
 
 On 7/1/19 1:04 AM, Levani Kokhreidze wrote:
> Actually, giving it more though - maybe enhancing Produced with num of 
> partitions configuration is not the best approach. Let me explain why:
> 
> 1) If we enhance Produced class with this configuration, this will also 
> affect KStream#to operation. Since KStream#to is the final sink of the 
> topology, for me, it seems to be reasonable assumption that user needs to 
> manually create sink topic in advance. And in that case, having num of 
> partitions configuration doesn’t make much sense. 
> 
> 2) Looking at Produced class, based on API contract, seems like Produced 
> is designed to be something that is explicitly for producer (key 
> serializer, value serializer, partitioner those all are producer specific 
> configurations) and num of partitions is topic level configuration. And I 
> don’t think mixing topic and producer level configurations together in 
> one class is the good approach.
> 
> 3) Looking at KStream interface, seems like Produced parameter is for 
> operations that work with non-internal (e.g topics created and managed 
> internally by Kafka Streams) topics and I think we should leave it as it 
> is in that case.
> 
> Taking all this things into account, I think we should distinguish 
> between DSL 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Levani Kokhreidze]

Hello,

Kind reminder about this KIP: 
https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
 


Regards,
Levani

> On Jul 9, 2019, at 11:38 AM, Levani Kokhreidze  wrote:
> 
> Hello,
> 
> In order to move this KIP forward, I’ve updated confluence page with the new 
> proposal 
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
>  
> 
> I’ve also filled “Rejected Alternatives” section. 
> 
> Looking forward to discuss this KIP :)
> 
> King regards,
> Levani
> 
> 
>> On Jul 3, 2019, at 1:08 PM, Levani Kokhreidze > > wrote:
>> 
>> Hello Matthias,
>> 
>> Thanks for the feedback and ideas. 
>> I like the idea of introducing dedicated `Topic` class for topic 
>> configuration for internal operators like `groupedBy`.
>> Would be great to hear others opinion about this as well.
>> 
>> Kind regards,
>> Levani 
>> 
>> 
>>> On Jul 3, 2019, at 7:00 AM, Matthias J. Sax >> > wrote:
>>> 
>>> Levani,
>>> 
>>> Thanks for picking up this KIP! And thanks for summarizing everything.
>>> Even if some points may have been discussed already (can't really
>>> remember), it's helpful to get a good summary to refresh the discussion.
>>> 
>>> I think your reasoning makes sense. With regard to the distinction
>>> between operators that manage topics and operators that use user-created
>>> topics: Following this argument, it might indicate that leaving
>>> `through()` as-is (as an operator that uses use-defined topics) and
>>> introducing a new `repartition()` operator (an operator that manages
>>> topics itself) might be good. Otherwise, there is one operator
>>> `through()` that sometimes manages topics but sometimes not; a different
>>> name, ie, new operator would make the distinction clearer.
>>> 
>>> About adding `numOfPartitions` to `Grouped`. I am wondering if the same
>>> argument as for `Produced` does apply and adding it is semantically
>>> questionable? Might be good to get opinions of others on this, too. I am
>>> not sure myself what solution I prefer atm.
>>> 
>>> So far, KS uses configuration objects that allow to configure a certain
>>> "entity" like a consumer, producer, store. If we assume that a topic is
>>> a similar entity, I am wonder if we should have a
>>> `Topic#withNumberOfPartitions()` class and method instead of a plain
>>> integer? This would allow us to add other configs, like replication
>>> factor, retention-time etc, easily, without the need to change the "main
>>> API".
>>> 
>>> Just want to give some ideas. Not sure if I like them myself. :)
>>> 
>>> 
>>> -Matthias
>>> 
>>> 
>>> 
>>> On 7/1/19 1:04 AM, Levani Kokhreidze wrote:
 Actually, giving it more though - maybe enhancing Produced with num of 
 partitions configuration is not the best approach. Let me explain why:
 
 1) If we enhance Produced class with this configuration, this will also 
 affect KStream#to operation. Since KStream#to is the final sink of the 
 topology, for me, it seems to be reasonable assumption that user needs to 
 manually create sink topic in advance. And in that case, having num of 
 partitions configuration doesn’t make much sense. 
 
 2) Looking at Produced class, based on API contract, seems like Produced 
 is designed to be something that is explicitly for producer (key 
 serializer, value serializer, partitioner those all are producer specific 
 configurations) and num of partitions is topic level configuration. And I 
 don’t think mixing topic and producer level configurations together in one 
 class is the good approach.
 
 3) Looking at KStream interface, seems like Produced parameter is for 
 operations that work with non-internal (e.g topics created and managed 
 internally by Kafka Streams) topics and I think we should leave it as it 
 is in that case.
 
 Taking all this things into account, I think we should distinguish between 
 DSL operations, where Kafka Streams should create and manage internal 
 topics (KStream#groupBy) vs topics that should be created in advance (e.g 
 KStream#to).
 
 To sum it up, I think adding numPartitions configuration in Produced will 
 result in mixing topic and producer level configuration in one class and 
 it’s gonna break existing API semantics.
 
 Regarding making topic name optional in KStream#through - I think 
 underline idea is very useful and giving users possibility to specify num 
 of partitions there is even more useful :) Considering arguments against 
 adding num of 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Levani Kokhreidze]

Hello,

In order to move this KIP forward, I’ve updated confluence page with the new 
proposal 
https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+KStream+with+Connecting+Topic+Creation+and+Repartition+Hint
 

I’ve also filled “Rejected Alternatives” section. 

Looking forward to discuss this KIP :)

King regards,
Levani


> On Jul 3, 2019, at 1:08 PM, Levani Kokhreidze  wrote:
> 
> Hello Matthias,
> 
> Thanks for the feedback and ideas. 
> I like the idea of introducing dedicated `Topic` class for topic 
> configuration for internal operators like `groupedBy`.
> Would be great to hear others opinion about this as well.
> 
> Kind regards,
> Levani 
> 
> 
>> On Jul 3, 2019, at 7:00 AM, Matthias J. Sax  wrote:
>> 
>> Levani,
>> 
>> Thanks for picking up this KIP! And thanks for summarizing everything.
>> Even if some points may have been discussed already (can't really
>> remember), it's helpful to get a good summary to refresh the discussion.
>> 
>> I think your reasoning makes sense. With regard to the distinction
>> between operators that manage topics and operators that use user-created
>> topics: Following this argument, it might indicate that leaving
>> `through()` as-is (as an operator that uses use-defined topics) and
>> introducing a new `repartition()` operator (an operator that manages
>> topics itself) might be good. Otherwise, there is one operator
>> `through()` that sometimes manages topics but sometimes not; a different
>> name, ie, new operator would make the distinction clearer.
>> 
>> About adding `numOfPartitions` to `Grouped`. I am wondering if the same
>> argument as for `Produced` does apply and adding it is semantically
>> questionable? Might be good to get opinions of others on this, too. I am
>> not sure myself what solution I prefer atm.
>> 
>> So far, KS uses configuration objects that allow to configure a certain
>> "entity" like a consumer, producer, store. If we assume that a topic is
>> a similar entity, I am wonder if we should have a
>> `Topic#withNumberOfPartitions()` class and method instead of a plain
>> integer? This would allow us to add other configs, like replication
>> factor, retention-time etc, easily, without the need to change the "main
>> API".
>> 
>> Just want to give some ideas. Not sure if I like them myself. :)
>> 
>> 
>> -Matthias
>> 
>> 
>> 
>> On 7/1/19 1:04 AM, Levani Kokhreidze wrote:
>>> Actually, giving it more though - maybe enhancing Produced with num of 
>>> partitions configuration is not the best approach. Let me explain why:
>>> 
>>> 1) If we enhance Produced class with this configuration, this will also 
>>> affect KStream#to operation. Since KStream#to is the final sink of the 
>>> topology, for me, it seems to be reasonable assumption that user needs to 
>>> manually create sink topic in advance. And in that case, having num of 
>>> partitions configuration doesn’t make much sense. 
>>> 
>>> 2) Looking at Produced class, based on API contract, seems like Produced is 
>>> designed to be something that is explicitly for producer (key serializer, 
>>> value serializer, partitioner those all are producer specific 
>>> configurations) and num of partitions is topic level configuration. And I 
>>> don’t think mixing topic and producer level configurations together in one 
>>> class is the good approach.
>>> 
>>> 3) Looking at KStream interface, seems like Produced parameter is for 
>>> operations that work with non-internal (e.g topics created and managed 
>>> internally by Kafka Streams) topics and I think we should leave it as it is 
>>> in that case.
>>> 
>>> Taking all this things into account, I think we should distinguish between 
>>> DSL operations, where Kafka Streams should create and manage internal 
>>> topics (KStream#groupBy) vs topics that should be created in advance (e.g 
>>> KStream#to).
>>> 
>>> To sum it up, I think adding numPartitions configuration in Produced will 
>>> result in mixing topic and producer level configuration in one class and 
>>> it’s gonna break existing API semantics.
>>> 
>>> Regarding making topic name optional in KStream#through - I think underline 
>>> idea is very useful and giving users possibility to specify num of 
>>> partitions there is even more useful :) Considering arguments against 
>>> adding num of partitions in Produced class, I see two options here:
>>> 1) Add following method overloads
>>> * through() - topic will be auto-generated and num of partitions will 
>>> be taken from source topic
>>> * through(final int numOfPartitions) - topic will be auto generated 
>>> with specified num of partitions
>>> * through(final int numOfPartitions, final Produced produced) - 
>>> topic will be with generated with specified num of partitions and 
>>> configuration taken from produced parameter.
>>> 2) Leave KStream#through as it is and introduce 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Levani Kokhreidze]

Hello Matthias,

Thanks for the feedback and ideas. 
I like the idea of introducing dedicated `Topic` class for topic configuration 
for internal operators like `groupedBy`.
Would be great to hear others opinion about this as well.

Kind regards,
Levani 


> On Jul 3, 2019, at 7:00 AM, Matthias J. Sax  wrote:
> 
> Levani,
> 
> Thanks for picking up this KIP! And thanks for summarizing everything.
> Even if some points may have been discussed already (can't really
> remember), it's helpful to get a good summary to refresh the discussion.
> 
> I think your reasoning makes sense. With regard to the distinction
> between operators that manage topics and operators that use user-created
> topics: Following this argument, it might indicate that leaving
> `through()` as-is (as an operator that uses use-defined topics) and
> introducing a new `repartition()` operator (an operator that manages
> topics itself) might be good. Otherwise, there is one operator
> `through()` that sometimes manages topics but sometimes not; a different
> name, ie, new operator would make the distinction clearer.
> 
> About adding `numOfPartitions` to `Grouped`. I am wondering if the same
> argument as for `Produced` does apply and adding it is semantically
> questionable? Might be good to get opinions of others on this, too. I am
> not sure myself what solution I prefer atm.
> 
> So far, KS uses configuration objects that allow to configure a certain
> "entity" like a consumer, producer, store. If we assume that a topic is
> a similar entity, I am wonder if we should have a
> `Topic#withNumberOfPartitions()` class and method instead of a plain
> integer? This would allow us to add other configs, like replication
> factor, retention-time etc, easily, without the need to change the "main
> API".
> 
> Just want to give some ideas. Not sure if I like them myself. :)
> 
> 
> -Matthias
> 
> 
> 
> On 7/1/19 1:04 AM, Levani Kokhreidze wrote:
>> Actually, giving it more though - maybe enhancing Produced with num of 
>> partitions configuration is not the best approach. Let me explain why:
>> 
>> 1) If we enhance Produced class with this configuration, this will also 
>> affect KStream#to operation. Since KStream#to is the final sink of the 
>> topology, for me, it seems to be reasonable assumption that user needs to 
>> manually create sink topic in advance. And in that case, having num of 
>> partitions configuration doesn’t make much sense. 
>> 
>> 2) Looking at Produced class, based on API contract, seems like Produced is 
>> designed to be something that is explicitly for producer (key serializer, 
>> value serializer, partitioner those all are producer specific 
>> configurations) and num of partitions is topic level configuration. And I 
>> don’t think mixing topic and producer level configurations together in one 
>> class is the good approach.
>> 
>> 3) Looking at KStream interface, seems like Produced parameter is for 
>> operations that work with non-internal (e.g topics created and managed 
>> internally by Kafka Streams) topics and I think we should leave it as it is 
>> in that case.
>> 
>> Taking all this things into account, I think we should distinguish between 
>> DSL operations, where Kafka Streams should create and manage internal topics 
>> (KStream#groupBy) vs topics that should be created in advance (e.g 
>> KStream#to).
>> 
>> To sum it up, I think adding numPartitions configuration in Produced will 
>> result in mixing topic and producer level configuration in one class and 
>> it’s gonna break existing API semantics.
>> 
>> Regarding making topic name optional in KStream#through - I think underline 
>> idea is very useful and giving users possibility to specify num of 
>> partitions there is even more useful :) Considering arguments against adding 
>> num of partitions in Produced class, I see two options here:
>> 1) Add following method overloads
>>  * through() - topic will be auto-generated and num of partitions will 
>> be taken from source topic
>>  * through(final int numOfPartitions) - topic will be auto generated 
>> with specified num of partitions
>>  * through(final int numOfPartitions, final Produced produced) - 
>> topic will be with generated with specified num of partitions and 
>> configuration taken from produced parameter.
>> 2) Leave KStream#through as it is and introduce new method - 
>> KStream#repartition (I think Matthias suggested this in one of the threads)
>> 
>> Considering all mentioned above I propose the following plan:
>> 
>> Option A:
>> 1) Leave Produced as it is
>> 2) Add num of partitions configuration to Grouped class (as mentioned in the 
>> KIP)
>> 3) Add following method overloads to KStream#through
>>  * through() - topic will be auto-generated and num of partitions will 
>> be taken from source topic
>>  * through(final int numOfPartitions) - topic will be auto generated 
>> with specified num of partitions
>>  * through(final int numOfPartitions, final 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Matthias J. Sax]

Levani,

Thanks for picking up this KIP! And thanks for summarizing everything.
Even if some points may have been discussed already (can't really
remember), it's helpful to get a good summary to refresh the discussion.

I think your reasoning makes sense. With regard to the distinction
between operators that manage topics and operators that use user-created
topics: Following this argument, it might indicate that leaving
`through()` as-is (as an operator that uses use-defined topics) and
introducing a new `repartition()` operator (an operator that manages
topics itself) might be good. Otherwise, there is one operator
`through()` that sometimes manages topics but sometimes not; a different
name, ie, new operator would make the distinction clearer.

About adding `numOfPartitions` to `Grouped`. I am wondering if the same
argument as for `Produced` does apply and adding it is semantically
questionable? Might be good to get opinions of others on this, too. I am
not sure myself what solution I prefer atm.

So far, KS uses configuration objects that allow to configure a certain
"entity" like a consumer, producer, store. If we assume that a topic is
a similar entity, I am wonder if we should have a
`Topic#withNumberOfPartitions()` class and method instead of a plain
integer? This would allow us to add other configs, like replication
factor, retention-time etc, easily, without the need to change the "main
API".

Just want to give some ideas. Not sure if I like them myself. :)


-Matthias



On 7/1/19 1:04 AM, Levani Kokhreidze wrote:
> Actually, giving it more though - maybe enhancing Produced with num of 
> partitions configuration is not the best approach. Let me explain why:
> 
> 1) If we enhance Produced class with this configuration, this will also 
> affect KStream#to operation. Since KStream#to is the final sink of the 
> topology, for me, it seems to be reasonable assumption that user needs to 
> manually create sink topic in advance. And in that case, having num of 
> partitions configuration doesn’t make much sense. 
> 
> 2) Looking at Produced class, based on API contract, seems like Produced is 
> designed to be something that is explicitly for producer (key serializer, 
> value serializer, partitioner those all are producer specific configurations) 
> and num of partitions is topic level configuration. And I don’t think mixing 
> topic and producer level configurations together in one class is the good 
> approach.
> 
> 3) Looking at KStream interface, seems like Produced parameter is for 
> operations that work with non-internal (e.g topics created and managed 
> internally by Kafka Streams) topics and I think we should leave it as it is 
> in that case.
> 
> Taking all this things into account, I think we should distinguish between 
> DSL operations, where Kafka Streams should create and manage internal topics 
> (KStream#groupBy) vs topics that should be created in advance (e.g 
> KStream#to).
> 
> To sum it up, I think adding numPartitions configuration in Produced will 
> result in mixing topic and producer level configuration in one class and it’s 
> gonna break existing API semantics.
> 
> Regarding making topic name optional in KStream#through - I think underline 
> idea is very useful and giving users possibility to specify num of partitions 
> there is even more useful :) Considering arguments against adding num of 
> partitions in Produced class, I see two options here:
> 1) Add following method overloads
>   * through() - topic will be auto-generated and num of partitions will 
> be taken from source topic
>   * through(final int numOfPartitions) - topic will be auto generated 
> with specified num of partitions
>   * through(final int numOfPartitions, final Produced produced) - 
> topic will be with generated with specified num of partitions and 
> configuration taken from produced parameter.
> 2) Leave KStream#through as it is and introduce new method - 
> KStream#repartition (I think Matthias suggested this in one of the threads)
> 
> Considering all mentioned above I propose the following plan:
> 
> Option A:
> 1) Leave Produced as it is
> 2) Add num of partitions configuration to Grouped class (as mentioned in the 
> KIP)
> 3) Add following method overloads to KStream#through
>   * through() - topic will be auto-generated and num of partitions will 
> be taken from source topic
>   * through(final int numOfPartitions) - topic will be auto generated 
> with specified num of partitions
>   * through(final int numOfPartitions, final Produced produced) - 
> topic will be with generated with specified num of partitions and 
> configuration taken from produced parameter.
> 
> Option B:
> 1) Leave Produced as it is
> 2) Add num of partitions configuration to Grouped class (as mentioned in the 
> KIP)
> 3) Add new operator KStream#repartition for creating and managing internal 
> repartition topics
> 
> P.S. I’m sorry if all of this was already discussed in the mailing 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Levani Kokhreidze]

Actually, giving it more though - maybe enhancing Produced with num of 
partitions configuration is not the best approach. Let me explain why:

1) If we enhance Produced class with this configuration, this will also affect 
KStream#to operation. Since KStream#to is the final sink of the topology, for 
me, it seems to be reasonable assumption that user needs to manually create 
sink topic in advance. And in that case, having num of partitions configuration 
doesn’t make much sense. 

2) Looking at Produced class, based on API contract, seems like Produced is 
designed to be something that is explicitly for producer (key serializer, value 
serializer, partitioner those all are producer specific configurations) and num 
of partitions is topic level configuration. And I don’t think mixing topic and 
producer level configurations together in one class is the good approach.

3) Looking at KStream interface, seems like Produced parameter is for 
operations that work with non-internal (e.g topics created and managed 
internally by Kafka Streams) topics and I think we should leave it as it is in 
that case.

Taking all this things into account, I think we should distinguish between DSL 
operations, where Kafka Streams should create and manage internal topics 
(KStream#groupBy) vs topics that should be created in advance (e.g KStream#to).

To sum it up, I think adding numPartitions configuration in Produced will 
result in mixing topic and producer level configuration in one class and it’s 
gonna break existing API semantics.

Regarding making topic name optional in KStream#through - I think underline 
idea is very useful and giving users possibility to specify num of partitions 
there is even more useful :) Considering arguments against adding num of 
partitions in Produced class, I see two options here:
1) Add following method overloads
* through() - topic will be auto-generated and num of partitions will 
be taken from source topic
* through(final int numOfPartitions) - topic will be auto generated 
with specified num of partitions
* through(final int numOfPartitions, final Produced produced) - 
topic will be with generated with specified num of partitions and configuration 
taken from produced parameter.
2) Leave KStream#through as it is and introduce new method - 
KStream#repartition (I think Matthias suggested this in one of the threads)

Considering all mentioned above I propose the following plan:

Option A:
1) Leave Produced as it is
2) Add num of partitions configuration to Grouped class (as mentioned in the 
KIP)
3) Add following method overloads to KStream#through
* through() - topic will be auto-generated and num of partitions will 
be taken from source topic
* through(final int numOfPartitions) - topic will be auto generated 
with specified num of partitions
* through(final int numOfPartitions, final Produced produced) - 
topic will be with generated with specified num of partitions and configuration 
taken from produced parameter.

Option B:
1) Leave Produced as it is
2) Add num of partitions configuration to Grouped class (as mentioned in the 
KIP)
3) Add new operator KStream#repartition for creating and managing internal 
repartition topics

P.S. I’m sorry if all of this was already discussed in the mailing list, but I 
kinda got with all the threads that were about this KIP :(

Kind regards,
Levani

> On Jul 1, 2019, at 9:56 AM, Levani Kokhreidze  wrote:
> 
> Hello,
> 
> I would like to resurrect discussion around KIP-221. Going through the 
> discussion thread, there’s seems to agreement around usefulness of this 
> feature. 
> Regarding the implementation, as far as I understood, the most optimal 
> solution for me seems the following:
> 
> 1) Add two method overloads to KStream#through method (essentially making 
> topic name optional)
> 2) Enhance Produced class with numOfPartitions configuration field.
> 
> Those two changes will allow DSL users to control parallelism and trigger 
> re-partition without doing stateful operations.
> 
> I will update KIP with interface changes around KStream#through if this 
> changes sound sensible.
> 
> Kind regards,
> Levani

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Levani Kokhreidze]

Hello,

I would like to resurrect discussion around KIP-221. Going through the 
discussion thread, there’s seems to agreement around usefulness of this 
feature. 
Regarding the implementation, as far as I understood, the most optimal solution 
for me seems the following:

1) Add two method overloads to KStream#through method (essentially making topic 
name optional)
2) Enhance Produced class with numOfPartitions configuration field.

Those two changes will allow DSL users to control parallelism and trigger 
re-partition without doing stateful operations.

I will update KIP with interface changes around KStream#through if this changes 
sound sensible.

Kind regards,
Levani

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Jeyhun Karimov]

Hi Lei,

Please feel free to take over the KIP.

Cheers,
Jeyhun

On Fri, Sep 21, 2018, 22:27 Lei Chen  wrote:

> Hi,
>
> Just want to know is anyone actively working on this and also KAFKA-4835
> ? Seems like the JIRA
> has been inactive for couple months. We want this feature and would like to
> move it forward if no one else is working on it.
>
>
> Lei
>
> On Wed, Jun 20, 2018 at 7:27 PM Matthias J. Sax 
> wrote:
>
>> No worries. It's just good to know. It seems that some other people are
>> interested to drive this further. So we will just "reassign" it to them.
>>
>> Thanks for letting us know.
>>
>>
>> -Matthias
>>
>> On 6/20/18 2:51 PM, Jeyhun Karimov wrote:
>> > Hi Matthias, all,
>> >
>> > Currently, I am not able to complete this KIP. Please accept my
>> > apologies for that.
>> >
>> >
>> > Cheers,
>> > Jeyhun
>> >
>> > On Mon, Jun 11, 2018 at 2:25 AM Matthias J. Sax > > > wrote:
>> >
>> > What is the status of this KIP?
>> >
>> > -Matthias
>> >
>> >
>> > On 2/13/18 1:43 PM, Matthias J. Sax wrote:
>> > > Is there any update for this KIP?
>> > >
>> > >
>> > > -Matthias
>> > >
>> > > On 12/4/17 2:08 PM, Matthias J. Sax wrote:
>> > >> Jeyhun,
>> > >>
>> > >> thanks for updating the KIP.
>> > >>
>> > >> I am wondering if you intend to add a new class `Produced`?
>> There is
>> > >> already `org.apache.kafka.streams.kstream.Produced`. So if we
>> want to
>> > >> add a new class, it must have a different name -- or we might be
>> > able to
>> > >> merge both into one?
>> > >>
>> > >> Also, for the KStream overlaods of `through()` and `to()`, can
>> > you add
>> > >> the different behavior using different overloads? It's not clear
>> from
>> > >> the KIP what the semantics are.
>> > >>
>> > >>
>> > >> -Matthias
>> > >>
>> > >> On 11/17/17 3:27 PM, Jeyhun Karimov wrote:
>> > >>> Hi,
>> > >>>
>> > >>> Thanks for your comments. I agree with Matthias partially.
>> > >>> I think we should relax some requirements related with to() and
>> > through()
>> > >>> methods.
>> > >>> IMHO, Produced class can cover (existing/to be created) topic
>> > information,
>> > >>> and which will ease our effort:
>> > >>>
>> > >>> KStream.to(Produced topicInfo)
>> > >>> KStream.through(Produced topicInfo)
>> > >>>
>> > >>> This will decrease the number of overloads but we will need to
>> > deprecate
>> > >>> the existing to() and through() methods, perhaps.
>> > >>> I updated the KIP accordingly.
>> > >>>
>> > >>>
>> > >>> Cheers,
>> > >>> Jeyhun
>> > >>>
>> > >>> On Thu, Nov 16, 2017 at 10:21 PM Matthias J. Sax
>> > mailto:matth...@confluent.io>>
>> > >>> wrote:
>> > >>>
>> >  @Jan:
>> > 
>> >  The `Produced` class was introduced in 1.0 to specify key and
>> valud
>> >  Serdes (and partitioner) if data is written into a topic.
>> > 
>> >  Old API:
>> > 
>> >  KStream#to("topic", keySerde, valueSerde);
>> > 
>> >  New API:
>> > 
>> >  KStream#to("topic", Produced.with(keySerde, valueSerde));
>> > 
>> > 
>> >  This allows to reduce the number of overloads for `to()` (and
>> >  `through()` that follows the same pattern) -- the second
>> > parameter is
>> >  used to cover all different variations of option parameters
>> > users can
>> >  specify, while we only have 2 overload for `to()` itself.
>> > 
>> >  What is still unclear to me it, what you mean by this topic
>> prefix
>> >  thing? Either a user cares about the topic name and thus, must
>> > create
>> >  and manage it manually. Or the user does not care, and Streams
>> > create
>> >  it. How would this prefix idea fit in here?
>> > 
>> > 
>> > 
>> >  @Guozhang:
>> > 
>> >  My idea was to extend `Produced` with the hint we want to give
>> for
>> >  creating internal topic and pass a optional `Produced`
>> > parameter. There
>> >  are multiple things we can do here:
>> > 
>> >  1) stream.through(null, Produced...).groupBy().aggregate()
>> >  -> just allow for `null` topic name indicating that Streams
>> should
>> >  create an internal topic
>> > 
>> >  2) stream.through(Produced...).groupBy().aggregate()
>> >  -> add one overload taking an mandatory `Produced`
>> > 
>> >  We use `Serialized` to picky back the information
>> > 
>> >  3) stream.groupBy(Serialized...).aggregate()
>> >  and stream.groupByKey(Serialized...).aggregate()
>> >  -> we don't need new top level overloads
>> > 
>> > 
>> >  

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Matthias J. Sax]

It seems Jeyhun (cc'ed) is not working on the KIP any longer. If there
is no response within a week from Jeyhun, feel free to take over the KIP.

One more side comment: we recently accepted KIP-372, that overlaps with
this KIP. Thus, if you resume KIP-221, please consider the changes of
KIP-372.


Thanks a lot!


-Matthias

On 9/21/18 11:27 AM, Lei Chen wrote:
> Hi,
> 
> Just want to know is anyone actively working on this and also KAFKA-4835
> ? Seems like the JIRA has
> been inactive for couple months. We want this feature and would like to
> move it forward if no one else is working on it.
> 
> Lei
> 
> On Wed, Jun 20, 2018 at 7:27 PM Matthias J. Sax 
> wrote:
> 
>> No worries. It's just good to know. It seems that some other people are
>> interested to drive this further. So we will just "reassign" it to them.
>>
>> Thanks for letting us know.
>>
>>
>> -Matthias
>>
>> On 6/20/18 2:51 PM, Jeyhun Karimov wrote:
>>> Hi Matthias, all,
>>>
>>> Currently, I am not able to complete this KIP. Please accept my
>>> apologies for that.
>>>
>>>
>>> Cheers,
>>> Jeyhun
>>>
>>> On Mon, Jun 11, 2018 at 2:25 AM Matthias J. Sax >> > wrote:
>>>
>>> What is the status of this KIP?
>>>
>>> -Matthias
>>>
>>>
>>> On 2/13/18 1:43 PM, Matthias J. Sax wrote:
>>> > Is there any update for this KIP?
>>> >
>>> >
>>> > -Matthias
>>> >
>>> > On 12/4/17 2:08 PM, Matthias J. Sax wrote:
>>> >> Jeyhun,
>>> >>
>>> >> thanks for updating the KIP.
>>> >>
>>> >> I am wondering if you intend to add a new class `Produced`? There
>> is
>>> >> already `org.apache.kafka.streams.kstream.Produced`. So if we
>> want to
>>> >> add a new class, it must have a different name -- or we might be
>>> able to
>>> >> merge both into one?
>>> >>
>>> >> Also, for the KStream overlaods of `through()` and `to()`, can
>>> you add
>>> >> the different behavior using different overloads? It's not clear
>> from
>>> >> the KIP what the semantics are.
>>> >>
>>> >>
>>> >> -Matthias
>>> >>
>>> >> On 11/17/17 3:27 PM, Jeyhun Karimov wrote:
>>> >>> Hi,
>>> >>>
>>> >>> Thanks for your comments. I agree with Matthias partially.
>>> >>> I think we should relax some requirements related with to() and
>>> through()
>>> >>> methods.
>>> >>> IMHO, Produced class can cover (existing/to be created) topic
>>> information,
>>> >>> and which will ease our effort:
>>> >>>
>>> >>> KStream.to(Produced topicInfo)
>>> >>> KStream.through(Produced topicInfo)
>>> >>>
>>> >>> This will decrease the number of overloads but we will need to
>>> deprecate
>>> >>> the existing to() and through() methods, perhaps.
>>> >>> I updated the KIP accordingly.
>>> >>>
>>> >>>
>>> >>> Cheers,
>>> >>> Jeyhun
>>> >>>
>>> >>> On Thu, Nov 16, 2017 at 10:21 PM Matthias J. Sax
>>> mailto:matth...@confluent.io>>
>>> >>> wrote:
>>> >>>
>>>  @Jan:
>>> 
>>>  The `Produced` class was introduced in 1.0 to specify key and
>> valud
>>>  Serdes (and partitioner) if data is written into a topic.
>>> 
>>>  Old API:
>>> 
>>>  KStream#to("topic", keySerde, valueSerde);
>>> 
>>>  New API:
>>> 
>>>  KStream#to("topic", Produced.with(keySerde, valueSerde));
>>> 
>>> 
>>>  This allows to reduce the number of overloads for `to()` (and
>>>  `through()` that follows the same pattern) -- the second
>>> parameter is
>>>  used to cover all different variations of option parameters
>>> users can
>>>  specify, while we only have 2 overload for `to()` itself.
>>> 
>>>  What is still unclear to me it, what you mean by this topic
>> prefix
>>>  thing? Either a user cares about the topic name and thus, must
>>> create
>>>  and manage it manually. Or the user does not care, and Streams
>>> create
>>>  it. How would this prefix idea fit in here?
>>> 
>>> 
>>> 
>>>  @Guozhang:
>>> 
>>>  My idea was to extend `Produced` with the hint we want to give
>> for
>>>  creating internal topic and pass a optional `Produced`
>>> parameter. There
>>>  are multiple things we can do here:
>>> 
>>>  1) stream.through(null, Produced...).groupBy().aggregate()
>>>  -> just allow for `null` topic name indicating that Streams
>> should
>>>  create an internal topic
>>> 
>>>  2) stream.through(Produced...).groupBy().aggregate()
>>>  -> add one overload taking an mandatory `Produced`
>>> 
>>>  We use `Serialized` to picky back the information
>>> 
>>>  3) stream.groupBy(Serialized...).aggregate()
>>>  and 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Lei Chen]

Hi,

Just want to know is anyone actively working on this and also KAFKA-4835
? Seems like the JIRA has
been inactive for couple months. We want this feature and would like to
move it forward if no one else is working on it.

Lei

On Wed, Jun 20, 2018 at 7:27 PM Matthias J. Sax 
wrote:

> No worries. It's just good to know. It seems that some other people are
> interested to drive this further. So we will just "reassign" it to them.
>
> Thanks for letting us know.
>
>
> -Matthias
>
> On 6/20/18 2:51 PM, Jeyhun Karimov wrote:
> > Hi Matthias, all,
> >
> > Currently, I am not able to complete this KIP. Please accept my
> > apologies for that.
> >
> >
> > Cheers,
> > Jeyhun
> >
> > On Mon, Jun 11, 2018 at 2:25 AM Matthias J. Sax  > > wrote:
> >
> > What is the status of this KIP?
> >
> > -Matthias
> >
> >
> > On 2/13/18 1:43 PM, Matthias J. Sax wrote:
> > > Is there any update for this KIP?
> > >
> > >
> > > -Matthias
> > >
> > > On 12/4/17 2:08 PM, Matthias J. Sax wrote:
> > >> Jeyhun,
> > >>
> > >> thanks for updating the KIP.
> > >>
> > >> I am wondering if you intend to add a new class `Produced`? There
> is
> > >> already `org.apache.kafka.streams.kstream.Produced`. So if we
> want to
> > >> add a new class, it must have a different name -- or we might be
> > able to
> > >> merge both into one?
> > >>
> > >> Also, for the KStream overlaods of `through()` and `to()`, can
> > you add
> > >> the different behavior using different overloads? It's not clear
> from
> > >> the KIP what the semantics are.
> > >>
> > >>
> > >> -Matthias
> > >>
> > >> On 11/17/17 3:27 PM, Jeyhun Karimov wrote:
> > >>> Hi,
> > >>>
> > >>> Thanks for your comments. I agree with Matthias partially.
> > >>> I think we should relax some requirements related with to() and
> > through()
> > >>> methods.
> > >>> IMHO, Produced class can cover (existing/to be created) topic
> > information,
> > >>> and which will ease our effort:
> > >>>
> > >>> KStream.to(Produced topicInfo)
> > >>> KStream.through(Produced topicInfo)
> > >>>
> > >>> This will decrease the number of overloads but we will need to
> > deprecate
> > >>> the existing to() and through() methods, perhaps.
> > >>> I updated the KIP accordingly.
> > >>>
> > >>>
> > >>> Cheers,
> > >>> Jeyhun
> > >>>
> > >>> On Thu, Nov 16, 2017 at 10:21 PM Matthias J. Sax
> > mailto:matth...@confluent.io>>
> > >>> wrote:
> > >>>
> >  @Jan:
> > 
> >  The `Produced` class was introduced in 1.0 to specify key and
> valud
> >  Serdes (and partitioner) if data is written into a topic.
> > 
> >  Old API:
> > 
> >  KStream#to("topic", keySerde, valueSerde);
> > 
> >  New API:
> > 
> >  KStream#to("topic", Produced.with(keySerde, valueSerde));
> > 
> > 
> >  This allows to reduce the number of overloads for `to()` (and
> >  `through()` that follows the same pattern) -- the second
> > parameter is
> >  used to cover all different variations of option parameters
> > users can
> >  specify, while we only have 2 overload for `to()` itself.
> > 
> >  What is still unclear to me it, what you mean by this topic
> prefix
> >  thing? Either a user cares about the topic name and thus, must
> > create
> >  and manage it manually. Or the user does not care, and Streams
> > create
> >  it. How would this prefix idea fit in here?
> > 
> > 
> > 
> >  @Guozhang:
> > 
> >  My idea was to extend `Produced` with the hint we want to give
> for
> >  creating internal topic and pass a optional `Produced`
> > parameter. There
> >  are multiple things we can do here:
> > 
> >  1) stream.through(null, Produced...).groupBy().aggregate()
> >  -> just allow for `null` topic name indicating that Streams
> should
> >  create an internal topic
> > 
> >  2) stream.through(Produced...).groupBy().aggregate()
> >  -> add one overload taking an mandatory `Produced`
> > 
> >  We use `Serialized` to picky back the information
> > 
> >  3) stream.groupBy(Serialized...).aggregate()
> >  and stream.groupByKey(Serialized...).aggregate()
> >  -> we don't need new top level overloads
> > 
> > 
> >  There are different trade-offs for those alternatives and maybe
> > there
> >  are other ways to change the API. It's just to push the
> > discussion further.
> > 
> > 
> >  -Matthias
> > 
> >  On 11/12/17 1:22 PM, Jan 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Matthias J. Sax]

No worries. It's just good to know. It seems that some other people are
interested to drive this further. So we will just "reassign" it to them.

Thanks for letting us know.


-Matthias

On 6/20/18 2:51 PM, Jeyhun Karimov wrote:
> Hi Matthias, all,
> 
> Currently, I am not able to complete this KIP. Please accept my
> apologies for that. 
> 
> 
> Cheers,
> Jeyhun
> 
> On Mon, Jun 11, 2018 at 2:25 AM Matthias J. Sax  > wrote:
> 
> What is the status of this KIP?
> 
> -Matthias
> 
> 
> On 2/13/18 1:43 PM, Matthias J. Sax wrote:
> > Is there any update for this KIP?
> >
> >
> > -Matthias
> >
> > On 12/4/17 2:08 PM, Matthias J. Sax wrote:
> >> Jeyhun,
> >>
> >> thanks for updating the KIP.
> >>
> >> I am wondering if you intend to add a new class `Produced`? There is
> >> already `org.apache.kafka.streams.kstream.Produced`. So if we want to
> >> add a new class, it must have a different name -- or we might be
> able to
> >> merge both into one?
> >>
> >> Also, for the KStream overlaods of `through()` and `to()`, can
> you add
> >> the different behavior using different overloads? It's not clear from
> >> the KIP what the semantics are.
> >>
> >>
> >> -Matthias
> >>
> >> On 11/17/17 3:27 PM, Jeyhun Karimov wrote:
> >>> Hi,
> >>>
> >>> Thanks for your comments. I agree with Matthias partially.
> >>> I think we should relax some requirements related with to() and
> through()
> >>> methods.
> >>> IMHO, Produced class can cover (existing/to be created) topic
> information,
> >>> and which will ease our effort:
> >>>
> >>> KStream.to(Produced topicInfo)
> >>> KStream.through(Produced topicInfo)
> >>>
> >>> This will decrease the number of overloads but we will need to
> deprecate
> >>> the existing to() and through() methods, perhaps.
> >>> I updated the KIP accordingly.
> >>>
> >>>
> >>> Cheers,
> >>> Jeyhun
> >>>
> >>> On Thu, Nov 16, 2017 at 10:21 PM Matthias J. Sax
> mailto:matth...@confluent.io>>
> >>> wrote:
> >>>
>  @Jan:
> 
>  The `Produced` class was introduced in 1.0 to specify key and valud
>  Serdes (and partitioner) if data is written into a topic.
> 
>  Old API:
> 
>  KStream#to("topic", keySerde, valueSerde);
> 
>  New API:
> 
>  KStream#to("topic", Produced.with(keySerde, valueSerde));
> 
> 
>  This allows to reduce the number of overloads for `to()` (and
>  `through()` that follows the same pattern) -- the second
> parameter is
>  used to cover all different variations of option parameters
> users can
>  specify, while we only have 2 overload for `to()` itself.
> 
>  What is still unclear to me it, what you mean by this topic prefix
>  thing? Either a user cares about the topic name and thus, must
> create
>  and manage it manually. Or the user does not care, and Streams
> create
>  it. How would this prefix idea fit in here?
> 
> 
> 
>  @Guozhang:
> 
>  My idea was to extend `Produced` with the hint we want to give for
>  creating internal topic and pass a optional `Produced`
> parameter. There
>  are multiple things we can do here:
> 
>  1) stream.through(null, Produced...).groupBy().aggregate()
>  -> just allow for `null` topic name indicating that Streams should
>  create an internal topic
> 
>  2) stream.through(Produced...).groupBy().aggregate()
>  -> add one overload taking an mandatory `Produced`
> 
>  We use `Serialized` to picky back the information
> 
>  3) stream.groupBy(Serialized...).aggregate()
>  and stream.groupByKey(Serialized...).aggregate()
>  -> we don't need new top level overloads
> 
> 
>  There are different trade-offs for those alternatives and maybe
> there
>  are other ways to change the API. It's just to push the
> discussion further.
> 
> 
>  -Matthias
> 
>  On 11/12/17 1:22 PM, Jan Filipiak wrote:
> > Hi Gouzhang,
> >
> > this felt like these questions are supposed to be answered by me.
> > I do not understand the first one. I don't understand why the user
> > shouldn't be able to specify a suffix for the topic name.
> >
> >  For the third question I am not 100% familiar if the Produced
> class
> > came to existence
> > at all. I remember proposing it somewhere in our redo DSL
> discussion that
> > I dropped out of later. Finally any call that does:
> >
> > 1. create the internal 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Jeyhun Karimov]

Hi Matthias, all,

Currently, I am not able to complete this KIP. Please accept my apologies
for that.


Cheers,
Jeyhun

On Mon, Jun 11, 2018 at 2:25 AM Matthias J. Sax 
wrote:

> What is the status of this KIP?
>
> -Matthias
>
>
> On 2/13/18 1:43 PM, Matthias J. Sax wrote:
> > Is there any update for this KIP?
> >
> >
> > -Matthias
> >
> > On 12/4/17 2:08 PM, Matthias J. Sax wrote:
> >> Jeyhun,
> >>
> >> thanks for updating the KIP.
> >>
> >> I am wondering if you intend to add a new class `Produced`? There is
> >> already `org.apache.kafka.streams.kstream.Produced`. So if we want to
> >> add a new class, it must have a different name -- or we might be able to
> >> merge both into one?
> >>
> >> Also, for the KStream overlaods of `through()` and `to()`, can you add
> >> the different behavior using different overloads? It's not clear from
> >> the KIP what the semantics are.
> >>
> >>
> >> -Matthias
> >>
> >> On 11/17/17 3:27 PM, Jeyhun Karimov wrote:
> >>> Hi,
> >>>
> >>> Thanks for your comments. I agree with Matthias partially.
> >>> I think we should relax some requirements related with to() and
> through()
> >>> methods.
> >>> IMHO, Produced class can cover (existing/to be created) topic
> information,
> >>> and which will ease our effort:
> >>>
> >>> KStream.to(Produced topicInfo)
> >>> KStream.through(Produced topicInfo)
> >>>
> >>> This will decrease the number of overloads but we will need to
> deprecate
> >>> the existing to() and through() methods, perhaps.
> >>> I updated the KIP accordingly.
> >>>
> >>>
> >>> Cheers,
> >>> Jeyhun
> >>>
> >>> On Thu, Nov 16, 2017 at 10:21 PM Matthias J. Sax <
> matth...@confluent.io>
> >>> wrote:
> >>>
>  @Jan:
> 
>  The `Produced` class was introduced in 1.0 to specify key and valud
>  Serdes (and partitioner) if data is written into a topic.
> 
>  Old API:
> 
>  KStream#to("topic", keySerde, valueSerde);
> 
>  New API:
> 
>  KStream#to("topic", Produced.with(keySerde, valueSerde));
> 
> 
>  This allows to reduce the number of overloads for `to()` (and
>  `through()` that follows the same pattern) -- the second parameter is
>  used to cover all different variations of option parameters users can
>  specify, while we only have 2 overload for `to()` itself.
> 
>  What is still unclear to me it, what you mean by this topic prefix
>  thing? Either a user cares about the topic name and thus, must create
>  and manage it manually. Or the user does not care, and Streams create
>  it. How would this prefix idea fit in here?
> 
> 
> 
>  @Guozhang:
> 
>  My idea was to extend `Produced` with the hint we want to give for
>  creating internal topic and pass a optional `Produced` parameter.
> There
>  are multiple things we can do here:
> 
>  1) stream.through(null, Produced...).groupBy().aggregate()
>  -> just allow for `null` topic name indicating that Streams should
>  create an internal topic
> 
>  2) stream.through(Produced...).groupBy().aggregate()
>  -> add one overload taking an mandatory `Produced`
> 
>  We use `Serialized` to picky back the information
> 
>  3) stream.groupBy(Serialized...).aggregate()
>  and stream.groupByKey(Serialized...).aggregate()
>  -> we don't need new top level overloads
> 
> 
>  There are different trade-offs for those alternatives and maybe there
>  are other ways to change the API. It's just to push the discussion
> further.
> 
> 
>  -Matthias
> 
>  On 11/12/17 1:22 PM, Jan Filipiak wrote:
> > Hi Gouzhang,
> >
> > this felt like these questions are supposed to be answered by me.
> > I do not understand the first one. I don't understand why the user
> > shouldn't be able to specify a suffix for the topic name.
> >
> >  For the third question I am not 100% familiar if the Produced class
> > came to existence
> > at all. I remember proposing it somewhere in our redo DSL discussion
> that
> > I dropped out of later. Finally any call that does:
> >
> > 1. create the internal topic
> > 2. register sink
> > 3. register source
> >
> > will always get the work done. If we have a Produced like class.
> putting
> > all the parameters
> > in there make sense. (Partitioner, serde, PartitionHint, internal,
> name
> > ... )
> >
> > Hope this helps?
> >
> >
> > On 10.11.2017 07:54, Guozhang Wang wrote:
> >> A few clarification questions on the proposal details.
> >>
> >> 1. API: although the repartition only happens at the final stateful
> >> operations like agg / join, the repartition flag info was actually
>  passed
> >> from an earlier operator like map / groupBy. So what should be the
> new
> >> API
> >> look like? For example, if we do
> >>
> >> stream.groupBy().through("topic-name", 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Matthias J. Sax]

What is the status of this KIP?

-Matthias


On 2/13/18 1:43 PM, Matthias J. Sax wrote:
> Is there any update for this KIP?
> 
> 
> -Matthias
> 
> On 12/4/17 2:08 PM, Matthias J. Sax wrote:
>> Jeyhun,
>>
>> thanks for updating the KIP.
>>
>> I am wondering if you intend to add a new class `Produced`? There is
>> already `org.apache.kafka.streams.kstream.Produced`. So if we want to
>> add a new class, it must have a different name -- or we might be able to
>> merge both into one?
>>
>> Also, for the KStream overlaods of `through()` and `to()`, can you add
>> the different behavior using different overloads? It's not clear from
>> the KIP what the semantics are.
>>
>>
>> -Matthias
>>
>> On 11/17/17 3:27 PM, Jeyhun Karimov wrote:
>>> Hi,
>>>
>>> Thanks for your comments. I agree with Matthias partially.
>>> I think we should relax some requirements related with to() and through()
>>> methods.
>>> IMHO, Produced class can cover (existing/to be created) topic information,
>>> and which will ease our effort:
>>>
>>> KStream.to(Produced topicInfo)
>>> KStream.through(Produced topicInfo)
>>>
>>> This will decrease the number of overloads but we will need to deprecate
>>> the existing to() and through() methods, perhaps.
>>> I updated the KIP accordingly.
>>>
>>>
>>> Cheers,
>>> Jeyhun
>>>
>>> On Thu, Nov 16, 2017 at 10:21 PM Matthias J. Sax 
>>> wrote:
>>>
 @Jan:

 The `Produced` class was introduced in 1.0 to specify key and valud
 Serdes (and partitioner) if data is written into a topic.

 Old API:

 KStream#to("topic", keySerde, valueSerde);

 New API:

 KStream#to("topic", Produced.with(keySerde, valueSerde));


 This allows to reduce the number of overloads for `to()` (and
 `through()` that follows the same pattern) -- the second parameter is
 used to cover all different variations of option parameters users can
 specify, while we only have 2 overload for `to()` itself.

 What is still unclear to me it, what you mean by this topic prefix
 thing? Either a user cares about the topic name and thus, must create
 and manage it manually. Or the user does not care, and Streams create
 it. How would this prefix idea fit in here?



 @Guozhang:

 My idea was to extend `Produced` with the hint we want to give for
 creating internal topic and pass a optional `Produced` parameter. There
 are multiple things we can do here:

 1) stream.through(null, Produced...).groupBy().aggregate()
 -> just allow for `null` topic name indicating that Streams should
 create an internal topic

 2) stream.through(Produced...).groupBy().aggregate()
 -> add one overload taking an mandatory `Produced`

 We use `Serialized` to picky back the information

 3) stream.groupBy(Serialized...).aggregate()
 and stream.groupByKey(Serialized...).aggregate()
 -> we don't need new top level overloads


 There are different trade-offs for those alternatives and maybe there
 are other ways to change the API. It's just to push the discussion further.


 -Matthias

 On 11/12/17 1:22 PM, Jan Filipiak wrote:
> Hi Gouzhang,
>
> this felt like these questions are supposed to be answered by me.
> I do not understand the first one. I don't understand why the user
> shouldn't be able to specify a suffix for the topic name.
>
>  For the third question I am not 100% familiar if the Produced class
> came to existence
> at all. I remember proposing it somewhere in our redo DSL discussion that
> I dropped out of later. Finally any call that does:
>
> 1. create the internal topic
> 2. register sink
> 3. register source
>
> will always get the work done. If we have a Produced like class. putting
> all the parameters
> in there make sense. (Partitioner, serde, PartitionHint, internal, name
> ... )
>
> Hope this helps?
>
>
> On 10.11.2017 07:54, Guozhang Wang wrote:
>> A few clarification questions on the proposal details.
>>
>> 1. API: although the repartition only happens at the final stateful
>> operations like agg / join, the repartition flag info was actually
 passed
>> from an earlier operator like map / groupBy. So what should be the new
>> API
>> look like? For example, if we do
>>
>> stream.groupBy().through("topic-name", Produced..).aggregate
>>
>> This would be add a bunch of APIs to GroupedKStream/KTable
>>
>> 2. Semantics: as Matthias mentioned, today any topics defined in
>> "through()" call is considered a user topic, and hence users are
>> responsible for managing them, including the topic name. For this KIP's
>> purpose, though, users would not care about the topic name. I.e. as a
>> user
>> I still want to make it be an internal topic so that I do not need to
>> worry
>> 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Matthias J. Sax]

Is there any update for this KIP?


-Matthias

On 12/4/17 2:08 PM, Matthias J. Sax wrote:
> Jeyhun,
> 
> thanks for updating the KIP.
> 
> I am wondering if you intend to add a new class `Produced`? There is
> already `org.apache.kafka.streams.kstream.Produced`. So if we want to
> add a new class, it must have a different name -- or we might be able to
> merge both into one?
> 
> Also, for the KStream overlaods of `through()` and `to()`, can you add
> the different behavior using different overloads? It's not clear from
> the KIP what the semantics are.
> 
> 
> -Matthias
> 
> On 11/17/17 3:27 PM, Jeyhun Karimov wrote:
>> Hi,
>>
>> Thanks for your comments. I agree with Matthias partially.
>> I think we should relax some requirements related with to() and through()
>> methods.
>> IMHO, Produced class can cover (existing/to be created) topic information,
>> and which will ease our effort:
>>
>> KStream.to(Produced topicInfo)
>> KStream.through(Produced topicInfo)
>>
>> This will decrease the number of overloads but we will need to deprecate
>> the existing to() and through() methods, perhaps.
>> I updated the KIP accordingly.
>>
>>
>> Cheers,
>> Jeyhun
>>
>> On Thu, Nov 16, 2017 at 10:21 PM Matthias J. Sax 
>> wrote:
>>
>>> @Jan:
>>>
>>> The `Produced` class was introduced in 1.0 to specify key and valud
>>> Serdes (and partitioner) if data is written into a topic.
>>>
>>> Old API:
>>>
>>> KStream#to("topic", keySerde, valueSerde);
>>>
>>> New API:
>>>
>>> KStream#to("topic", Produced.with(keySerde, valueSerde));
>>>
>>>
>>> This allows to reduce the number of overloads for `to()` (and
>>> `through()` that follows the same pattern) -- the second parameter is
>>> used to cover all different variations of option parameters users can
>>> specify, while we only have 2 overload for `to()` itself.
>>>
>>> What is still unclear to me it, what you mean by this topic prefix
>>> thing? Either a user cares about the topic name and thus, must create
>>> and manage it manually. Or the user does not care, and Streams create
>>> it. How would this prefix idea fit in here?
>>>
>>>
>>>
>>> @Guozhang:
>>>
>>> My idea was to extend `Produced` with the hint we want to give for
>>> creating internal topic and pass a optional `Produced` parameter. There
>>> are multiple things we can do here:
>>>
>>> 1) stream.through(null, Produced...).groupBy().aggregate()
>>> -> just allow for `null` topic name indicating that Streams should
>>> create an internal topic
>>>
>>> 2) stream.through(Produced...).groupBy().aggregate()
>>> -> add one overload taking an mandatory `Produced`
>>>
>>> We use `Serialized` to picky back the information
>>>
>>> 3) stream.groupBy(Serialized...).aggregate()
>>> and stream.groupByKey(Serialized...).aggregate()
>>> -> we don't need new top level overloads
>>>
>>>
>>> There are different trade-offs for those alternatives and maybe there
>>> are other ways to change the API. It's just to push the discussion further.
>>>
>>>
>>> -Matthias
>>>
>>> On 11/12/17 1:22 PM, Jan Filipiak wrote:
 Hi Gouzhang,

 this felt like these questions are supposed to be answered by me.
 I do not understand the first one. I don't understand why the user
 shouldn't be able to specify a suffix for the topic name.

  For the third question I am not 100% familiar if the Produced class
 came to existence
 at all. I remember proposing it somewhere in our redo DSL discussion that
 I dropped out of later. Finally any call that does:

 1. create the internal topic
 2. register sink
 3. register source

 will always get the work done. If we have a Produced like class. putting
 all the parameters
 in there make sense. (Partitioner, serde, PartitionHint, internal, name
 ... )

 Hope this helps?


 On 10.11.2017 07:54, Guozhang Wang wrote:
> A few clarification questions on the proposal details.
>
> 1. API: although the repartition only happens at the final stateful
> operations like agg / join, the repartition flag info was actually
>>> passed
> from an earlier operator like map / groupBy. So what should be the new
> API
> look like? For example, if we do
>
> stream.groupBy().through("topic-name", Produced..).aggregate
>
> This would be add a bunch of APIs to GroupedKStream/KTable
>
> 2. Semantics: as Matthias mentioned, today any topics defined in
> "through()" call is considered a user topic, and hence users are
> responsible for managing them, including the topic name. For this KIP's
> purpose, though, users would not care about the topic name. I.e. as a
> user
> I still want to make it be an internal topic so that I do not need to
> worry
> about it at all, but only specify num.partitions.
>
> 3. Details: in Produced we do not have specs for specifying the
> num.partitions or should we repartition or not. So it is still not
> 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Matthias J. Sax]

Jeyhun,

thanks for updating the KIP.

I am wondering if you intend to add a new class `Produced`? There is
already `org.apache.kafka.streams.kstream.Produced`. So if we want to
add a new class, it must have a different name -- or we might be able to
merge both into one?

Also, for the KStream overlaods of `through()` and `to()`, can you add
the different behavior using different overloads? It's not clear from
the KIP what the semantics are.


-Matthias

On 11/17/17 3:27 PM, Jeyhun Karimov wrote:
> Hi,
> 
> Thanks for your comments. I agree with Matthias partially.
> I think we should relax some requirements related with to() and through()
> methods.
> IMHO, Produced class can cover (existing/to be created) topic information,
> and which will ease our effort:
> 
> KStream.to(Produced topicInfo)
> KStream.through(Produced topicInfo)
> 
> This will decrease the number of overloads but we will need to deprecate
> the existing to() and through() methods, perhaps.
> I updated the KIP accordingly.
> 
> 
> Cheers,
> Jeyhun
> 
> On Thu, Nov 16, 2017 at 10:21 PM Matthias J. Sax 
> wrote:
> 
>> @Jan:
>>
>> The `Produced` class was introduced in 1.0 to specify key and valud
>> Serdes (and partitioner) if data is written into a topic.
>>
>> Old API:
>>
>> KStream#to("topic", keySerde, valueSerde);
>>
>> New API:
>>
>> KStream#to("topic", Produced.with(keySerde, valueSerde));
>>
>>
>> This allows to reduce the number of overloads for `to()` (and
>> `through()` that follows the same pattern) -- the second parameter is
>> used to cover all different variations of option parameters users can
>> specify, while we only have 2 overload for `to()` itself.
>>
>> What is still unclear to me it, what you mean by this topic prefix
>> thing? Either a user cares about the topic name and thus, must create
>> and manage it manually. Or the user does not care, and Streams create
>> it. How would this prefix idea fit in here?
>>
>>
>>
>> @Guozhang:
>>
>> My idea was to extend `Produced` with the hint we want to give for
>> creating internal topic and pass a optional `Produced` parameter. There
>> are multiple things we can do here:
>>
>> 1) stream.through(null, Produced...).groupBy().aggregate()
>> -> just allow for `null` topic name indicating that Streams should
>> create an internal topic
>>
>> 2) stream.through(Produced...).groupBy().aggregate()
>> -> add one overload taking an mandatory `Produced`
>>
>> We use `Serialized` to picky back the information
>>
>> 3) stream.groupBy(Serialized...).aggregate()
>> and stream.groupByKey(Serialized...).aggregate()
>> -> we don't need new top level overloads
>>
>>
>> There are different trade-offs for those alternatives and maybe there
>> are other ways to change the API. It's just to push the discussion further.
>>
>>
>> -Matthias
>>
>> On 11/12/17 1:22 PM, Jan Filipiak wrote:
>>> Hi Gouzhang,
>>>
>>> this felt like these questions are supposed to be answered by me.
>>> I do not understand the first one. I don't understand why the user
>>> shouldn't be able to specify a suffix for the topic name.
>>>
>>>  For the third question I am not 100% familiar if the Produced class
>>> came to existence
>>> at all. I remember proposing it somewhere in our redo DSL discussion that
>>> I dropped out of later. Finally any call that does:
>>>
>>> 1. create the internal topic
>>> 2. register sink
>>> 3. register source
>>>
>>> will always get the work done. If we have a Produced like class. putting
>>> all the parameters
>>> in there make sense. (Partitioner, serde, PartitionHint, internal, name
>>> ... )
>>>
>>> Hope this helps?
>>>
>>>
>>> On 10.11.2017 07:54, Guozhang Wang wrote:
 A few clarification questions on the proposal details.

 1. API: although the repartition only happens at the final stateful
 operations like agg / join, the repartition flag info was actually
>> passed
 from an earlier operator like map / groupBy. So what should be the new
 API
 look like? For example, if we do

 stream.groupBy().through("topic-name", Produced..).aggregate

 This would be add a bunch of APIs to GroupedKStream/KTable

 2. Semantics: as Matthias mentioned, today any topics defined in
 "through()" call is considered a user topic, and hence users are
 responsible for managing them, including the topic name. For this KIP's
 purpose, though, users would not care about the topic name. I.e. as a
 user
 I still want to make it be an internal topic so that I do not need to
 worry
 about it at all, but only specify num.partitions.

 3. Details: in Produced we do not have specs for specifying the
 num.partitions or should we repartition or not. So it is still not
 clear to
 me how we would make use of that to achieve what's in the old
 proposal's RepartitionHint class.



 Guozhang


 On Mon, Nov 6, 2017 at 1:21 PM, Ted Yu  wrote:

> bq. 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Jeyhun Karimov]

Hi,

Thanks for your comments. I agree with Matthias partially.
I think we should relax some requirements related with to() and through()
methods.
IMHO, Produced class can cover (existing/to be created) topic information,
and which will ease our effort:

KStream.to(Produced topicInfo)
KStream.through(Produced topicInfo)

This will decrease the number of overloads but we will need to deprecate
the existing to() and through() methods, perhaps.
I updated the KIP accordingly.


Cheers,
Jeyhun

On Thu, Nov 16, 2017 at 10:21 PM Matthias J. Sax 
wrote:

> @Jan:
>
> The `Produced` class was introduced in 1.0 to specify key and valud
> Serdes (and partitioner) if data is written into a topic.
>
> Old API:
>
> KStream#to("topic", keySerde, valueSerde);
>
> New API:
>
> KStream#to("topic", Produced.with(keySerde, valueSerde));
>
>
> This allows to reduce the number of overloads for `to()` (and
> `through()` that follows the same pattern) -- the second parameter is
> used to cover all different variations of option parameters users can
> specify, while we only have 2 overload for `to()` itself.
>
> What is still unclear to me it, what you mean by this topic prefix
> thing? Either a user cares about the topic name and thus, must create
> and manage it manually. Or the user does not care, and Streams create
> it. How would this prefix idea fit in here?
>
>
>
> @Guozhang:
>
> My idea was to extend `Produced` with the hint we want to give for
> creating internal topic and pass a optional `Produced` parameter. There
> are multiple things we can do here:
>
> 1) stream.through(null, Produced...).groupBy().aggregate()
> -> just allow for `null` topic name indicating that Streams should
> create an internal topic
>
> 2) stream.through(Produced...).groupBy().aggregate()
> -> add one overload taking an mandatory `Produced`
>
> We use `Serialized` to picky back the information
>
> 3) stream.groupBy(Serialized...).aggregate()
> and stream.groupByKey(Serialized...).aggregate()
> -> we don't need new top level overloads
>
>
> There are different trade-offs for those alternatives and maybe there
> are other ways to change the API. It's just to push the discussion further.
>
>
> -Matthias
>
> On 11/12/17 1:22 PM, Jan Filipiak wrote:
> > Hi Gouzhang,
> >
> > this felt like these questions are supposed to be answered by me.
> > I do not understand the first one. I don't understand why the user
> > shouldn't be able to specify a suffix for the topic name.
> >
> >  For the third question I am not 100% familiar if the Produced class
> > came to existence
> > at all. I remember proposing it somewhere in our redo DSL discussion that
> > I dropped out of later. Finally any call that does:
> >
> > 1. create the internal topic
> > 2. register sink
> > 3. register source
> >
> > will always get the work done. If we have a Produced like class. putting
> > all the parameters
> > in there make sense. (Partitioner, serde, PartitionHint, internal, name
> > ... )
> >
> > Hope this helps?
> >
> >
> > On 10.11.2017 07:54, Guozhang Wang wrote:
> >> A few clarification questions on the proposal details.
> >>
> >> 1. API: although the repartition only happens at the final stateful
> >> operations like agg / join, the repartition flag info was actually
> passed
> >> from an earlier operator like map / groupBy. So what should be the new
> >> API
> >> look like? For example, if we do
> >>
> >> stream.groupBy().through("topic-name", Produced..).aggregate
> >>
> >> This would be add a bunch of APIs to GroupedKStream/KTable
> >>
> >> 2. Semantics: as Matthias mentioned, today any topics defined in
> >> "through()" call is considered a user topic, and hence users are
> >> responsible for managing them, including the topic name. For this KIP's
> >> purpose, though, users would not care about the topic name. I.e. as a
> >> user
> >> I still want to make it be an internal topic so that I do not need to
> >> worry
> >> about it at all, but only specify num.partitions.
> >>
> >> 3. Details: in Produced we do not have specs for specifying the
> >> num.partitions or should we repartition or not. So it is still not
> >> clear to
> >> me how we would make use of that to achieve what's in the old
> >> proposal's RepartitionHint class.
> >>
> >>
> >>
> >> Guozhang
> >>
> >>
> >> On Mon, Nov 6, 2017 at 1:21 PM, Ted Yu  wrote:
> >>
> >>> bq. enlarge the score of through()
> >>>
> >>> I guess you meant scope.
> >>>
> >>> On Mon, Nov 6, 2017 at 1:15 PM, Jeyhun Karimov 
> >>> wrote:
> >>>
>  Hi,
> 
>  Sorry for the late reply. I am convinced that we should enlarge the
>  score
>  of through() (add more overloads) instead of introducing a separate
> set
> >>> of
>  overloads to other methods.
>  I will update the KIP soon based on the discussion and inform.
> 
> 
>  Cheers,
>  Jeyhun
> 
>  On Mon, Nov 6, 2017 at 9:18 PM Jan Filipiak  >

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Matthias J. Sax]

@Jan:

The `Produced` class was introduced in 1.0 to specify key and valud
Serdes (and partitioner) if data is written into a topic.

Old API:

KStream#to("topic", keySerde, valueSerde);

New API:

KStream#to("topic", Produced.with(keySerde, valueSerde));


This allows to reduce the number of overloads for `to()` (and
`through()` that follows the same pattern) -- the second parameter is
used to cover all different variations of option parameters users can
specify, while we only have 2 overload for `to()` itself.

What is still unclear to me it, what you mean by this topic prefix
thing? Either a user cares about the topic name and thus, must create
and manage it manually. Or the user does not care, and Streams create
it. How would this prefix idea fit in here?



@Guozhang:

My idea was to extend `Produced` with the hint we want to give for
creating internal topic and pass a optional `Produced` parameter. There
are multiple things we can do here:

1) stream.through(null, Produced...).groupBy().aggregate()
-> just allow for `null` topic name indicating that Streams should
create an internal topic

2) stream.through(Produced...).groupBy().aggregate()
-> add one overload taking an mandatory `Produced`

We use `Serialized` to picky back the information

3) stream.groupBy(Serialized...).aggregate()
and stream.groupByKey(Serialized...).aggregate()
-> we don't need new top level overloads


There are different trade-offs for those alternatives and maybe there
are other ways to change the API. It's just to push the discussion further.


-Matthias

On 11/12/17 1:22 PM, Jan Filipiak wrote:
> Hi Gouzhang,
> 
> this felt like these questions are supposed to be answered by me.
> I do not understand the first one. I don't understand why the user
> shouldn't be able to specify a suffix for the topic name.
> 
>  For the third question I am not 100% familiar if the Produced class
> came to existence
> at all. I remember proposing it somewhere in our redo DSL discussion that
> I dropped out of later. Finally any call that does:
> 
> 1. create the internal topic
> 2. register sink
> 3. register source
> 
> will always get the work done. If we have a Produced like class. putting
> all the parameters
> in there make sense. (Partitioner, serde, PartitionHint, internal, name
> ... )
> 
> Hope this helps?
> 
> 
> On 10.11.2017 07:54, Guozhang Wang wrote:
>> A few clarification questions on the proposal details.
>>
>> 1. API: although the repartition only happens at the final stateful
>> operations like agg / join, the repartition flag info was actually passed
>> from an earlier operator like map / groupBy. So what should be the new
>> API
>> look like? For example, if we do
>>
>> stream.groupBy().through("topic-name", Produced..).aggregate
>>
>> This would be add a bunch of APIs to GroupedKStream/KTable
>>
>> 2. Semantics: as Matthias mentioned, today any topics defined in
>> "through()" call is considered a user topic, and hence users are
>> responsible for managing them, including the topic name. For this KIP's
>> purpose, though, users would not care about the topic name. I.e. as a
>> user
>> I still want to make it be an internal topic so that I do not need to
>> worry
>> about it at all, but only specify num.partitions.
>>
>> 3. Details: in Produced we do not have specs for specifying the
>> num.partitions or should we repartition or not. So it is still not
>> clear to
>> me how we would make use of that to achieve what's in the old
>> proposal's RepartitionHint class.
>>
>>
>>
>> Guozhang
>>
>>
>> On Mon, Nov 6, 2017 at 1:21 PM, Ted Yu  wrote:
>>
>>> bq. enlarge the score of through()
>>>
>>> I guess you meant scope.
>>>
>>> On Mon, Nov 6, 2017 at 1:15 PM, Jeyhun Karimov 
>>> wrote:
>>>
 Hi,

 Sorry for the late reply. I am convinced that we should enlarge the
 score
 of through() (add more overloads) instead of introducing a separate set
>>> of
 overloads to other methods.
 I will update the KIP soon based on the discussion and inform.


 Cheers,
 Jeyhun

 On Mon, Nov 6, 2017 at 9:18 PM Jan Filipiak 
 wrote:

> Sorry for not beeing 100% up to date.
> Back then we had the discussion that when an operation puts a >Sink<
> into the topology, a >Produced<
> parameter is added. This produced parameter could have internal or
> external. If internal I think the name would still make
> a great suffix for the topic name
>
> Is this plan still around? Otherwise having the name as suffix is
> probably always good it can help the user quicker to identify hot
>>> topics
> that need more
> partitions if he has many of these internal repartitions
>
> Best Jan
>
>
> On 06.11.2017 20:13, Matthias J. Sax wrote:
>> I absolute agree with what you say. It's not a requirement to
>>> specify a
>> topic name -- and this was the idea -- if user does 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Jan Filipiak]

Hi Gouzhang,

this felt like these questions are supposed to be answered by me.
I do not understand the first one. I don't understand why the user
shouldn't be able to specify a suffix for the topic name.

 For the third question I am not 100% familiar if the Produced class 
came to existence

at all. I remember proposing it somewhere in our redo DSL discussion that
I dropped out of later. Finally any call that does:

1. create the internal topic
2. register sink
3. register source

will always get the work done. If we have a Produced like class. putting 
all the parameters
in there make sense. (Partitioner, serde, PartitionHint, internal, name 
... )


Hope this helps?


On 10.11.2017 07:54, Guozhang Wang wrote:

A few clarification questions on the proposal details.

1. API: although the repartition only happens at the final stateful
operations like agg / join, the repartition flag info was actually passed
from an earlier operator like map / groupBy. So what should be the new API
look like? For example, if we do

stream.groupBy().through("topic-name", Produced..).aggregate

This would be add a bunch of APIs to GroupedKStream/KTable

2. Semantics: as Matthias mentioned, today any topics defined in
"through()" call is considered a user topic, and hence users are
responsible for managing them, including the topic name. For this KIP's
purpose, though, users would not care about the topic name. I.e. as a user
I still want to make it be an internal topic so that I do not need to worry
about it at all, but only specify num.partitions.

3. Details: in Produced we do not have specs for specifying the
num.partitions or should we repartition or not. So it is still not clear to
me how we would make use of that to achieve what's in the old
proposal's RepartitionHint class.



Guozhang


On Mon, Nov 6, 2017 at 1:21 PM, Ted Yu  wrote:


bq. enlarge the score of through()

I guess you meant scope.

On Mon, Nov 6, 2017 at 1:15 PM, Jeyhun Karimov 
wrote:


Hi,

Sorry for the late reply. I am convinced that we should enlarge the score
of through() (add more overloads) instead of introducing a separate set

of

overloads to other methods.
I will update the KIP soon based on the discussion and inform.


Cheers,
Jeyhun

On Mon, Nov 6, 2017 at 9:18 PM Jan Filipiak 
wrote:


Sorry for not beeing 100% up to date.
Back then we had the discussion that when an operation puts a >Sink<
into the topology, a >Produced<
parameter is added. This produced parameter could have internal or
external. If internal I think the name would still make
a great suffix for the topic name

Is this plan still around? Otherwise having the name as suffix is
probably always good it can help the user quicker to identify hot

topics

that need more
partitions if he has many of these internal repartitions

Best Jan


On 06.11.2017 20:13, Matthias J. Sax wrote:

I absolute agree with what you say. It's not a requirement to

specify a

topic name -- and this was the idea -- if user does specify a name,

we

treat as is -- if users does not specify a name, Streams create an
internal topic.

The goal of the Jira is to allow a simplified way to control
repartitioning (atm, user needs to manually create a topic and use

via

through()).

Thus, the idea is to make the topic name parameter of through

optional.

It's of course just an idea. Happy do have a other API design. The

goal

was, to avoid to many new overloads.


Could you clarify exactly what you mean by keeping the current

distinction?

Current distinction is: user topics are created manually and user
specifies the name -- internal topics are created by Kafka Streams

and

an name is generated automatically.

-> through("user-topic")
-> through(TopicConfig.withNumberOfPartitions(5)) // Streams creates

an

internal topic


-Matthias


On 11/6/17 6:56 PM, Thomas Becker wrote:

Could you clarify exactly what you mean by keeping the current

distinction?

Actually, re-reading the KIP and JIRA, it's not clear that being

able

to specify a custom name is actually a requirement. If the goal is to
control repartitioning and tune parallelism, maybe we can just sidestep
this issue altogether by removing the ability to set a different name.

On Mon, 2017-11-06 at 16:51 +0100, Matthias J. Sax wrote:

That's a good point. In current design, we strictly distinguish

both.

For example, the reset tools deletes internal topics (starting with
prefix `-` and ending with either `-repartition` or
`-changelog`.

Thus, from my point of view, it would make sense to keep the current
distinction.

-Matthias

On 11/6/17 4:45 PM, Thomas Becker wrote:


I think this sounds good as well. It's worth clarifying whether

topics

that are named by the user but created by streams are considered

"internal"

topics also.

On Sun, 2017-11-05 at 23:02 +0100, Matthias J. Sax wrote:

My idea was, to relax the requirement for through() that a topic

must

be

created 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Guozhang Wang]

A few clarification questions on the proposal details.

1. API: although the repartition only happens at the final stateful
operations like agg / join, the repartition flag info was actually passed
from an earlier operator like map / groupBy. So what should be the new API
look like? For example, if we do

stream.groupBy().through("topic-name", Produced..).aggregate

This would be add a bunch of APIs to GroupedKStream/KTable

2. Semantics: as Matthias mentioned, today any topics defined in
"through()" call is considered a user topic, and hence users are
responsible for managing them, including the topic name. For this KIP's
purpose, though, users would not care about the topic name. I.e. as a user
I still want to make it be an internal topic so that I do not need to worry
about it at all, but only specify num.partitions.

3. Details: in Produced we do not have specs for specifying the
num.partitions or should we repartition or not. So it is still not clear to
me how we would make use of that to achieve what's in the old
proposal's RepartitionHint class.



Guozhang


On Mon, Nov 6, 2017 at 1:21 PM, Ted Yu  wrote:

> bq. enlarge the score of through()
>
> I guess you meant scope.
>
> On Mon, Nov 6, 2017 at 1:15 PM, Jeyhun Karimov 
> wrote:
>
> > Hi,
> >
> > Sorry for the late reply. I am convinced that we should enlarge the score
> > of through() (add more overloads) instead of introducing a separate set
> of
> > overloads to other methods.
> > I will update the KIP soon based on the discussion and inform.
> >
> >
> > Cheers,
> > Jeyhun
> >
> > On Mon, Nov 6, 2017 at 9:18 PM Jan Filipiak 
> > wrote:
> >
> > > Sorry for not beeing 100% up to date.
> > > Back then we had the discussion that when an operation puts a >Sink<
> > > into the topology, a >Produced<
> > > parameter is added. This produced parameter could have internal or
> > > external. If internal I think the name would still make
> > > a great suffix for the topic name
> > >
> > > Is this plan still around? Otherwise having the name as suffix is
> > > probably always good it can help the user quicker to identify hot
> topics
> > > that need more
> > > partitions if he has many of these internal repartitions
> > >
> > > Best Jan
> > >
> > >
> > > On 06.11.2017 20:13, Matthias J. Sax wrote:
> > > > I absolute agree with what you say. It's not a requirement to
> specify a
> > > > topic name -- and this was the idea -- if user does specify a name,
> we
> > > > treat as is -- if users does not specify a name, Streams create an
> > > > internal topic.
> > > >
> > > > The goal of the Jira is to allow a simplified way to control
> > > > repartitioning (atm, user needs to manually create a topic and use
> via
> > > > through()).
> > > >
> > > > Thus, the idea is to make the topic name parameter of through
> optional.
> > > >
> > > > It's of course just an idea. Happy do have a other API design. The
> goal
> > > > was, to avoid to many new overloads.
> > > >
> > > >>> Could you clarify exactly what you mean by keeping the current
> > > distinction?
> > > > Current distinction is: user topics are created manually and user
> > > > specifies the name -- internal topics are created by Kafka Streams
> and
> > > > an name is generated automatically.
> > > >
> > > > -> through("user-topic")
> > > > -> through(TopicConfig.withNumberOfPartitions(5)) // Streams creates
> > an
> > > > internal topic
> > > >
> > > >
> > > > -Matthias
> > > >
> > > >
> > > > On 11/6/17 6:56 PM, Thomas Becker wrote:
> > > >> Could you clarify exactly what you mean by keeping the current
> > > distinction?
> > > >>
> > > >> Actually, re-reading the KIP and JIRA, it's not clear that being
> able
> > > to specify a custom name is actually a requirement. If the goal is to
> > > control repartitioning and tune parallelism, maybe we can just sidestep
> > > this issue altogether by removing the ability to set a different name.
> > > >>
> > > >> On Mon, 2017-11-06 at 16:51 +0100, Matthias J. Sax wrote:
> > > >>
> > > >> That's a good point. In current design, we strictly distinguish
> both.
> > > >> For example, the reset tools deletes internal topics (starting with
> > > >> prefix `-` and ending with either `-repartition` or
> > > >> `-changelog`.
> > > >>
> > > >> Thus, from my point of view, it would make sense to keep the current
> > > >> distinction.
> > > >>
> > > >> -Matthias
> > > >>
> > > >> On 11/6/17 4:45 PM, Thomas Becker wrote:
> > > >>
> > > >>
> > > >> I think this sounds good as well. It's worth clarifying whether
> topics
> > > that are named by the user but created by streams are considered
> > "internal"
> > > topics also.
> > > >>
> > > >> On Sun, 2017-11-05 at 23:02 +0100, Matthias J. Sax wrote:
> > > >>
> > > >> My idea was, to relax the requirement for through() that a topic
> must
> > be
> > > >> created manually before startup.
> > > >>
> > > >> Thus, if no through() call is made, a (internal) topic is 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Ted Yu]

bq. enlarge the score of through()

I guess you meant scope.

On Mon, Nov 6, 2017 at 1:15 PM, Jeyhun Karimov  wrote:

> Hi,
>
> Sorry for the late reply. I am convinced that we should enlarge the score
> of through() (add more overloads) instead of introducing a separate set of
> overloads to other methods.
> I will update the KIP soon based on the discussion and inform.
>
>
> Cheers,
> Jeyhun
>
> On Mon, Nov 6, 2017 at 9:18 PM Jan Filipiak 
> wrote:
>
> > Sorry for not beeing 100% up to date.
> > Back then we had the discussion that when an operation puts a >Sink<
> > into the topology, a >Produced<
> > parameter is added. This produced parameter could have internal or
> > external. If internal I think the name would still make
> > a great suffix for the topic name
> >
> > Is this plan still around? Otherwise having the name as suffix is
> > probably always good it can help the user quicker to identify hot topics
> > that need more
> > partitions if he has many of these internal repartitions
> >
> > Best Jan
> >
> >
> > On 06.11.2017 20:13, Matthias J. Sax wrote:
> > > I absolute agree with what you say. It's not a requirement to specify a
> > > topic name -- and this was the idea -- if user does specify a name, we
> > > treat as is -- if users does not specify a name, Streams create an
> > > internal topic.
> > >
> > > The goal of the Jira is to allow a simplified way to control
> > > repartitioning (atm, user needs to manually create a topic and use via
> > > through()).
> > >
> > > Thus, the idea is to make the topic name parameter of through optional.
> > >
> > > It's of course just an idea. Happy do have a other API design. The goal
> > > was, to avoid to many new overloads.
> > >
> > >>> Could you clarify exactly what you mean by keeping the current
> > distinction?
> > > Current distinction is: user topics are created manually and user
> > > specifies the name -- internal topics are created by Kafka Streams and
> > > an name is generated automatically.
> > >
> > > -> through("user-topic")
> > > -> through(TopicConfig.withNumberOfPartitions(5)) // Streams creates
> an
> > > internal topic
> > >
> > >
> > > -Matthias
> > >
> > >
> > > On 11/6/17 6:56 PM, Thomas Becker wrote:
> > >> Could you clarify exactly what you mean by keeping the current
> > distinction?
> > >>
> > >> Actually, re-reading the KIP and JIRA, it's not clear that being able
> > to specify a custom name is actually a requirement. If the goal is to
> > control repartitioning and tune parallelism, maybe we can just sidestep
> > this issue altogether by removing the ability to set a different name.
> > >>
> > >> On Mon, 2017-11-06 at 16:51 +0100, Matthias J. Sax wrote:
> > >>
> > >> That's a good point. In current design, we strictly distinguish both.
> > >> For example, the reset tools deletes internal topics (starting with
> > >> prefix `-` and ending with either `-repartition` or
> > >> `-changelog`.
> > >>
> > >> Thus, from my point of view, it would make sense to keep the current
> > >> distinction.
> > >>
> > >> -Matthias
> > >>
> > >> On 11/6/17 4:45 PM, Thomas Becker wrote:
> > >>
> > >>
> > >> I think this sounds good as well. It's worth clarifying whether topics
> > that are named by the user but created by streams are considered
> "internal"
> > topics also.
> > >>
> > >> On Sun, 2017-11-05 at 23:02 +0100, Matthias J. Sax wrote:
> > >>
> > >> My idea was, to relax the requirement for through() that a topic must
> be
> > >> created manually before startup.
> > >>
> > >> Thus, if no through() call is made, a (internal) topic is created the
> > >> same way we do it currently.
> > >>
> > >> If one uses `through(String topicName)` we keep the current behavior
> and
> > >> require users to create the topic manually.
> > >>
> > >> The reasoning is as follows: if a user creates a topic manually, a
> user
> > >> can just use it for repartitioning. As the topic is already there,
> there
> > >> is no need to specify any topic configs.
> > >>
> > >> We add a new `through()` overload (details TBD) that allows to specify
> > >> topic configs and Streams create the topic with those configs.
> > >>
> > >> Reasoning: user don't want to manage topic manually, thus, it's still
> an
> > >> internal topic and Streams create the topic name automatically as for
> > >> all other internal topics. However, users gets some more control about
> > >> topic parameters like number of partitions (we should discuss what
> other
> > >> configs would be useful).
> > >>
> > >>
> > >> Does this make sense?
> > >>
> > >>
> > >> -Matthias
> > >>
> > >>
> > >> On 11/5/17 1:21 AM, Jan Filipiak wrote:
> > >>
> > >>
> > >> Hi.
> > >>
> > >>
> > >> Im not 100 % up to date what version 1.0 DSL looks like ATM.
> > >> I just would argue that repartitioning should be an own API call like
> > >> through or something.
> > >> One can use through or to already to get this. I would argue one
> should
> > >> look there instead of 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Jeyhun Karimov]

Hi,

Sorry for the late reply. I am convinced that we should enlarge the score
of through() (add more overloads) instead of introducing a separate set of
overloads to other methods.
I will update the KIP soon based on the discussion and inform.


Cheers,
Jeyhun

On Mon, Nov 6, 2017 at 9:18 PM Jan Filipiak 
wrote:

> Sorry for not beeing 100% up to date.
> Back then we had the discussion that when an operation puts a >Sink<
> into the topology, a >Produced<
> parameter is added. This produced parameter could have internal or
> external. If internal I think the name would still make
> a great suffix for the topic name
>
> Is this plan still around? Otherwise having the name as suffix is
> probably always good it can help the user quicker to identify hot topics
> that need more
> partitions if he has many of these internal repartitions
>
> Best Jan
>
>
> On 06.11.2017 20:13, Matthias J. Sax wrote:
> > I absolute agree with what you say. It's not a requirement to specify a
> > topic name -- and this was the idea -- if user does specify a name, we
> > treat as is -- if users does not specify a name, Streams create an
> > internal topic.
> >
> > The goal of the Jira is to allow a simplified way to control
> > repartitioning (atm, user needs to manually create a topic and use via
> > through()).
> >
> > Thus, the idea is to make the topic name parameter of through optional.
> >
> > It's of course just an idea. Happy do have a other API design. The goal
> > was, to avoid to many new overloads.
> >
> >>> Could you clarify exactly what you mean by keeping the current
> distinction?
> > Current distinction is: user topics are created manually and user
> > specifies the name -- internal topics are created by Kafka Streams and
> > an name is generated automatically.
> >
> > -> through("user-topic")
> > -> through(TopicConfig.withNumberOfPartitions(5)) // Streams creates an
> > internal topic
> >
> >
> > -Matthias
> >
> >
> > On 11/6/17 6:56 PM, Thomas Becker wrote:
> >> Could you clarify exactly what you mean by keeping the current
> distinction?
> >>
> >> Actually, re-reading the KIP and JIRA, it's not clear that being able
> to specify a custom name is actually a requirement. If the goal is to
> control repartitioning and tune parallelism, maybe we can just sidestep
> this issue altogether by removing the ability to set a different name.
> >>
> >> On Mon, 2017-11-06 at 16:51 +0100, Matthias J. Sax wrote:
> >>
> >> That's a good point. In current design, we strictly distinguish both.
> >> For example, the reset tools deletes internal topics (starting with
> >> prefix `-` and ending with either `-repartition` or
> >> `-changelog`.
> >>
> >> Thus, from my point of view, it would make sense to keep the current
> >> distinction.
> >>
> >> -Matthias
> >>
> >> On 11/6/17 4:45 PM, Thomas Becker wrote:
> >>
> >>
> >> I think this sounds good as well. It's worth clarifying whether topics
> that are named by the user but created by streams are considered "internal"
> topics also.
> >>
> >> On Sun, 2017-11-05 at 23:02 +0100, Matthias J. Sax wrote:
> >>
> >> My idea was, to relax the requirement for through() that a topic must be
> >> created manually before startup.
> >>
> >> Thus, if no through() call is made, a (internal) topic is created the
> >> same way we do it currently.
> >>
> >> If one uses `through(String topicName)` we keep the current behavior and
> >> require users to create the topic manually.
> >>
> >> The reasoning is as follows: if a user creates a topic manually, a user
> >> can just use it for repartitioning. As the topic is already there, there
> >> is no need to specify any topic configs.
> >>
> >> We add a new `through()` overload (details TBD) that allows to specify
> >> topic configs and Streams create the topic with those configs.
> >>
> >> Reasoning: user don't want to manage topic manually, thus, it's still an
> >> internal topic and Streams create the topic name automatically as for
> >> all other internal topics. However, users gets some more control about
> >> topic parameters like number of partitions (we should discuss what other
> >> configs would be useful).
> >>
> >>
> >> Does this make sense?
> >>
> >>
> >> -Matthias
> >>
> >>
> >> On 11/5/17 1:21 AM, Jan Filipiak wrote:
> >>
> >>
> >> Hi.
> >>
> >>
> >> Im not 100 % up to date what version 1.0 DSL looks like ATM.
> >> I just would argue that repartitioning should be an own API call like
> >> through or something.
> >> One can use through or to already to get this. I would argue one should
> >> look there instead of overloads
> >>
> >> Best Jan
> >>
> >> On 04.11.2017 16:01, Jeyhun Karimov wrote:
> >>
> >>
> >> Dear community,
> >>
> >> I would like to initiate discussion on KIP-221 [1] based on issue [2].
> >> Please feel free to comment.
> >>
> >> [1]
> >>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Repartition+Topic+Hints+in+Streams
> >>
> >> [2] https://issues.apache.org/jira/browse/KAFKA-6037

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Jan Filipiak]

Sorry for not beeing 100% up to date.
Back then we had the discussion that when an operation puts a >Sink< 
into the topology, a >Produced<
parameter is added. This produced parameter could have internal or 
external. If internal I think the name would still make

a great suffix for the topic name

Is this plan still around? Otherwise having the name as suffix is 
probably always good it can help the user quicker to identify hot topics 
that need more

partitions if he has many of these internal repartitions

Best Jan


On 06.11.2017 20:13, Matthias J. Sax wrote:

I absolute agree with what you say. It's not a requirement to specify a
topic name -- and this was the idea -- if user does specify a name, we
treat as is -- if users does not specify a name, Streams create an
internal topic.

The goal of the Jira is to allow a simplified way to control
repartitioning (atm, user needs to manually create a topic and use via
through()).

Thus, the idea is to make the topic name parameter of through optional.

It's of course just an idea. Happy do have a other API design. The goal
was, to avoid to many new overloads.


Could you clarify exactly what you mean by keeping the current distinction?

Current distinction is: user topics are created manually and user
specifies the name -- internal topics are created by Kafka Streams and
an name is generated automatically.

-> through("user-topic")
-> through(TopicConfig.withNumberOfPartitions(5)) // Streams creates an
internal topic


-Matthias


On 11/6/17 6:56 PM, Thomas Becker wrote:

Could you clarify exactly what you mean by keeping the current distinction?

Actually, re-reading the KIP and JIRA, it's not clear that being able to 
specify a custom name is actually a requirement. If the goal is to control 
repartitioning and tune parallelism, maybe we can just sidestep this issue 
altogether by removing the ability to set a different name.

On Mon, 2017-11-06 at 16:51 +0100, Matthias J. Sax wrote:

That's a good point. In current design, we strictly distinguish both.
For example, the reset tools deletes internal topics (starting with
prefix `-` and ending with either `-repartition` or
`-changelog`.

Thus, from my point of view, it would make sense to keep the current
distinction.

-Matthias

On 11/6/17 4:45 PM, Thomas Becker wrote:


I think this sounds good as well. It's worth clarifying whether topics that are named by 
the user but created by streams are considered "internal" topics also.

On Sun, 2017-11-05 at 23:02 +0100, Matthias J. Sax wrote:

My idea was, to relax the requirement for through() that a topic must be
created manually before startup.

Thus, if no through() call is made, a (internal) topic is created the
same way we do it currently.

If one uses `through(String topicName)` we keep the current behavior and
require users to create the topic manually.

The reasoning is as follows: if a user creates a topic manually, a user
can just use it for repartitioning. As the topic is already there, there
is no need to specify any topic configs.

We add a new `through()` overload (details TBD) that allows to specify
topic configs and Streams create the topic with those configs.

Reasoning: user don't want to manage topic manually, thus, it's still an
internal topic and Streams create the topic name automatically as for
all other internal topics. However, users gets some more control about
topic parameters like number of partitions (we should discuss what other
configs would be useful).


Does this make sense?


-Matthias


On 11/5/17 1:21 AM, Jan Filipiak wrote:


Hi.


Im not 100 % up to date what version 1.0 DSL looks like ATM.
I just would argue that repartitioning should be an own API call like
through or something.
One can use through or to already to get this. I would argue one should
look there instead of overloads

Best Jan

On 04.11.2017 16:01, Jeyhun Karimov wrote:


Dear community,

I would like to initiate discussion on KIP-221 [1] based on issue [2].
Please feel free to comment.

[1]
https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Repartition+Topic+Hints+in+Streams

[2] https://issues.apache.org/jira/browse/KAFKA-6037



Cheers,
Jeyhun











This email and any attachments may contain confidential and privileged material 
for the sole use of the intended recipient. Any review, copying, or 
distribution of this email (or any attachments) by others is prohibited. If you 
are not the intended recipient, please contact the sender immediately and 
permanently delete this email and any attachments. No employee or agent of TiVo 
Inc. is authorized to conclude any binding agreement on behalf of TiVo Inc. by 
email. Binding agreements with TiVo Inc. may only be made by a signed written 
agreement.








This email and any attachments may contain confidential and privileged material 
for the sole use of the intended recipient. Any review, copying, or 
distribution of this email 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Matthias J. Sax]

I absolute agree with what you say. It's not a requirement to specify a
topic name -- and this was the idea -- if user does specify a name, we
treat as is -- if users does not specify a name, Streams create an
internal topic.

The goal of the Jira is to allow a simplified way to control
repartitioning (atm, user needs to manually create a topic and use via
through()).

Thus, the idea is to make the topic name parameter of through optional.

It's of course just an idea. Happy do have a other API design. The goal
was, to avoid to many new overloads.

>> Could you clarify exactly what you mean by keeping the current distinction?

Current distinction is: user topics are created manually and user
specifies the name -- internal topics are created by Kafka Streams and
an name is generated automatically.

-> through("user-topic")
-> through(TopicConfig.withNumberOfPartitions(5)) // Streams creates an
internal topic


-Matthias


On 11/6/17 6:56 PM, Thomas Becker wrote:
> Could you clarify exactly what you mean by keeping the current distinction?
> 
> Actually, re-reading the KIP and JIRA, it's not clear that being able to 
> specify a custom name is actually a requirement. If the goal is to control 
> repartitioning and tune parallelism, maybe we can just sidestep this issue 
> altogether by removing the ability to set a different name.
> 
> On Mon, 2017-11-06 at 16:51 +0100, Matthias J. Sax wrote:
> 
> That's a good point. In current design, we strictly distinguish both.
> For example, the reset tools deletes internal topics (starting with
> prefix `-` and ending with either `-repartition` or
> `-changelog`.
> 
> Thus, from my point of view, it would make sense to keep the current
> distinction.
> 
> -Matthias
> 
> On 11/6/17 4:45 PM, Thomas Becker wrote:
> 
> 
> I think this sounds good as well. It's worth clarifying whether topics that 
> are named by the user but created by streams are considered "internal" topics 
> also.
> 
> On Sun, 2017-11-05 at 23:02 +0100, Matthias J. Sax wrote:
> 
> My idea was, to relax the requirement for through() that a topic must be
> created manually before startup.
> 
> Thus, if no through() call is made, a (internal) topic is created the
> same way we do it currently.
> 
> If one uses `through(String topicName)` we keep the current behavior and
> require users to create the topic manually.
> 
> The reasoning is as follows: if a user creates a topic manually, a user
> can just use it for repartitioning. As the topic is already there, there
> is no need to specify any topic configs.
> 
> We add a new `through()` overload (details TBD) that allows to specify
> topic configs and Streams create the topic with those configs.
> 
> Reasoning: user don't want to manage topic manually, thus, it's still an
> internal topic and Streams create the topic name automatically as for
> all other internal topics. However, users gets some more control about
> topic parameters like number of partitions (we should discuss what other
> configs would be useful).
> 
> 
> Does this make sense?
> 
> 
> -Matthias
> 
> 
> On 11/5/17 1:21 AM, Jan Filipiak wrote:
> 
> 
> Hi.
> 
> 
> Im not 100 % up to date what version 1.0 DSL looks like ATM.
> I just would argue that repartitioning should be an own API call like
> through or something.
> One can use through or to already to get this. I would argue one should
> look there instead of overloads
> 
> Best Jan
> 
> On 04.11.2017 16:01, Jeyhun Karimov wrote:
> 
> 
> Dear community,
> 
> I would like to initiate discussion on KIP-221 [1] based on issue [2].
> Please feel free to comment.
> 
> [1]
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Repartition+Topic+Hints+in+Streams
> 
> [2] https://issues.apache.org/jira/browse/KAFKA-6037
> 
> 
> 
> Cheers,
> Jeyhun
> 
> 
> 
> 
> 
> 
> 
> 
> 
> 
> 
> This email and any attachments may contain confidential and privileged 
> material for the sole use of the intended recipient. Any review, copying, or 
> distribution of this email (or any attachments) by others is prohibited. If 
> you are not the intended recipient, please contact the sender immediately and 
> permanently delete this email and any attachments. No employee or agent of 
> TiVo Inc. is authorized to conclude any binding agreement on behalf of TiVo 
> Inc. by email. Binding agreements with TiVo Inc. may only be made by a signed 
> written agreement.
> 
> 
> 
> 
> 
> 
> 
> 
> This email and any attachments may contain confidential and privileged 
> material for the sole use of the intended recipient. Any review, copying, or 
> distribution of this email (or any attachments) by others is prohibited. If 
> you are not the intended recipient, please contact the sender immediately and 
> permanently delete this email and any attachments. No employee or agent of 
> TiVo Inc. is authorized to conclude any binding agreement on behalf of TiVo 
> Inc. by email. Binding agreements with TiVo 

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Thomas Becker]

Could you clarify exactly what you mean by keeping the current distinction?

Actually, re-reading the KIP and JIRA, it's not clear that being able to 
specify a custom name is actually a requirement. If the goal is to control 
repartitioning and tune parallelism, maybe we can just sidestep this issue 
altogether by removing the ability to set a different name.

On Mon, 2017-11-06 at 16:51 +0100, Matthias J. Sax wrote:

That's a good point. In current design, we strictly distinguish both.
For example, the reset tools deletes internal topics (starting with
prefix `-` and ending with either `-repartition` or
`-changelog`.

Thus, from my point of view, it would make sense to keep the current
distinction.

-Matthias

On 11/6/17 4:45 PM, Thomas Becker wrote:


I think this sounds good as well. It's worth clarifying whether topics that are 
named by the user but created by streams are considered "internal" topics also.

On Sun, 2017-11-05 at 23:02 +0100, Matthias J. Sax wrote:

My idea was, to relax the requirement for through() that a topic must be
created manually before startup.

Thus, if no through() call is made, a (internal) topic is created the
same way we do it currently.

If one uses `through(String topicName)` we keep the current behavior and
require users to create the topic manually.

The reasoning is as follows: if a user creates a topic manually, a user
can just use it for repartitioning. As the topic is already there, there
is no need to specify any topic configs.

We add a new `through()` overload (details TBD) that allows to specify
topic configs and Streams create the topic with those configs.

Reasoning: user don't want to manage topic manually, thus, it's still an
internal topic and Streams create the topic name automatically as for
all other internal topics. However, users gets some more control about
topic parameters like number of partitions (we should discuss what other
configs would be useful).


Does this make sense?


-Matthias


On 11/5/17 1:21 AM, Jan Filipiak wrote:


Hi.


Im not 100 % up to date what version 1.0 DSL looks like ATM.
I just would argue that repartitioning should be an own API call like
through or something.
One can use through or to already to get this. I would argue one should
look there instead of overloads

Best Jan

On 04.11.2017 16:01, Jeyhun Karimov wrote:


Dear community,

I would like to initiate discussion on KIP-221 [1] based on issue [2].
Please feel free to comment.

[1]
https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Repartition+Topic+Hints+in+Streams

[2] https://issues.apache.org/jira/browse/KAFKA-6037



Cheers,
Jeyhun











This email and any attachments may contain confidential and privileged material 
for the sole use of the intended recipient. Any review, copying, or 
distribution of this email (or any attachments) by others is prohibited. If you 
are not the intended recipient, please contact the sender immediately and 
permanently delete this email and any attachments. No employee or agent of TiVo 
Inc. is authorized to conclude any binding agreement on behalf of TiVo Inc. by 
email. Binding agreements with TiVo Inc. may only be made by a signed written 
agreement.








This email and any attachments may contain confidential and privileged material 
for the sole use of the intended recipient. Any review, copying, or 
distribution of this email (or any attachments) by others is prohibited. If you 
are not the intended recipient, please contact the sender immediately and 
permanently delete this email and any attachments. No employee or agent of TiVo 
Inc. is authorized to conclude any binding agreement on behalf of TiVo Inc. by 
email. Binding agreements with TiVo Inc. may only be made by a signed written 
agreement.

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Matthias J. Sax]

That's a good point. In current design, we strictly distinguish both.
For example, the reset tools deletes internal topics (starting with
prefix `-` and ending with either `-repartition` or
`-changelog`.

Thus, from my point of view, it would make sense to keep the current
distinction.

-Matthias

On 11/6/17 4:45 PM, Thomas Becker wrote:
> I think this sounds good as well. It's worth clarifying whether topics that 
> are named by the user but created by streams are considered "internal" topics 
> also.
> 
> On Sun, 2017-11-05 at 23:02 +0100, Matthias J. Sax wrote:
> 
> My idea was, to relax the requirement for through() that a topic must be
> created manually before startup.
> 
> Thus, if no through() call is made, a (internal) topic is created the
> same way we do it currently.
> 
> If one uses `through(String topicName)` we keep the current behavior and
> require users to create the topic manually.
> 
> The reasoning is as follows: if a user creates a topic manually, a user
> can just use it for repartitioning. As the topic is already there, there
> is no need to specify any topic configs.
> 
> We add a new `through()` overload (details TBD) that allows to specify
> topic configs and Streams create the topic with those configs.
> 
> Reasoning: user don't want to manage topic manually, thus, it's still an
> internal topic and Streams create the topic name automatically as for
> all other internal topics. However, users gets some more control about
> topic parameters like number of partitions (we should discuss what other
> configs would be useful).
> 
> 
> Does this make sense?
> 
> 
> -Matthias
> 
> 
> On 11/5/17 1:21 AM, Jan Filipiak wrote:
> 
> 
> Hi.
> 
> 
> Im not 100 % up to date what version 1.0 DSL looks like ATM.
> I just would argue that repartitioning should be an own API call like
> through or something.
> One can use through or to already to get this. I would argue one should
> look there instead of overloads
> 
> Best Jan
> 
> On 04.11.2017 16:01, Jeyhun Karimov wrote:
> 
> 
> Dear community,
> 
> I would like to initiate discussion on KIP-221 [1] based on issue [2].
> Please feel free to comment.
> 
> [1]
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Repartition+Topic+Hints+in+Streams
> 
> [2] https://issues.apache.org/jira/browse/KAFKA-6037
> 
> 
> 
> Cheers,
> Jeyhun
> 
> 
> 
> 
> 
> 
> 
> 
> 
> 
> 
> This email and any attachments may contain confidential and privileged 
> material for the sole use of the intended recipient. Any review, copying, or 
> distribution of this email (or any attachments) by others is prohibited. If 
> you are not the intended recipient, please contact the sender immediately and 
> permanently delete this email and any attachments. No employee or agent of 
> TiVo Inc. is authorized to conclude any binding agreement on behalf of TiVo 
> Inc. by email. Binding agreements with TiVo Inc. may only be made by a signed 
> written agreement.
> 



signature.asc
Description: OpenPGP digital signature

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Thomas Becker]

I think this sounds good as well. It's worth clarifying whether topics that are 
named by the user but created by streams are considered "internal" topics also.

On Sun, 2017-11-05 at 23:02 +0100, Matthias J. Sax wrote:

My idea was, to relax the requirement for through() that a topic must be
created manually before startup.

Thus, if no through() call is made, a (internal) topic is created the
same way we do it currently.

If one uses `through(String topicName)` we keep the current behavior and
require users to create the topic manually.

The reasoning is as follows: if a user creates a topic manually, a user
can just use it for repartitioning. As the topic is already there, there
is no need to specify any topic configs.

We add a new `through()` overload (details TBD) that allows to specify
topic configs and Streams create the topic with those configs.

Reasoning: user don't want to manage topic manually, thus, it's still an
internal topic and Streams create the topic name automatically as for
all other internal topics. However, users gets some more control about
topic parameters like number of partitions (we should discuss what other
configs would be useful).


Does this make sense?


-Matthias


On 11/5/17 1:21 AM, Jan Filipiak wrote:


Hi.


Im not 100 % up to date what version 1.0 DSL looks like ATM.
I just would argue that repartitioning should be an own API call like
through or something.
One can use through or to already to get this. I would argue one should
look there instead of overloads

Best Jan

On 04.11.2017 16:01, Jeyhun Karimov wrote:


Dear community,

I would like to initiate discussion on KIP-221 [1] based on issue [2].
Please feel free to comment.

[1]
https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Repartition+Topic+Hints+in+Streams

[2] https://issues.apache.org/jira/browse/KAFKA-6037



Cheers,
Jeyhun











This email and any attachments may contain confidential and privileged material 
for the sole use of the intended recipient. Any review, copying, or 
distribution of this email (or any attachments) by others is prohibited. If you 
are not the intended recipient, please contact the sender immediately and 
permanently delete this email and any attachments. No employee or agent of TiVo 
Inc. is authorized to conclude any binding agreement on behalf of TiVo Inc. by 
email. Binding agreements with TiVo Inc. may only be made by a signed written 
agreement.

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Jan Filipiak]

That would be nice I think

On 05.11.2017 23:02, Matthias J. Sax wrote:

My idea was, to relax the requirement for through() that a topic must be
created manually before startup.

Thus, if no through() call is made, a (internal) topic is created the
same way we do it currently.

If one uses `through(String topicName)` we keep the current behavior and
require users to create the topic manually.

The reasoning is as follows: if a user creates a topic manually, a user
can just use it for repartitioning. As the topic is already there, there
is no need to specify any topic configs.

We add a new `through()` overload (details TBD) that allows to specify
topic configs and Streams create the topic with those configs.

Reasoning: user don't want to manage topic manually, thus, it's still an
internal topic and Streams create the topic name automatically as for
all other internal topics. However, users gets some more control about
topic parameters like number of partitions (we should discuss what other
configs would be useful).


Does this make sense?


-Matthias


On 11/5/17 1:21 AM, Jan Filipiak wrote:

Hi.


Im not 100 % up to date what version 1.0 DSL looks like ATM.
I just would argue that repartitioning should be an own API call like
through or something.
One can use through or to already to get this. I would argue one should
look there instead of overloads

Best Jan

On 04.11.2017 16:01, Jeyhun Karimov wrote:

Dear community,

I would like to initiate discussion on KIP-221 [1] based on issue [2].
Please feel free to comment.

[1]
https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Repartition+Topic+Hints+in+Streams

[2] https://issues.apache.org/jira/browse/KAFKA-6037



Cheers,
Jeyhun

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Matthias J. Sax]

My idea was, to relax the requirement for through() that a topic must be
created manually before startup.

Thus, if no through() call is made, a (internal) topic is created the
same way we do it currently.

If one uses `through(String topicName)` we keep the current behavior and
require users to create the topic manually.

The reasoning is as follows: if a user creates a topic manually, a user
can just use it for repartitioning. As the topic is already there, there
is no need to specify any topic configs.

We add a new `through()` overload (details TBD) that allows to specify
topic configs and Streams create the topic with those configs.

Reasoning: user don't want to manage topic manually, thus, it's still an
internal topic and Streams create the topic name automatically as for
all other internal topics. However, users gets some more control about
topic parameters like number of partitions (we should discuss what other
configs would be useful).


Does this make sense?


-Matthias


On 11/5/17 1:21 AM, Jan Filipiak wrote:
> Hi.
> 
> 
> Im not 100 % up to date what version 1.0 DSL looks like ATM.
> I just would argue that repartitioning should be an own API call like
> through or something.
> One can use through or to already to get this. I would argue one should
> look there instead of overloads
> 
> Best Jan
> 
> On 04.11.2017 16:01, Jeyhun Karimov wrote:
>> Dear community,
>>
>> I would like to initiate discussion on KIP-221 [1] based on issue [2].
>> Please feel free to comment.
>>
>> [1]
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Repartition+Topic+Hints+in+Streams
>>
>> [2] https://issues.apache.org/jira/browse/KAFKA-6037
>>
>>
>>
>> Cheers,
>> Jeyhun
>>
> 



signature.asc
Description: OpenPGP digital signature

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Jan Filipiak]

Hi.


Im not 100 % up to date what version 1.0 DSL looks like ATM.
I just would argue that repartitioning should be an own API call like 
through or something.
One can use through or to already to get this. I would argue one should 
look there instead of overloads


Best Jan

On 04.11.2017 16:01, Jeyhun Karimov wrote:

Dear community,

I would like to initiate discussion on KIP-221 [1] based on issue [2].
Please feel free to comment.

[1]
https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Repartition+Topic+Hints+in+Streams
[2] https://issues.apache.org/jira/browse/KAFKA-6037



Cheers,
Jeyhun

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Jeyhun Karimov]

Hi,


Thanks for your comments.

@Ted

API is given without much javadoc on the role / meaning of method
> parameters.


- I thought they are self-explanatory but I will add some more comments in
the document.

@Matthias

 - how does this relate to `KStream#through()` ?


- The main difference between `KStream#through()` and this KIP is that
the repartition
topic should be created manually (when using `KStream#through()`). However,
it is better if we can handle some part of manual work transparent to the
user.
I thought to add this functionality to `Produced`. However, there was some
conceptual issues.

In all `KStream#through()` method overloads, we state that topics should be
created manually. If we overload another `KStream#through()` method (say,
with different parameters) saying the topic name/partitions is up to the
user, this might seem weird.
Either we can enlarge the scope of `KStream#through()`  or we create new
overloads and keep both (`KStream#through()` and RepartitionHint) of them.



Cheers,
Jeyhun

On Sat, Nov 4, 2017 at 7:22 PM Ted Yu  wrote:

> w.r.t. KIP-182, KAFKA-5651 has been resolved.
> But the KIP is still labeled Under Discussion.
>
> Should KIP-182 be moved to Adopted state ?
>
> On Sat, Nov 4, 2017 at 10:06 AM, Matthias J. Sax 
> wrote:
>
> > Thanks for the KIP.
> >
> > Two comments/questions:
> >
> >  - how does this relate to `KStream#through()` ?
> >  - do we really need to add new overload or can we just extends exiting
> > options (cf KIP-182); it would be create if we could just extend
> > existing classes like `Produced` instead of adding new methods to "main"
> > classes
> >
> > Maybe, we could use `through()` to create a topics in the back ground
> > and picky back all to `Produced`.
> >
> >
> > -Matthias
> >
> > On 11/4/17 4:16 PM, Ted Yu wrote:
> > > API is given without much javadoc on the role / meaning of method
> > > parameters.
> > >
> > > Can you enrich with descriptive javadoc ?
> > >
> > > On Sat, Nov 4, 2017 at 8:01 AM, Jeyhun Karimov 
> > wrote:
> > >
> > >> Dear community,
> > >>
> > >> I would like to initiate discussion on KIP-221 [1] based on issue [2].
> > >> Please feel free to comment.
> > >>
> > >> [1]
> > >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> > >> 221%3A+Repartition+Topic+Hints+in+Streams
> > >> [2] https://issues.apache.org/jira/browse/KAFKA-6037
> > >>
> > >>
> > >>
> > >> Cheers,
> > >> Jeyhun
> > >>
> > >
> >
> >
>

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Ted Yu]

w.r.t. KIP-182, KAFKA-5651 has been resolved.
But the KIP is still labeled Under Discussion.

Should KIP-182 be moved to Adopted state ?

On Sat, Nov 4, 2017 at 10:06 AM, Matthias J. Sax 
wrote:

> Thanks for the KIP.
>
> Two comments/questions:
>
>  - how does this relate to `KStream#through()` ?
>  - do we really need to add new overload or can we just extends exiting
> options (cf KIP-182); it would be create if we could just extend
> existing classes like `Produced` instead of adding new methods to "main"
> classes
>
> Maybe, we could use `through()` to create a topics in the back ground
> and picky back all to `Produced`.
>
>
> -Matthias
>
> On 11/4/17 4:16 PM, Ted Yu wrote:
> > API is given without much javadoc on the role / meaning of method
> > parameters.
> >
> > Can you enrich with descriptive javadoc ?
> >
> > On Sat, Nov 4, 2017 at 8:01 AM, Jeyhun Karimov 
> wrote:
> >
> >> Dear community,
> >>
> >> I would like to initiate discussion on KIP-221 [1] based on issue [2].
> >> Please feel free to comment.
> >>
> >> [1]
> >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> >> 221%3A+Repartition+Topic+Hints+in+Streams
> >> [2] https://issues.apache.org/jira/browse/KAFKA-6037
> >>
> >>
> >>
> >> Cheers,
> >> Jeyhun
> >>
> >
>
>

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Matthias J. Sax]

Thanks for the KIP.

Two comments/questions:

 - how does this relate to `KStream#through()` ?
 - do we really need to add new overload or can we just extends exiting
options (cf KIP-182); it would be create if we could just extend
existing classes like `Produced` instead of adding new methods to "main"
classes

Maybe, we could use `through()` to create a topics in the back ground
and picky back all to `Produced`.


-Matthias

On 11/4/17 4:16 PM, Ted Yu wrote:
> API is given without much javadoc on the role / meaning of method
> parameters.
> 
> Can you enrich with descriptive javadoc ?
> 
> On Sat, Nov 4, 2017 at 8:01 AM, Jeyhun Karimov  wrote:
> 
>> Dear community,
>>
>> I would like to initiate discussion on KIP-221 [1] based on issue [2].
>> Please feel free to comment.
>>
>> [1]
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-
>> 221%3A+Repartition+Topic+Hints+in+Streams
>> [2] https://issues.apache.org/jira/browse/KAFKA-6037
>>
>>
>>
>> Cheers,
>> Jeyhun
>>
> 



signature.asc
Description: OpenPGP digital signature

Re: [DISCUSS] KIP-221: Repartition Topic Hints in Streams

[Ted Yu]

API is given without much javadoc on the role / meaning of method
parameters.

Can you enrich with descriptive javadoc ?

On Sat, Nov 4, 2017 at 8:01 AM, Jeyhun Karimov  wrote:

> Dear community,
>
> I would like to initiate discussion on KIP-221 [1] based on issue [2].
> Please feel free to comment.
>
> [1]
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> 221%3A+Repartition+Topic+Hints+in+Streams
> [2] https://issues.apache.org/jira/browse/KAFKA-6037
>
>
>
> Cheers,
> Jeyhun
>