@SuppressWarnings({"rawtypes", "unchecked"})
    private static <T> Serde<T> prepareSerde(final Serde<T> specificSerde, final SerdeGetter getter, final boolean isKey) {
        final Serde<T> serdeToUse;
        if (specificSerde == null) {
            serdeToUse = (Serde<T>) (isKey ? getter.keySerde() : getter.valueSerde());
        } else {
            serdeToUse = specificSerde;
        }
        if (serdeToUse instanceof WrappingNullableSerde) {
            ((WrappingNullableSerde) serdeToUse).setIfUnset(getter);
        }
        return serdeToUse;
    }
If a component's serdes are Wrapping serdes, then they require a little extra setup
to be fully initialized at run time.
streams/src/main/java/org/apache/kafka/streams/kstream/internals/WrappingNullableUtils.java