@SuppressWarnings({"rawtypes", "unchecked"})
    public static <T> void initNullableSerializer(final Serializer<T> specificSerializer, final SerdeGetter getter) {
        if (specificSerializer instanceof WrappingNullableSerializer) {
            ((WrappingNullableSerializer) specificSerializer).setIfUnset(getter);
        }
    }
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