KAFKA-13785: [7/N][Emit final] emit final for sliding window #12135

guozhangwang · 2022-05-07T15:23:27Z

This is a copy PR of #12037.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

guozhangwang · 2022-05-07T16:37:03Z

guozhangwang · 2022-05-07T15:35:51Z

streams/src/main/java/org/apache/kafka/streams/kstream/EmitStrategy.java

    enum StrategyType {
-        ON_WINDOW_CLOSE,
-        ON_WINDOW_UPDATE
+        ON_WINDOW_UPDATE(0, new WindowUpdateStrategy()),


I augmented the enum with a code so that users can translate between the type to the actual strategy easily --- previously in code one has to use a switch to map.

mjsax

Did not go over the testing code yet.

mjsax · 2022-05-10T17:19:54Z

streams/src/main/java/org/apache/kafka/streams/kstream/EmitStrategy.java

+            return this.strategy;
+        }
+
+        StrategyType(final int code, final EmitStrategy strategy) {


Why are you using int here? Seems cleaner to use short ?

This is because Java's default type is int -- i.e. line 36/37 above would take the value 0/1 as int. So we basically need to either do the conversion in each line of 36/37 above, or just do the conversion once here.

I followed our other enums like Errors to do the conversion here.

mjsax · 2022-05-10T17:21:13Z

streams/src/main/java/org/apache/kafka/streams/kstream/EmitStrategy.java

+        static {
+            for (final StrategyType type : StrategyType.values()) {
+                if (TYPE_TO_STRATEGY.put(type.code(), type.strategy()) != null)
+                    throw new IllegalStateException("Code " + type.code() + " for type " +


Never seen anything like this before -- is it best practice to have a guard like this?

I was following the enum Errors as well to add this guard.

mjsax · 2022-05-10T17:28:54Z

.../org/apache/kafka/streams/kstream/internals/AbstractKStreamTimeWindowAggregateProcessor.java

+            );
+            timeTracker.setEmitInterval(emitInterval);
+        } else {
+            tupleForwarder = new TimestampedTupleForwarder<>(


Why do we init the tupleForwarder only for "emit on change" -- don't we also need the forwarder for "emit final" (and just use it differently)?

In the original code, we setup TimestampedTupleForwarder w/ or w/o the TimestampedCacheFlushListener for both cases.

In the emit on final case, the tuple forwarder would not be used (you can see that in the old code, we have a separate constructor for it in emit on final which does not pass in the cache at all) since we always rely on the logic to scan/emit to downstream that does not call the tuple forwarder. I think it's cleaner to not construct the object at all for this case.

EDIT: actually you're right! :) The maybeForward should still be called in emit final case.

mjsax · 2022-05-10T17:29:54Z

.../org/apache/kafka/streams/kstream/internals/AbstractKStreamTimeWindowAggregateProcessor.java

+        while (windowToEmit.hasNext()) {
+            emittedCount++;
+            final KeyValue<Windowed<KIn>, ValueAndTimestamp<VAgg>> kv = windowToEmit.next();
+            tupleForwarder.maybeForward(


Seems we use tupleForwarded for the "emit final" case here, but it seems it was not initialized (cf my commend above on init() method)?

mjsax · 2022-05-10T17:31:32Z

.../org/apache/kafka/streams/kstream/internals/AbstractKStreamTimeWindowAggregateProcessor.java

+                    .withHeaders(record.headers()));
+        }
+        emittedRecordsSensor.record(emittedCount);
+        emitFinalLatencySensor.record(time.milliseconds() - startMs);


What's the definition for this sensor? Wondering about the semantics given the current implemenation?

It measures each time when emit final is triggered, how long it took to scan the store and emit the record for downstream processing

As we do depth-first processing, it includes the time of downstream processing of the emitted record, right? Is this intentional and easy to understand for users?

Yes I was concerned about this too.. but in another thought I think in most cases this processor should be the last of the sub-topologies, plus the main goal is to see if a single record's processing, that including the emitting procedure, could be taking too long to become a problem, hence just measuring the store scan time is not sufficient anyways.

mjsax · 2022-05-10T22:25:50Z

.../org/apache/kafka/streams/kstream/internals/AbstractKStreamTimeWindowAggregateProcessor.java

+        internalProcessorContext.addProcessorMetadataKeyValue(storeName, closeTime);
+    }
+
+    abstract protected void maybeForwardFinalResult(final Record<KIn, VIn> record, final long windowCloseTime);


I though fetchAndEmit would take care of "emit final" -- what is this method about?

The instantiated maybeForwardFinalResult for time / sliding windows has a slight different validation check before calling fetchAndEmit.

mjsax · 2022-05-10T22:30:31Z

.../org/apache/kafka/streams/kstream/internals/AbstractKStreamTimeWindowAggregateProcessor.java

+                .withTimestamp(newTimestamp));
+    }
+
+    protected boolean shouldEmitFinal(final long closeTime) {


It seems this method is called as a first step inside the implementation of maybeForwardFinalResult -- thus, I am wondering if we should not make it private and call inside maybeMeasureEmitFinalLatency ?

Hmm, yes I think that's a good point --- we are double measuring the emitFinalLatencySensor both here and inside the fetchAndEmit, and I think we should only keep the latter. Will update.

mjsax · 2022-05-10T22:31:40Z

.../org/apache/kafka/streams/kstream/internals/AbstractKStreamTimeWindowAggregateProcessor.java

+
+    abstract protected void maybeForwardFinalResult(final Record<KIn, VIn> record, final long windowCloseTime);
+
+    protected void maybeMeasureEmitFinalLatency(final Record<KIn, VIn> record, final long windowCloseTime) {


Nit: Should we make this method final ?

I've removed this function since we already measure emitFinalLatencySensor inside the fetchAndEmit function.

mjsax · 2022-05-10T22:34:42Z

.../src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSlidingWindowAggregate.java

+            }
+
+            final long emitRangeLowerBoundInclusive = lastEmitWindowCloseTime == ConsumerRecord.NO_TIMESTAMP ?
+                0L : lastEmitWindowCloseTime - windows.timeDifferenceMs();


Should this be:

final long emitRangeLowerBoundInclusive = Math.max(0L, lastEmitWindowCloseTime - windows.timeDifferenceMs());

Otherwise, if lastEmitWindowCloseTime < windows.timeDifferenceMs() the result could still be negative?

So far it seems lastEmitWindowCloseTime should always be no smaller than window size in either sliding or time windows, but when there's bugs it's possible that the read value from the processor metadata is small. I will update it accordingly.

mjsax · 2022-05-10T22:40:06Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregate.java

+            // Because we only get here when emitRangeUpperBoundInclusive > 0 which means closeTime > windows.size()
+            // Since we set lastEmitCloseTime to closeTime before storing to processor metadata
+            // lastEmitCloseTime - windows.size() is always > 0
+            // Set emitRangeLowerBoundInclusive to -1L if not set so that when we fetchAll, we fetch from 0L
            final long emitRangeLowerBoundInclusive = lastEmitWindowCloseTime == ConsumerRecord.NO_TIMESTAMP ?
                -1L : lastEmitWindowCloseTime - windows.size();


The code up to here is basically identical, except the use of window.size() vs window. timeDifferenceMs() -- would it be worth to unify?

I've made some refactoring, LMK what do you think.

…12037

mjsax

Did not fully review test code -- it's very time consuming...

Overall LGTM. Few more minor comments. Feel free to merge.

mjsax · 2022-05-13T23:05:55Z

.../org/apache/kafka/streams/kstream/internals/AbstractKStreamTimeWindowAggregateProcessor.java

+        if (emitStrategy.type() == StrategyType.ON_WINDOW_CLOSE) {
+            return;
+        } else if (tupleForwarder == null) {
+            throw new IllegalStateException("Emit strategy type is " + emitStrategy.type() + " but flush listener is not initialized.");


tupleForwarded should never be null?

mjsax · 2022-05-13T23:09:28Z

.../org/apache/kafka/streams/kstream/internals/AbstractKStreamTimeWindowAggregateProcessor.java

+                    .withHeaders(record.headers()));
+        }
+        emittedRecordsSensor.record(emittedCount);
+        emitFinalLatencySensor.record(time.milliseconds() - startMs);


As we do depth-first processing, it includes the time of downstream processing of the emitted record, right? Is this intentional and easy to understand for users?

mjsax · 2022-05-13T23:11:43Z

.../src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSlidingWindowAggregate.java


    private boolean sendOldValues = false;

    public KStreamSlidingWindowAggregate(final SlidingWindows windows,
                                         final String storeName,
                                         final Initializer<VAgg> initializer,
                                         final Aggregator<? super KIn, ? super VIn, VAgg> aggregator) {
+        this(windows, storeName, EmitStrategy.onWindowUpdate(), initializer, aggregator);


Nit: would it be better to not have this constructor with a default emit strategy, but force the caller to pick on explicitly?

I replied in the other comment that this is now only used in cogroup which do not have emit-on-final yet, but I guess we can always just call it explicitly.

mjsax · 2022-05-13T23:12:29Z

.../src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSlidingWindowAggregate.java

-                        recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset(),
+                        recordMetadata.topic(),
+                        recordMetadata.partition(),
+                        recordMetadata.offset(),


Thanks for fixing the formatting!

mjsax · 2022-05-13T23:14:21Z

.../src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSlidingWindowAggregate.java

+        protected long emitRangeUpperBound(final long windowCloseTime) {
+            // Sliding window's start and end timestamps are inclusive, so
+            // we should minus 1 for the inclusive closed window-end upper bound
+            return windowCloseTime - windows.timeDifferenceMs() - 1;


nit: Maybe add a comment and explain why we don't need a guard for a negative result?

I can actually be negative, and we would skip the range fetching in that case.

mjsax · 2022-05-13T23:53:55Z

.../test/java/org/apache/kafka/streams/kstream/internals/KStreamSlidingWindowAggregateTest.java

+                        29)),
+                actual
+            );
+        }
    }

    @Test
    public void testJoin() {


Same question as above: why do we have join()-test in the class?

mjsax · 2022-05-13T23:59:09Z

...va/org/apache/kafka/streams/state/internals/TimeOrderedCachingPersistentWindowStoreTest.java

+import static org.junit.Assert.assertTrue;
+
+@RunWith(Parameterized.class)
+public class TimeOrderedCachingPersistentWindowStoreTest {


Side comment: Might have been better to add this test in a separate PR?

mjsax · 2022-05-14T00:00:31Z

...va/org/apache/kafka/streams/state/internals/TimeOrderedCachingPersistentWindowStoreTest.java

+
+        final RocksDBTimeOrderedWindowStore inner = EasyMock.mock(RocksDBTimeOrderedWindowStore.class);
+        // Nothing happens
+        new TimeOrderedCachingWindowStore(inner, WINDOW_SIZE, SEGMENT_INTERVAL);


Seems redundant to test the happy path?

mjsax · 2022-05-14T00:01:46Z

...va/org/apache/kafka/streams/state/internals/TimeOrderedCachingPersistentWindowStoreTest.java

+
+    @Test
+    public void shouldNotReturnDuplicatesInRanges() {
+        final StreamsBuilder builder = new StreamsBuilder();


Why do we need to test this using StreamsBuilder? Can't we call the store methods directly?

mjsax · 2022-05-14T00:09:24Z

...va/org/apache/kafka/streams/state/internals/TimeOrderedCachingPersistentWindowStoreTest.java

+    }
+
+    @Test
+    public void shouldForwardOldValuesWhenDisabled() {


...WhenEnabled -- what does it refer to? (seems same applies to shouldForwardOldValuesWhenEnabled ?)

It should be shouldNot.. will update.

guozhangwang · 2022-05-14T02:29:08Z

Merged to trunk.

lihaosky and others added 24 commits April 10, 2022 23:46

cache for time ordered window store

5446f3a

comments

8ba3a3e

optimization

a274f1a

checkstyle refactor

74a259b

[Emit final][5/N] emit final for TimeWindowedKStreamImpl

213204c

address comments and bug fix

0b917dc

unit tests

3e40ba9

integration test

505c8c0

comments

16efb96

checkstyle

72d4e13

update metrics

12d59ed

checkstyle

fdd2e8f

checkstyle again...

bd6fbe7

delete log

333e111

update metrics name

36b789d

KAFKA-13772: [Emit final][7/N] emit final sliding window

6958ece

tests

d2f9f60

tests

adc46f6

more tests

f92beeb

refactor

3e0ac08

resolve conflicts and rebase on trunk

18d34e8

unit test

fce8d80

more unit tests

61c1453

more tweaks

4f6aeb3

guozhangwang mentioned this pull request May 7, 2022

KAFKA-13785: [7/N][Emit final] emit final for sliding window #12037

Closed

guozhangwang commented May 7, 2022

View reviewed changes

mjsax reviewed May 10, 2022

View reviewed changes

guozhangwang added 2 commits May 13, 2022 09:39

Merge branch 'trunk' of https://github.com/apache/kafka into KReview-…

898db1c

…12037

github comments

8e9481c

mjsax approved these changes May 14, 2022

View reviewed changes

github comments, and also fix a minor test bug

845953a

guozhangwang merged commit 46efb72 into apache:trunk May 14, 2022


		abstract protected void maybeForwardFinalResult(final Record<KIn, VIn> record, final long windowCloseTime);

		protected void maybeMeasureEmitFinalLatency(final Record<KIn, VIn> record, final long windowCloseTime) {

KAFKA-13785: [7/N][Emit final] emit final for sliding window #12135

KAFKA-13785: [7/N][Emit final] emit final for sliding window #12135

Conversation

guozhangwang commented May 7, 2022

Committer Checklist (excluded from commit message)

guozhangwang commented May 7, 2022

Choose a reason for hiding this comment

mjsax left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang commented May 14, 2022