@SuppressWarnings("unchecked")
    public boolean enableSendingOldValues(final boolean forceMaterialization) {
        if (!sendOldValues) {
            if (processorSupplier instanceof KTableSource) {
                final KTableSource<K, ?> source = (KTableSource<K, V>) processorSupplier;
                if (!forceMaterialization && !source.materialized()) {
                    return false;
                }
                source.enableSendingOldValues();
            } else if (processorSupplier instanceof KStreamAggProcessorSupplier) {
                ((KStreamAggProcessorSupplier<?, K, S, V>) processorSupplier).enableSendingOldValues();
            } else if (processorSupplier instanceof KTableProcessorSupplier) {
                final KTableProcessorSupplier<?, ?, ?, ?> tableProcessorSupplier =
                    (KTableProcessorSupplier<?, ?, ?, ?>) processorSupplier;
                if (!tableProcessorSupplier.enableSendingOldValues(forceMaterialization)) {
                    return false;
                }
            }
            sendOldValues = true;
        }
        return true;
    }
The implementation class of {@link KTable}.

@param <K> the key type
@param <S> the source's (parent's) value type
@param <V> the value type
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java