@Override
    public Processor<KIn, VIn, KOut, VOut> get() {
        return new KStreamFlatTransformProcessor<>(transformerSupplier.get());
    }
@Test
    public void shouldGetFlatTransformProcessor() {
        @SuppressWarnings("unchecked")
        final org.apache.kafka.streams.kstream.TransformerSupplier<Number, Number, Iterable<KeyValue<Integer, Integer>>> transformerSupplier =
            mock(org.apache.kafka.streams.kstream.TransformerSupplier.class);
        final KStreamFlatTransform<Number, Number, Integer, Integer> processorSupplier =
            new KStreamFlatTransform<>(transformerSupplier);

        when(transformerSupplier.get()).thenReturn(transformer);

        final Processor<Number, Number, Integer, Integer> processor = processorSupplier.get();

        assertInstanceOf(KStreamFlatTransformProcessor.class, processor);
    }