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

            final boolean isLatest = record.value().isLatest;
            if (useVersionedSemantics && !isLatest) {
                // skip out-of-order records when aggregating a versioned table, since the
                // aggregate should include latest-by-timestamp records only. as an optimization,
                // do not forward the out-of-order record downstream to the repartition topic either.
                return;
            }

            // if the value is null, we do not need to forward its selected key-value further
            final KeyValue<? extends K1, ? extends V1> newPair = record.value().newValue == null ? null :
                mapper.apply(record.key(), record.value().newValue);
            final KeyValue<? extends K1, ? extends V1> oldPair = record.value().oldValue == null ? null :
                mapper.apply(record.key(), record.value().oldValue);

            // if the selected repartition key or value is null, skip
            // forward oldPair first, to be consistent with reduce and aggregate
            final boolean oldPairNotNull = oldPair != null && oldPair.key != null && oldPair.value != null;
            final boolean newPairNotNull = newPair != null && newPair.key != null && newPair.value != null;
            if (isNotUpgrade && oldPairNotNull && newPairNotNull && oldPair.key.equals(newPair.key)) {
                context().forward(record.withKey(oldPair.key).withValue(new Change<>(newPair.value, oldPair.value, isLatest)));
            } else {
                if (oldPairNotNull) {
                    context().forward(record.withKey(oldPair.key).withValue(new Change<>(null, oldPair.value, isLatest)));
                }

                if (newPairNotNull) {
                    context().forward(record.withKey(newPair.key).withValue(new Change<>(newPair.value, null, isLatest)));
                }
            }

        }
@throws StreamsException if key is null
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableRepartitionMap.java