private void updateWindowAndForward(final Window window,
                                            final ValueAndTimestamp<VAgg> valueAndTime,
                                            final Record<KIn, VIn> record,
                                            final long windowCloseTime) {
            final long windowStart = window.start();
            final long windowEnd = window.end();

            if (windowEnd >= windowCloseTime) {
                // get aggregate from existing window
                final VAgg oldAgg = getValueOrNull(valueAndTime);
                final VAgg newAgg = aggregator.apply(record.key(), record.value(), oldAgg);

                final long newTimestamp = oldAgg == null ? record.timestamp() : Math.max(record.timestamp(), valueAndTime.timestamp());
                windowStore.put(
                    record.key(),
                    ValueAndTimestamp.make(newAgg, newTimestamp),
                    windowStart);
                maybeForwardUpdate(record, window, oldAgg, newAgg, newTimestamp);
            } else {
                final String windowString = "[" + windowStart + "," + windowEnd + "]";
                logSkippedRecordForExpiredWindow(log, record.timestamp(), windowCloseTime, windowString);
            }
        }
Created to handle records where 0 < inputRecordTimestamp < timeDifferenceMs. These records would create
windows with negative start times, which is not supported. Instead, we will put them into the [0, timeDifferenceMs]
window as a "workaround", and we will update or create their right windows as new records come in later
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSlidingWindowAggregate.java