Repository: apache/kafka
File: streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamFlatTransformTest.java (Java)

Code before:

    @Test
    public void shouldInitialiseFlatTransformProcessor() {
        transformer.init(context);
        replayAll();

        processor.init(context);

        verifyAll();
    }

    @Test
    public void shouldTransformInputRecordToMultipleOutputRecords() {
        final Iterable<KeyValue<Integer, Integer>> outputRecords = Arrays.asList(
                KeyValue.pair(2, 20),
                KeyValue.pair(3, 30),
                KeyValue.pair(4, 40));
        processor.init(context);
        EasyMock.reset(transformer);

        EasyMock.expect(transformer.transform(inputKey, inputValue)).andReturn(outputRecords);
        for (final KeyValue<Integer, Integer> outputRecord : outputRecords) {
            context.forward(new Record<>(outputRecord.key, outputRecord.value, 0L));
        }
        replayAll();

        processor.process(new Record<>(inputKey, inputValue, 0L));

        verifyAll();
    }

    @Test
    public void shouldAllowEmptyListAsResultOfTransform() {
        processor.init(context);
        EasyMock.reset(transformer);

        EasyMock.expect(transformer.transform(inputKey, inputValue))
            .andReturn(Collections.emptyList());
        replayAll();

        processor.process(new Record<>(inputKey, inputValue, 0L));

        verifyAll();
    }

    @Test
    public void shouldAllowNullAsResultOfTransform() {
        processor.init(context);
        EasyMock.reset(transformer);

        EasyMock.expect(transformer.transform(inputKey, inputValue))

Review comment (refactor):
nit: I would remove this line. I usually try to have a block for the setup, a block for the call under test, and a block for the verifications. Does not always work, but often.
This comment applies also to other places in this PR.

Code after:
    private InOrder inOrder;

    private KStreamFlatTransformProcessor<Number, Number, Integer, Integer> processor;

    @Before
    public void setUp() {
        inputKey = 1;
        inputValue = 10;
        inOrder = inOrder(context);
        processor = new KStreamFlatTransformProcessor<>(transformer);
    }

    @Test
    public void shouldInitialiseFlatTransformProcessor() {
        processor.init(context);

        verify(transformer).init(context);
    }

    @Test
    public void shouldTransformInputRecordToMultipleOutputRecords() {
        final Iterable<KeyValue<Integer, Integer>> outputRecords = Arrays.asList(
                KeyValue.pair(2, 20),
                KeyValue.pair(3, 30),
                KeyValue.pair(4, 40));

        processor.init(context);

        when(transformer.transform(inputKey, inputValue)).thenReturn(outputRecords);

        processor.process(new Record<>(inputKey, inputValue, 0L));

        for (final KeyValue<Integer, Integer> outputRecord : outputRecords) {
            inOrder.verify(context).forward(new Record<>(outputRecord.key, outputRecord.value, 0L));
        }
    }

    @Test
    public void shouldAllowEmptyListAsResultOfTransform() {
        processor.init(context);

        when(transformer.transform(inputKey, inputValue)).thenReturn(Collections.emptyList());

        processor.process(new Record<>(inputKey, inputValue, 0L));

        inOrder.verify(context, never()).forward(ArgumentMatchers.<Record<Integer, Integer>>any());
    }

    @Test
    public void shouldAllowNullAsResultOfTransform() {
        processor.init(context);