public <KEY> Where<KEY> where(KeySelector<T1, KEY> keySelector) {
        requireNonNull(keySelector);
        final TypeInformation<KEY> keyType =
                TypeExtractor.getKeySelectorTypes(keySelector, input1.getType());
        return where(keySelector, keyType);
    }
@Test
    void testDelegateToCoGrouped() {
        Duration lateness = Duration.ofMillis(42L);

        JoinedStreams.WithWindow<String, String, String, TimeWindow> withLateness =
                dataStream1
                        .join(dataStream2)
                        .where(keySelector)
                        .equalTo(keySelector)
                        .window(tsAssigner)
                        .allowedLateness(lateness);

        withLateness.apply(joinFunction, BasicTypeInfo.STRING_TYPE_INFO);

        assertThat(withLateness.getCoGroupedWindowedStream().getAllowedLatenessDuration())
                .hasValue(lateness);
    }