# Fresh aggregate for each join like BlockingQueue#drainTo **URL:** \ **Category:** Kafka Streams\ **Created:** [13 January 2023 13:08 UTC](https://forum.confluent.io/t/fresh-aggregate-for-each-join-like-blockingqueue-drainto/7009 "2023-01-13T13:08:42Z")\ **Posts on this page:** 13\ **Page:** 1
**Author:** ![michaldo](https://avatars.discourse-cdn.com/v4/letter/m/f05b48/32.png) [@michaldo](https://forum.confluent.io/u/michaldo)\ **Post date:** [13 January 2023 13:08 UTC](https://forum.confluent.io/t/fresh-aggregate-for-each-join-like-blockingqueue-drainto/7009/1 "2023-01-13T13:08:42Z")
I want join stream and table. If there are many events matching table element, I would like to process them together (like BlockingQueue.drainTo) My topology is: ```auto eventStream.groupByKey() .aggregate( StringBuilder::new, (key, event, aggregate) -> aggregate.append("|").append(event)), .toStream() .join(table, (eventBag, tElem) -> "table element " + tElem + " received events " + eventBag.toString()) .peek((key, value) -> System.out.println(key +" : " + value)) .sleep(10 seconds) ``` I enter events to eventStream: - s1:run - s1:stop - s1:kill I expected output > s1 : table element received events run > s1 : table element received events stop|kill I expected output > s1 : table element received events run > s1 : table element received events run|stop|kill I see that KSTREAM-AGGREGATE-STATE-STORE only accumulates events. How reset accumulation after join? ---
**Author:** ![mjsax](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mjsax/32/3113_2.png) [@mjsax](https://forum.confluent.io/u/mjsax)\ **Post date:** [13 January 2023 23:26 UTC](https://forum.confluent.io/t/fresh-aggregate-for-each-join-like-blockingqueue-drainto/7009/2 "2023-01-13T23:26:45Z")
You can’t the aggregation is independent of the join, and it’s just accumulating all records for the same key across all time. In your example, I am not sure why you want to process run as “one group” and “stop,kill” as the second group. What is you condition to process event together? You could maybe us an windowed aggregation upstream (using “emit final” semantics) to group input records based on time, to process them together downstream. You could also do a custom aggregation upstream using `transform()` – this way, you have full control what records you want to process together, and what “groups” of records you set downstream into the join (including a reset of the aggregation). ---
**Author:** ![michaldo](https://avatars.discourse-cdn.com/v4/letter/m/f05b48/32.png) [@michaldo](https://forum.confluent.io/u/michaldo)\ **Post date:** [14 January 2023 10:28 UTC](https://forum.confluent.io/t/fresh-aggregate-for-each-join-like-blockingqueue-drainto/7009/3 "2023-01-14T10:28:03Z")
In my case I’m afraid that topology processing will be slower than incoming event pressure. That is emulated by sleep. Why I want process 2 groups independently? Because if first event is **sell 1** , second event is **sell 2** , third event is **sell 3** , I want process ( **sell 1** ) then ( **sell 2 | sell 3** ) not ( **sell 1** ) then ( **sell 1 | sell 2 | sell 3** ) because correct effective processing is **sell 1** then **sell 5** not **sell 1** then **sell 6.** I think “window” is not my case because I want process incoming events immediately, without delay and “group by key” is optimization. My “true” join is stream-table - without window. I will explore `transform()` ---
**Author:** ![mjsax](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mjsax/32/3113_2.png) [@mjsax](https://forum.confluent.io/u/mjsax)\ **Post date:** [18 January 2023 08:22 UTC](https://forum.confluent.io/t/fresh-aggregate-for-each-join-like-blockingqueue-drainto/7009/4 "2023-01-18T08:22:07Z")
> In my case I’m afraid that topology processing will be slower than incoming event pressure. That is emulated by sleep. Why do you think so? What is the expected throughput? Kafka Streams scales very well, and can processed hundreds of thousands of messages per second… > I want process ( **sell 1** ) then ( **sell 2 | sell 3** ) Your example is still rather high level. Why process sell-1 in the first group, and sell-2+sell-3 in the second group, instead of sell-1+sell-2 in the first group, and sell-3 in the second group, or even process in three groups? What is the actual criteria when you split a group? > I think “window” is not my case because I want process incoming events immediately, Records are processed immediately. Also, if you say you want to process sell2+sell3 in one group, you actually don’t process sell2 right away, but you wait for sell3 to arrive, too… Using `transform()` might still be the best way forward. Let us know how it goes. ---
**Author:** ![michaldo](https://avatars.discourse-cdn.com/v4/letter/m/f05b48/32.png) [@michaldo](https://forum.confluent.io/u/michaldo)\ **Post date:** [18 January 2023 18:52 UTC](https://forum.confluent.io/t/fresh-aggregate-for-each-join-like-blockingqueue-drainto/7009/5 "2023-01-18T18:52:08Z")
> In my case I’m afraid that topology processing will be slower than incoming event pressure. That is emulated by sleep. > Why do you think so? What is the expected throughput? Kafka Streams scales very well, and can processed hundreds of thousands of messages per second… I believe that Kafka Streams is efficient, but I’m afraid that my own code, part of topology. will be slow. Not very slow, but slow enough to have a lag against sell stream. > I want process ( **sell 1** ) then ( **sell 2 | sell 3** ) > Your example is still rather high level. Why process sell-1 in the first group, and sell-2+sell-3 in the second group, instead of sell-1+sell-2 in the first group, and sell-3 in the second group, or even process in three groups? What is the actual criteria when you split a group? In my head I have analogy to Java BlockingQueue. You may call `poll()` and process events one by one. But you may call `drainTo()` and get all waiting events at once. I know Kafka Streams have caching and can poll few events from topic in advance and merge them as single event for KTable. Going back BlockingQueue, if I use `drainTo()` and incoming events are [sell1, sell2, sell3, sell4, sell5] depending on timeline I can receive [sell1] [sell2, sell3] [sell4, sell5]. As Kafka polls topic in advance, it could be theoretically possible. But I think it is not done and I must accept that my topology will process [sell1] [sell2] [sell3] [sell4] [sell5] > I think “window” is not my case because I want process incoming events immediately, > Records are processed immediately. Also, if you say you want to process sell2+sell3 in one group, you actually don’t process sell2 right away, but you wait for sell3 to arrive, too… I don’t want wait for **sell3**. But If Kafka Stream can look ahead on partition and see **sell3** is ready, I could process both sells at once > Using `transform()` might still be the best way forward. Let us know how it goes. I tried but I it does not look promising. Event if possible, it needs reimplementing cache, which Kafka already has. [/quote] ---
**Author:** ![mjsax](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mjsax/32/3113_2.png) [@mjsax](https://forum.confluent.io/u/mjsax)\ **Post date:** [20 January 2023 03:26 UTC](https://forum.confluent.io/t/fresh-aggregate-for-each-join-like-blockingqueue-drainto/7009/6 "2023-01-20T03:26:15Z")
I see. Kafka Streams abstracts away the underlying polling and “batches” of data that are fetched, but provides a record-by-record processing API. Hence, if you want to process data together, you need to build it manually by buffering up records in a state store. What is till unclear to me it, _why_ you would want/need to process multiple records together, and why you cannot just each record individually? ---
**Author:** ![michaldo](https://avatars.discourse-cdn.com/v4/letter/m/f05b48/32.png) [@michaldo](https://forum.confluent.io/u/michaldo)\ **Post date:** [20 January 2023 15:48 UTC](https://forum.confluent.io/t/fresh-aggregate-for-each-join-like-blockingqueue-drainto/7009/7 "2023-01-20T15:48:44Z")
I’m exploring Kafka Streams opportunities and one question to me is about _“drainTo()”_ Basically I expect very frequent updates to a topic and I explore option to reduce lag. I have no concrete use-case. > Hence, if you want to process data together, you need to build it manually by buffering up records in a state store. I will just start my adventure with Kafka Stream, so I will use standard patterns at the beginning. I just wanted to know if there is a standard pattern for “drainTo” and it looks it isn’t Thank you for help ---
**Author:** ![mjsax](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mjsax/32/3113_2.png) [@mjsax](https://forum.confluent.io/u/mjsax)\ **Post date:** [20 January 2023 16:37 UTC](https://forum.confluent.io/t/fresh-aggregate-for-each-join-like-blockingqueue-drainto/7009/8 "2023-01-20T16:37:52Z")
That right, there is no “drainTo”. Such a feature was never requested, and given that Kafka Streams is distributed, it’s also unclear to me what it would exactly mean? - It seems it would need to be scoped to a single partition? - Given how KS is built, it just adding a `KStream#drainTo()` sound question able, becaues the upstream program could be arbitraty complex (not show how/if we could limit it to the “source node” case, ie, only allow it after `StreamsBuilder#stream()#drainTo()`) - If there is lag, would you want to “drainTo” endOffset? (What if endOffset moves in-between?) - Would you want to base it on consumer.poll()? Also hard to do and depends on many consumer configs (like `max.poll.records` and `fetch.max.bytes`) In the end, KS is somewhat opinionated (for a good reason) but still tries to give you as many building blocks to customize your logic. It’s often hard (to impossible) to just add a new feature, and often it’s also not require because the pattern you want to apply might actually not be ideal to begin with for stream processing. ---
**Author:** ![michaldo](https://avatars.discourse-cdn.com/v4/letter/m/f05b48/32.png) [@michaldo](https://forum.confluent.io/u/michaldo)\ **Post date:** [20 January 2023 18:07 UTC](https://forum.confluent.io/t/fresh-aggregate-for-each-join-like-blockingqueue-drainto/7009/9 "2023-01-20T18:07:06Z")
Let’s me answer: 1. It seems it would need to be scoped to a single partition? Yes 2. Given how KS is built, it just adding a `KStream#drainTo()` sound question able, becaues the upstream program could be arbitraty complex Consider topology (keys and values String) ```auto eventStream .mapValues(s -> s.toUpperCase()) .toTable() .join(table, (eventItem, tableItem) -> tableItem.concat(eventItem)) .toStream() .peek((key, value) -> System.out.println(key +" : " + value)) .sleep(10 seconds) ``` KTable has entry: **k:.** 3 events goes to KStream: **k:a, k:b, k:c** output: > k:.A > k:.C Upstream is complex (`.mapValues(s -> s.toUpperCase()`) but Kafka Stream can manage it and replace k:B, k:C with k:C. So it should be possible replace k:b, k:c with list [k:b, k:c] - or at least it does not depend on upstream complexity. 1. If there is lag, would you want to “drainTo” endOffset? (What if endOffset moves in-between?) I think it would good enough “drainTo” **k:c** event, like in previous example 1. Would you want to base it on consumer.poll()? Too technical for me. But I think that if Kafka Stream can omit event when build KTable, then can also not omit event but append to “drainTo()” ---
**Author:** ![mjsax](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mjsax/32/3113_2.png) [@mjsax](https://forum.confluent.io/u/mjsax)\ **Post date:** [23 January 2023 18:00 UTC](https://forum.confluent.io/t/fresh-aggregate-for-each-join-like-blockingqueue-drainto/7009/10 "2023-01-23T18:00:11Z")
> but Kafka Stream can manage it and replace k:B, k:C with k:C That’s semantically not sound. A `KStream` transformation cannot just “drop” k:B on the floor. A `KStream` is semantically independent events, not updates. > So it should be possible replace k:b, k:c with list [k:b, k:c] - or at least it does not depend on upstream complexity. But what is the _criteria_ to put k:b and k:c into a list? How should KS know? What is the algorithm for it? ---
**Author:** ![michaldo](https://avatars.discourse-cdn.com/v4/letter/m/f05b48/32.png) [@michaldo](https://forum.confluent.io/u/michaldo)\ **Post date:** [24 January 2023 07:42 UTC](https://forum.confluent.io/t/fresh-aggregate-for-each-join-like-blockingqueue-drainto/7009/11 "2023-01-24T07:42:14Z")
> _but Kafka Stream can manage it and replace k:B, k:C with k:C_ > That’s semantically not sound. A `KStream` transformation cannot just “drop” k:B on the floor. A `KStream` is semantically independent events, not updates. Please remember that original question was: “upstream program could be arbitraty complex”. I put this example to show, that regardless upstream complexity, KS can optimize join: - if k:b, k:c comes “faster that topology sleep” the only one join is processed: **k:.** - **k:C** - if k:b, k:c comes “slower that topology sleep” two joins are processed: **k:.** - **k:B** and **k:.** - **k:C** > _So it should be possible replace k:b, k:c with list [k:b, k:c] - or at least it does not depend on upstream complexity._ > But what is the _criteria_ to put k:b and k:c into a list? How should KS know? What is the algorithm for it? That is clear: ```auto eventStream.groupByKey() .aggregate(new List(), (key, value, aggregate) -> aggregate.add(value)) ``` But when aggregate list is passed downstream, new aggregation list should be started ---
**Author:** ![mjsax](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mjsax/32/3113_2.png) [@mjsax](https://forum.confluent.io/u/mjsax)\ **Post date:** [24 January 2023 21:02 UTC](https://forum.confluent.io/t/fresh-aggregate-for-each-join-like-blockingqueue-drainto/7009/12 "2023-01-24T21:02:24Z")
> But when aggregate list is passed downstream, new aggregation list should be started But when to sent the list downstream? When it has 2 elements in it? Or 3 elements? How is the decision make when to start a new list? ---
**Author:** ![michaldo](https://avatars.discourse-cdn.com/v4/letter/m/f05b48/32.png) [@michaldo](https://forum.confluent.io/u/michaldo)\ **Post date:** [25 January 2023 08:28 UTC](https://forum.confluent.io/t/fresh-aggregate-for-each-join-like-blockingqueue-drainto/7009/13 "2023-01-25T08:28:45Z")
> [@mjsax](#): > > But when to sent the list downstream? When it has 2 elements in it? Or 3 elements? How is the decision make when to start a new list? Let me remind second sample topology ```auto eventStream .mapValues(s -> s.toUpperCase()) .toTable() .join(table, (eventItem, tableItem) -> tableItem.concat(eventItem)) ``` eventStream has events **k:a, k:b, k:c** join has **k:A, k:C** Answer to question _when to sent the list downstream_: KS in some moment decided to sent downstream **k:C** and in the same moment the list should be sent down _How is the decision make when to start a new list?_ New list should be started when old list is sent downstream.