@Override
    public <VO, VR> KStream<K, VR> join(final KStream<K, VO> otherStream,
                                        final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
                                        final JoinWindows windows) {
        return join(otherStream, toValueJoinerWithKey(joiner), windows);
    }
@Test
    public void shouldNotAllowNullTableOnJoinWithGlobalTable() {
        final NullPointerException exception = assertThrows(
            NullPointerException.class,
            () -> testStream.join(null, MockMapper.selectValueMapper(), MockValueJoiner.TOSTRING_JOINER));
        assertThat(exception.getMessage(), equalTo("globalTable can't be null"));
    }