@SuppressWarnings("deprecation")
    static <K, V, VR> ValueTransformerWithKeySupplier<K, V, VR> toValueTransformerWithKeySupplier(
        final org.apache.kafka.streams.kstream.ValueTransformerSupplier<V, VR> valueTransformerSupplier) {
        Objects.requireNonNull(valueTransformerSupplier, "valueTransformerSupplier can't be null");
        ApiUtils.checkSupplier(valueTransformerSupplier);
        return new ValueTransformerWithKeySupplier<K, V, VR>() {
            @Override
            public ValueTransformerWithKey<K, V, VR> get() {
                final org.apache.kafka.streams.kstream.ValueTransformer<V, VR> valueTransformer = valueTransformerSupplier.get();
                return new ValueTransformerWithKey<K, V, VR>() {
                    @Override
                    public void init(final ProcessorContext context) {
                        valueTransformer.init(context);
                    }

                    @Override
                    public VR transform(final K readOnlyKey, final V value) {
                        return valueTransformer.transform(value);
                    }

                    @Override
                    public void close() {
                        valueTransformer.close();
                    }
                };
            }

            @Override
            public Set<StoreBuilder<?>> stores() {
                return valueTransformerSupplier.stores();
            }
        };
    }
@SuppressWarnings("deprecation")
    @Test
    public void testToInternalValueTransformerSupplierSuppliesNewTransformers() {
        final org.apache.kafka.streams.kstream.ValueTransformerSupplier<?, ?> valueTransformerSupplier =
            mock(org.apache.kafka.streams.kstream.ValueTransformerSupplier.class);
        when(valueTransformerSupplier.get())
            .thenReturn(new NoopValueTransformer<>())
            .thenReturn(new NoopValueTransformer<>());
        final ValueTransformerWithKeySupplier<?, ?, ?> valueTransformerWithKeySupplier =
            AbstractStream.toValueTransformerWithKeySupplier(valueTransformerSupplier);
        valueTransformerWithKeySupplier.get();
        valueTransformerWithKeySupplier.get();
        valueTransformerWithKeySupplier.get();
    }