@Test
@SuppressWarnings("unchecked")
public void testEitherWatermarkExtractor() {

    try {
        new DummyFlinkPulsarSource<String>()
                .assignTimestampsAndWatermarks((AssignerWithPeriodicWatermarks<String>) null);
        fail();
    } catch (NullPointerException ignored) {
    }
    try {
        new DummyFlinkPulsarSource<String>()
                .assignTimestampsAndWatermarks((AssignerWithPunctuatedWatermarks<String>) null);
        fail();
    } catch (NullPointerException ignored) {
    }
    final AssignerWithPeriodicWatermarks<String> periodicAssigner = mock(AssignerWithPeriodicWatermarks.class);
    final AssignerWithPunctuatedWatermarks<String> punctuatedAssigner =
            mock(AssignerWithPunctuatedWatermarks.class);
    DummyFlinkPulsarSource<String> c1 = new DummyFlinkPulsarSource<>();
    c1.assignTimestampsAndWatermarks(periodicAssigner);
    try {
        c1.assignTimestampsAndWatermarks(punctuatedAssigner);
        fail();
    } catch (IllegalStateException ignored) {
    }
    DummyFlinkPulsarSource<String> c2 = new DummyFlinkPulsarSource<>();
    c2.assignTimestampsAndWatermarks(punctuatedAssigner);
    try {
        c2.assignTimestampsAndWatermarks(periodicAssigner);
        fail();
    } catch (IllegalStateException ignored) {
    }
}