### Question:
KStreamTransformValues implements ProcessorSupplier<K, V> { @Override public Processor<K, V> get() { return new KStreamTransformValuesProcessor<>(valueTransformerSupplier.get()); } KStreamTransformValues(ValueTransformerSupplier<V, R> valueTransformerSupplier); @Override Processor<K, V> get();  }### Answer:
@Test public void testTransform() { KStreamBuilder builder = new KStreamBuilder(); ValueTransformerSupplier<Number, Integer> valueTransformerSupplier = new ValueTransformerSupplier<Number, Integer>() { public ValueTransformer<Number, Integer> get() { return new ValueTransformer<Number, Integer>() { private int total = 0; @Override public void init(ProcessorContext context) { } @Override public Integer transform(Number value) { total += value.intValue(); return total; } @Override public Integer punctuate(long timestamp) { return null; } @Override public void close() { } }; } }; final int[] expectedKeys = {1, 10, 100, 1000}; KStream<Integer, Integer> stream; MockProcessorSupplier<Integer, Integer> processor = new MockProcessorSupplier<>(); stream = builder.stream(intSerde, intSerde, topicName); stream.transformValues(valueTransformerSupplier).process(processor); driver = new KStreamTestDriver(builder); for (int expectedKey : expectedKeys) { driver.process(topicName, expectedKey, expectedKey * 10); } assertEquals(4, processor.processed.size()); String[] expected = {"1:10", "10:110", "100:1110", "1000:11110"}; for (int i = 0; i < expected.length; i++) { assertEquals(expected[i], processor.processed.get(i)); } }
@Test public void shouldNotAllowValueTransformerToCallInternalProcessorContextMethods() { final KStreamTransformValues<Integer, Integer, Integer> transformValue = new KStreamTransformValues<>(new ValueTransformerSupplier<Integer, Integer>() { @Override public ValueTransformer<Integer, Integer> get() { return new BadValueTransformer(); } }); final Processor transformValueProcessor = transformValue.get(); transformValueProcessor.init(null); try { transformValueProcessor.process(null, 0); fail("should not allow call to context.forward() within ValueTransformer"); } catch (final StreamsException e) { } try { transformValueProcessor.process(null, 1); fail("should not allow call to context.forward() within ValueTransformer"); } catch (final StreamsException e) { } try { transformValueProcessor.process(null, 2); fail("should not allow call to context.forward() within ValueTransformer"); } catch (final StreamsException e) { } try { transformValueProcessor.punctuate(0); fail("should not allow ValueTransformer#puntuate() to return not-null value"); } catch (final StreamsException e) { } }