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

Code before:
        replayAll();

        final TransformerSupplierAdapter<String, String, Integer, Integer> adapter =
            new TransformerSupplierAdapter<>(transformerSupplier);
        final Transformer<String, String, Iterable<KeyValue<Integer, Integer>>> adaptedTransformer = adapter.get();
        final Iterator<KeyValue<Integer, Integer>> iterator = adaptedTransformer.transform(key, value).iterator();

        verifyAll();
        assertThat(iterator.hasNext(), equalTo(true));
        iterator.next();
        assertThat(iterator.hasNext(), equalTo(false));
    }

    @Test
    public void shouldCallTransformOfAdapteeTransformerAndReturnEmptyIterable() {
        EasyMock.expect(transformerSupplier.get()).andReturn(transformer);
        EasyMock.expect(transformer.transform(key, value)).andReturn(null);
        replayAll();

        final TransformerSupplierAdapter<String, String, Integer, Integer> adapter =
            new TransformerSupplierAdapter<>(transformerSupplier);
        final Transformer<String, String, Iterable<KeyValue<Integer, Integer>>> adaptedTransformer = adapter.get();
        final Iterator<KeyValue<Integer, Integer>> iterator = adaptedTransformer.transform(key, value).iterator();

        verifyAll();
        assertThat(iterator.hasNext(), equalTo(false));
    }

    @Test
    public void shouldAlwaysGetNewAdapterTransformer() {
        final Transformer<String, String, KeyValue<Integer, Integer>> transformer1 = mock(Transformer.class);
        final Transformer<String, String, KeyValue<Integer, Integer>> transformer2 = mock(Transformer.class);
        final Transformer<String, String, KeyValue<Integer, Integer>> transformer3 = mock(Transformer.class);
        EasyMock.expect(transformerSupplier.get()).andReturn(transformer1);
        transformer1.init(context);
        EasyMock.expect(transformerSupplier.get()).andReturn(transformer2);
        transformer2.init(context);
        EasyMock.expect(transformerSupplier.get()).andReturn(transformer3);
        transformer3.init(context);
        replayAll();

        final TransformerSupplierAdapter<String, String, Integer, Integer> adapter =
            new TransformerSupplierAdapter<>(transformerSupplier);
        final Transformer<String, String, Iterable<KeyValue<Integer, Integer>>> adapterTransformer1 = adapter.get();
        adapterTransformer1.init(context);
        final Transformer<String, String, Iterable<KeyValue<Integer, Integer>>> adapterTransformer2 = adapter.get();
        adapterTransformer2.init(context);
        final Transformer<String, String, Iterable<KeyValue<Integer, Integer>>> adapterTransformer3 = adapter.get();
        adapterTransformer3.init(context);

        verifyAll();

Review comment (suggestion):
nit: see my comment above about avoiding suppressing warnings. Maybe you can just rename the global `transformer` to `transformer1` and then move `transformer2` and `transformer3` to the object fields.

Code after:
        assertThat(iterator.hasNext(), equalTo(true));
        iterator.next();
        assertThat(iterator.hasNext(), equalTo(false));
    }

    @Test
    public void shouldCallTransformOfAdapteeTransformerAndReturnEmptyIterable() {
        when(transformerSupplier.get()).thenReturn(transformer);
        when(transformer.transform(key, value)).thenReturn(null);

        final TransformerSupplierAdapter<String, String, Integer, Integer> adapter =
            new TransformerSupplierAdapter<>(transformerSupplier);
        final Transformer<String, String, Iterable<KeyValue<Integer, Integer>>> adaptedTransformer = adapter.get();
        final Iterator<KeyValue<Integer, Integer>> iterator = adaptedTransformer.transform(key, value).iterator();

        assertThat(iterator.hasNext(), equalTo(false));
    }

    @Test
    public void shouldAlwaysGetNewAdapterTransformer() {
        @SuppressWarnings("unchecked")
        final Transformer<String, String, KeyValue<Integer, Integer>> transformer1 = mock(Transformer.class);
        @SuppressWarnings("unchecked")
        final Transformer<String, String, KeyValue<Integer, Integer>> transformer2 = mock(Transformer.class);
        @SuppressWarnings("unchecked")
        final Transformer<String, String, KeyValue<Integer, Integer>> transformer3 = mock(Transformer.class);
        when(transformerSupplier.get()).thenReturn(transformer1).thenReturn(transformer2).thenReturn(transformer3);

        final TransformerSupplierAdapter<String, String, Integer, Integer> adapter =
            new TransformerSupplierAdapter<>(transformerSupplier);
        final Transformer<String, String, Iterable<KeyValue<Integer, Integer>>> adapterTransformer1 = adapter.get();
        adapterTransformer1.init(context);
        final Transformer<String, String, Iterable<KeyValue<Integer, Integer>>> adapterTransformer2 = adapter.get();
        adapterTransformer2.init(context);
        final Transformer<String, String, Iterable<KeyValue<Integer, Integer>>> adapterTransformer3 = adapter.get();
        adapterTransformer3.init(context);

        assertThat(adapterTransformer1, not(sameInstance(adapterTransformer2)));
        assertThat(adapterTransformer2, not(sameInstance(adapterTransformer3)));
        assertThat(adapterTransformer3, not(sameInstance(adapterTransformer1)));
        verify(transformer1).init(context);
        verify(transformer2).init(context);
        verify(transformer3).init(context);
    }

}