ableegoldman commented on a change in pull request #9157: URL: https://github.com/apache/kafka/pull/9157#discussion_r478768335
########## File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSlidingWindowAggregate.java ########## @@ -0,0 +1,386 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.streams.kstream.internals; + +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.metrics.Sensor; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.kstream.Aggregator; +import org.apache.kafka.streams.kstream.Initializer; +import org.apache.kafka.streams.kstream.Window; +import org.apache.kafka.streams.kstream.Windowed; +import org.apache.kafka.streams.kstream.SlidingWindows; +import org.apache.kafka.streams.processor.AbstractProcessor; +import org.apache.kafka.streams.processor.Processor; +import org.apache.kafka.streams.processor.ProcessorContext; +import org.apache.kafka.streams.processor.internals.InternalProcessorContext; +import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl; +import org.apache.kafka.streams.state.KeyValueIterator; +import org.apache.kafka.streams.state.TimestampedWindowStore; +import org.apache.kafka.streams.state.ValueAndTimestamp; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import java.util.HashSet; +import java.util.Set; + +import static org.apache.kafka.streams.processor.internals.metrics.TaskMetrics.droppedRecordsSensorOrLateRecordDropSensor; +import static org.apache.kafka.streams.processor.internals.metrics.TaskMetrics.droppedRecordsSensorOrSkippedRecordsSensor; +import static org.apache.kafka.streams.state.ValueAndTimestamp.getValueOrNull; + +public class KStreamSlidingWindowAggregate<K, V, Agg> implements KStreamAggProcessorSupplier<K, Windowed<K>, V, Agg> { + private final Logger log = LoggerFactory.getLogger(getClass()); + + private final String storeName; + private final SlidingWindows windows; + private final Initializer<Agg> initializer; + private final Aggregator<? super K, ? super V, Agg> aggregator; + + private boolean sendOldValues = false; + + public KStreamSlidingWindowAggregate(final SlidingWindows windows, + final String storeName, + final Initializer<Agg> initializer, + final Aggregator<? super K, ? super V, Agg> aggregator) { + this.windows = windows; + this.storeName = storeName; + this.initializer = initializer; + this.aggregator = aggregator; + } + + @Override + public Processor<K, V> get() { + return new KStreamSlidingWindowAggregateProcessor(); + } + + public SlidingWindows windows() { + return windows; + } + + @Override + public void enableSendingOldValues() { + sendOldValues = true; + } + + private class KStreamSlidingWindowAggregateProcessor extends AbstractProcessor<K, V> { + private TimestampedWindowStore<K, Agg> windowStore; + private TimestampedTupleForwarder<Windowed<K>, Agg> tupleForwarder; + private StreamsMetricsImpl metrics; + private InternalProcessorContext internalProcessorContext; + private Sensor lateRecordDropSensor; + private Sensor droppedRecordsSensor; + private long observedStreamTime = ConsumerRecord.NO_TIMESTAMP; + + @SuppressWarnings("unchecked") + @Override + public void init(final ProcessorContext context) { + super.init(context); + internalProcessorContext = (InternalProcessorContext) context; + metrics = internalProcessorContext.metrics(); + final String threadId = Thread.currentThread().getName(); + lateRecordDropSensor = droppedRecordsSensorOrLateRecordDropSensor( + threadId, + context.taskId().toString(), + internalProcessorContext.currentNode().name(), + metrics + ); + droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(threadId, context.taskId().toString(), metrics); + windowStore = (TimestampedWindowStore<K, Agg>) context.getStateStore(storeName); + tupleForwarder = new TimestampedTupleForwarder<>( + windowStore, + context, + new TimestampedCacheFlushListener<>(context), + sendOldValues); + } + + @Override + public void process(final K key, final V value) { + if (key == null || value == null) { + log.warn( + "Skipping record due to null key or value. value=[{}] topic=[{}] partition=[{}] offset=[{}]", + value, context().topic(), context().partition(), context().offset() + ); + droppedRecordsSensor.record(); + return; + } + + final long timestamp = context().timestamp(); + processInOrder(key, value, timestamp); + } + + public void processInOrder(final K key, final V value, final long timestamp) { + + observedStreamTime = Math.max(observedStreamTime, timestamp); + final long closeTime = observedStreamTime - windows.gracePeriodMs(); + + if (timestamp < windows.timeDifferenceMs()) { + processEarly(key, value, timestamp, closeTime); + return; + } + + final Set<Long> windowStartTimes = new HashSet<>(); + + // aggregate that will go in the current record’s left/right window (if needed) + ValueAndTimestamp<Agg> leftWinAgg = null; + ValueAndTimestamp<Agg> rightWinAgg = null; + + //if current record's left/right windows already exist + boolean leftWinAlreadyCreated = false; + boolean rightWinAlreadyCreated = false; + + // keep the left type window closest to the record + KeyValue<Windowed<K>, ValueAndTimestamp<Agg>> latestLeftTypeWindow = null; + try ( + final KeyValueIterator<Windowed<K>, ValueAndTimestamp<Agg>> iterator = windowStore.fetch( + key, + key, + Math.max(0, timestamp - 2 * windows.timeDifferenceMs()), + // to catch the current record's right window, if it exists, without more calls to the store + timestamp + 1) + ) { + KeyValue<Windowed<K>, ValueAndTimestamp<Agg>> next; + while (iterator.hasNext()) { + next = iterator.next(); + windowStartTimes.add(next.key.window().start()); + final long startTime = next.key.window().start(); + final long endTime = startTime + windows.timeDifferenceMs(); + + if (endTime < timestamp) { + leftWinAgg = next.value; + // store the combined window (window from [0,timeDifference] that contains any early records) + // if it is found so that a right window can be created for the combined window's max record, as needed + if (isLeftWindow(next) || endTime == windows.timeDifferenceMs()) { + latestLeftTypeWindow = next; + } + } else if (endTime == timestamp) { + leftWinAlreadyCreated = true; + // if current record's left window is the combined window, need to check later if there is a + // record that needs a right window within the combined window + if (endTime == windows.timeDifferenceMs()) { + latestLeftTypeWindow = next; + } + putAndForward(next.key.window(), next.value, key, value, closeTime, timestamp); + } else if (endTime > timestamp && startTime <= timestamp) { + rightWinAgg = next.value; + putAndForward(next.key.window(), next.value, key, value, closeTime, timestamp); + } else { + rightWinAlreadyCreated = true; + } + } + } + //create right window for previous record + if (latestLeftTypeWindow != null) { + final long previousWindowEnd = latestLeftTypeWindow.key.window().end(); + final long rightWinStart = previousWindowEnd == windows.timeDifferenceMs() ? latestLeftTypeWindow.value.timestamp() + 1 : previousWindowEnd + 1; + if (rightWindowNecessaryAndPossible(windowStartTimes, rightWinStart, timestamp)) { + final TimeWindow window = new TimeWindow(rightWinStart, rightWinStart + windows.timeDifferenceMs()); + final ValueAndTimestamp<Agg> valueAndTime = ValueAndTimestamp.make(initializer.apply(), timestamp); + putAndForward(window, valueAndTime, key, value, closeTime, timestamp); + } + } + + //create left window for new record + if (!leftWinAlreadyCreated) { + final ValueAndTimestamp<Agg> valueAndTime; + // if there's a right window that the new record could create && max record falls within left window --> new record's left window is not empty + if (latestLeftTypeWindow != null && recordFallsWithinCurrentRecordsLeftWindow(latestLeftTypeWindow.value.timestamp(), timestamp)) { + valueAndTime = ValueAndTimestamp.make(leftWinAgg.value(), timestamp); + } else { + valueAndTime = ValueAndTimestamp.make(initializer.apply(), timestamp); + } + final TimeWindow window = new TimeWindow(timestamp - windows.timeDifferenceMs(), timestamp); + putAndForward(window, valueAndTime, key, value, closeTime, timestamp); + } + //create right window for new record + if (!rightWinAlreadyCreated && rightWindowIsNotEmpty(rightWinAgg, timestamp)) { + final TimeWindow window = new TimeWindow(timestamp + 1, timestamp + 1 + windows.timeDifferenceMs()); + final ValueAndTimestamp<Agg> valueAndTime = ValueAndTimestamp.make(getValueOrNull(rightWinAgg), Math.max(rightWinAgg.timestamp(), timestamp)); + putAndForward(window, valueAndTime, key, value, closeTime, timestamp); + } + } + + /** + * Created to handle records that have a timestamp > 0 but < timeDifference. These records would create + * windows with negative start times, which is not supported. Instead, they will fall within the [0, timeDifference] + * window, and we will update their right windows as new records come in later + */ + private void processEarly(final K key, final V value, final long timestamp, final long closeTime) { + ValueAndTimestamp<Agg> rightWinAgg = null; + //window from [0,timeDifference] that holds all early records + KeyValue<Windowed<K>, ValueAndTimestamp<Agg>> combinedWindow = null; + boolean rightWinAlreadyCreated = false; + final Set<Long> windowStartTimes = new HashSet<>(); + + try ( + final KeyValueIterator<Windowed<K>, ValueAndTimestamp<Agg>> iterator = windowStore.fetch( + key, + key, + Math.max(0, timestamp - 2 * windows.timeDifferenceMs()), + // to catch the current record's right window, if it exists, without more calls to the store + timestamp + 1) + ) { + KeyValue<Windowed<K>, ValueAndTimestamp<Agg>> next; + while (iterator.hasNext()) { + next = iterator.next(); + windowStartTimes.add(next.key.window().start()); + final long startTime = next.key.window().start(); + final long endTime = startTime + windows.timeDifferenceMs(); + + if (startTime == 0) { + combinedWindow = next; + } else if (startTime <= timestamp) { + rightWinAgg = next.value; + putAndForward(next.key.window(), next.value, key, value, closeTime, timestamp); + } else if (startTime == timestamp + 1) { + rightWinAlreadyCreated = true; + } + } + } + + // if there wasn't a right window agg found and we need a right window for our new record, + // the current aggregate in the combined window will go in the new record's right window + if (rightWinAgg == null && combinedWindow != null && combinedWindow.value.timestamp() > timestamp) { + rightWinAgg = combinedWindow.value; + } + + if (combinedWindow == null) { + final TimeWindow window = new TimeWindow(0, windows.timeDifferenceMs()); + final ValueAndTimestamp<Agg> valueAndTime = ValueAndTimestamp.make(initializer.apply(), timestamp); + putAndForward(window, valueAndTime, key, value, closeTime, timestamp); + + } else { + //create the right window for the combined window's max record before the current record was added + final long maxRightWindowStart = combinedWindow.value.timestamp() + 1; Review comment: This is also kind of unclear (what is a max right window?), but I get that we can't call it `previousRecordRightWindow` since we don't know that it is a previous record or not at this point. I think yet again, just keeping track of the previous record's timestamp as we iterate through the windows, will be the most clear; if `previousRecordTimestamp` is still null by this point, we know right away that we don't have to create a previous right window. And then we can actually drop the `rightWindowNecessaryAndPossible` check altogether, since we know the current record has to be in range of the right window of the previous record (since we're in `processEarly`). The one exception is if the previous record and current record are on the same timestamp, so we can actually skip the previous right window creation if `previousRecordTimestamp ` is `null` OR equal to `timestamp` ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
