[
https://issues.apache.org/jira/browse/KAFKA-4835?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15907523#comment-15907523
]
Michal Borowiecki commented on KAFKA-4835:
------------------------------------------
My point above was that if the customerRef (which is what we partition by) was
part of the key (not the whole key) then we'd still need to modify the key for
the purpose of the join operation.
We'd need to do that for both streams, even though they would both be
partitioned by the same part of the key, hence the re-partitioning (forced
automatically by kafka streams) would be totally unnecessary.
In more generic terms, I think this can be a common use case. Let' consider it
using DDD concepts. We have an aggregate comprised of multiple entities. We
send messages for each entity (not the whole aggregate) but to ensure
sequential processing for entities belonging to the same aggregate, the
messages are partitioned by the aggregate id. The entity id is still important,
especially for compacted topics it would be needed for deletion markers, as the
key is all there is in that case. Hence it comes naturally to compose the
message key as <aggregate-id>:<entity-type>:<entity-id>
Then, if you want to join two such streams by aggregate id, you should be able
to do it without repartitioning (since both partitioned by the aggregate-id
part of the msg key). However, since joins are only supported on the whole msg
key, you're forced to re-map the key to just <aggregate-id> prior to the join
which in turn currently forces repartitioning.
> Allow users control over repartitioning
> ---------------------------------------
>
> Key: KAFKA-4835
> URL: https://issues.apache.org/jira/browse/KAFKA-4835
> Project: Kafka
> Issue Type: Improvement
> Components: streams
> Affects Versions: 0.10.2.0
> Reporter: Michal Borowiecki
> Labels: needs-kip
>
> From
> https://issues.apache.org/jira/browse/KAFKA-4601?focusedCommentId=15881030&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-15881030
> ...it would be good to provide users more control over the repartitioning.
> My use case is as follows (unrelated bits omitted for brevity):
> {code}
> KTable<String, Activity> loggedInCustomers = builder
> .stream("customerLogins")
> .groupBy((key, activity) ->
> activity.getCustomerRef())
> .reduce((first,second) -> second, loginStore());
>
> builder
> .stream("balanceUpdates")
> .map((key, activity) -> new KeyValue<>(
> activity.getCustomerRef(),
> activity))
> .join(loggedInCustomers, (activity, session) -> ...
> .to("sessions");
> {code}
> Both "groupBy" and "map" in the underlying implementation set the
> repartitionRequired flag (since the key changes), and the aggregation/join
> that follows will create the repartitioned topic.
> However, in our case I know that both input streams are already partitioned
> by the customerRef value, which I'm mapping into the key (because it's
> required by the join operation).
> So there are 2 unnecessary intermediate topics created with their associated
> overhead, while the ultimate goal is simply to do a join on a value that we
> already use to partition the original streams anyway.
> (Note, we don't have the option to re-implement the original input streams to
> make customerRef the message key.)
> I think it would be better to allow the user to decide (from their knowledge
> of the incoming streams) whether a repartition is mandatory on aggregation
> and join operations (overloaded version of the methods with the
> repartitionRequired flag exposed maybe?)
> An alternative would be to allow users to perform a join on a value other
> than the key (a keyValueMapper parameter to join, like the one used for joins
> with global tables), but I expect that to be more involved and error-prone to
> use for people who don't understand the partitioning requirements well
> (whereas it's safe for global tables).
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
