@Override
        public void process(final Record<K, Change<V>> record) {
            // the keys should never be null
            if (record.key() == null) {
                throw new StreamsException("Record key for KTable reduce operator with state " + storeName + " should not be null.");
            }

            final ValueAndTimestamp<V> oldAggAndTimestamp = store.get(record.key());
            final V oldAgg = getValueOrNull(oldAggAndTimestamp);
            final V intermediateAgg;
            long newTimestamp;

            // first try to remove the old value
            if (record.value().oldValue != null && oldAgg != null) {
                intermediateAgg = removeReducer.apply(oldAgg, record.value().oldValue);
                newTimestamp = Math.max(record.timestamp(), oldAggAndTimestamp.timestamp());
            } else {
                intermediateAgg = oldAgg;
                newTimestamp = record.timestamp();
            }

            // then try to add the new value
            final V newAgg;
            if (record.value().newValue != null) {
                if (intermediateAgg == null) {
                    newAgg = record.value().newValue;
                } else {
                    newAgg = addReducer.apply(intermediateAgg, record.value().newValue);
                    newTimestamp = Math.max(record.timestamp(), oldAggAndTimestamp.timestamp());
                }
            } else {
                newAgg = intermediateAgg;
            }

            // update the store with the new value
            final long putReturnCode = store.put(record.key(), newAgg, newTimestamp);
            // if not put to store, do not forward downstream either
            if (putReturnCode != PUT_RETURN_CODE_NOT_PUT) {
                tupleForwarder.maybeForward(
                    record.withValue(new Change<>(newAgg, sendOldValues ? oldAgg : null, putReturnCode == PUT_RETURN_CODE_IS_LATEST))
                        .withTimestamp(newTimestamp));
            }
        }
@throws StreamsException if key is null
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableReduce.java