Is it safe to assume that KGroupedTable#aggregate subtractor is called before adder?

Ben · 8 March 2021 16:30

The documentation re KGroupedTable#aggregate
at https://kafka.apache.org/26/documentation/streams/developer-guide/dsl-api.html#streams-developer-guide-dsl-aggregating states:

When subsequent non-null values are received for a key (e.g., UPDATE), then (1) the subtractor is called with the old value as stored in the table and (2) the adder is called with the new value of the input record that was just received. The order of execution for the subtractor and adder is not defined.

The phrase “the order of execution for the subtractor and adder is not defined” seems to warn against making any assumptions re their order of execution.

Viewing the source I find that the order of execution does seem to be defined:

github.com

apache/kafka/blob/77e3ca019efe38195ec28ddab87aed52e1c403c8/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableAggregate.java#L92


      
          // the keys should never be null
          if (key == null) {
              throw new StreamsException("Record key for KTable aggregate operator with state " + storeName + " should not be null.");
          }
          
          final ValueAndTimestamp<T> oldAggAndTimestamp = store.get(key);
          final T oldAgg = getValueOrNull(oldAggAndTimestamp);
          final T intermediateAgg;
          long newTimestamp = context().timestamp();
          
          // first try to remove the old value
          if (value.oldValue != null && oldAgg != null) {
              intermediateAgg = remove.apply(key, value.oldValue, oldAgg);
              newTimestamp = Math.max(context().timestamp(), oldAggAndTimestamp.timestamp());
          } else {
              intermediateAgg = oldAgg;
          }
          
          // then try to add the new value
          final T newAgg;
          if (value.newValue != null) {

And (cross-post alert) I filed a question here about the Kafka Music Example:

github.com/confluentinc/kafka-streams-examples

Question about KafkaMusic's top five songs aggregator

opened 05:25PM - 06 Mar 21 UTC

closed 01:49AM - 09 Mar 21 UTC

benissimo

https://github.com/confluentinc/kafka-streams-examples/blob/6.0.1-post/src/main/…java/io/confluent/examples/streams/interactivequeries/kafkamusic/KafkaMusicExample.java#L345 ^ I note that the aggregator defines an adder and a subtractor, so that top five songs aggregate adds and removes SongPlayCount entries. This should work provided that, in the case of an update, the subtractor is called prior to the adder. But if an update triggered adder() + subtractor() (in that order), the subtractor() call would effectively undo the result of the adder(). But per https://kafka.apache.org/26/documentation/streams/developer-guide/dsl-api.html#streams-developer-guide-dsl-aggregating > When subsequent non-null values are received for a key (e.g., UPDATE), then (1) the subtractor is called with the old value as stored in the table and (2) the adder is called with the new value of the input record that was just received. The order of execution for the subtractor and adder is not defined. I'm confused here by "the order of execution for the subtractor and adder is not defined". Does that mean, as I'm reading it, that it's possible the subtractor() gets called after the adder()? If so, then how would the top five songs aggregate in KafkaMusicExample handle that? I would hope that the order of execution _is_ defined, as that would make designing custom aggregates much easier.

which also seems to rely on the current behavior (subtractor is called first, then adder). Or am I missing something? Would that Kafka Music Example work correctly even if adder were called first, then subtractor? (I don’t see how it would; but perhaps I’m missing something)

I’m curious to hear from fellow users of KGroupedTable#aggregate. Do you rely on an explicit order (subtractor first, etc.)?

mjsax · 9 March 2021 01:47

You should not rely on the order. It’s of course hard-coded (so you could exploit it), but the order may change between releases, and thus, if you rely on it and upgrade to a newer version of Kafka Streams you code might break.

As similar question was asked on StackOverflow, and you can find a more extended answer with more background there: apache kafka - Clarify "the order of execution for the subtractor and adder is not defined" - Stack Overflow

Topic		Replies	Views
Fresh aggregate for each join like BlockingQueue#drainTo Kafka Streams	12	2159	25 January 2023
Aggregations on Windowed KTables Kafka Streams	2	2871	7 March 2022
Batch processing in kstreams Kafka Streams	9	4759	19 September 2021
Delete row from aggregation table/performance effects if not doing so ksqlDB	1	1070	28 September 2023
NULL values are retuned after JOIN in ksqlDB ksqlDB	1	1327	20 August 2023

Is it safe to assume that KGroupedTable#aggregate subtractor is called before adder?

Related Topics