@SuppressWarnings("unchecked")
    @Override
    public void configure(final Map<String, ?> configs, final boolean isKey) {
        final String windowedInnerClassSerdeConfig = (String) configs.get(StreamsConfig.WINDOWED_INNER_CLASS_SERDE);

        Serde<T> windowInnerClassSerde = null;

        if (windowedInnerClassSerdeConfig != null) {
            try {
                windowInnerClassSerde = Utils.newInstance(windowedInnerClassSerdeConfig, Serde.class);
            } catch (final ClassNotFoundException e) {
                throw new ConfigException(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, windowedInnerClassSerdeConfig,
                    "Serde class " + windowedInnerClassSerdeConfig + " could not be found.");
            }
        }

        if (inner != null && windowedInnerClassSerdeConfig != null) {
            if (!inner.getClass().getName().equals(windowInnerClassSerde.deserializer().getClass().getName())) {
                throw new IllegalArgumentException("Inner class deserializer set using constructor "
                    + "(" + inner.getClass().getName() + ")" +
                    " is different from the one set in windowed.inner.class.serde config " +
                    "(" + windowInnerClassSerde.deserializer().getClass().getName() + ").");
            }
        } else if (inner == null && windowedInnerClassSerdeConfig == null) {
            throw new IllegalArgumentException("Inner class deserializer should be set either via constructor " +
                "or via the windowed.inner.class.serde config");
        } else if (inner == null)
            inner = windowInnerClassSerde.deserializer();
    }
@Test
    public void shouldThrowErrorIfDeserialisersConflictInConstructorAndConfig() {
        props.put(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, Serdes.ByteArraySerde.class.getName());
        assertThrows(IllegalArgumentException.class, () -> sessionWindowedDeserializer.configure(props, false));
    }