### Question:
FlinkKafkaConsumerBase extends RichParallelSourceFunction<T> implements
		CheckpointListener,
		ResultTypeQueryable<T>,
		CheckpointedFunction { public FlinkKafkaConsumerBase<T> assignTimestampsAndWatermarks(AssignerWithPunctuatedWatermarks<T> assigner) { checkNotNull(assigner); if (this.periodicWatermarkAssigner != null) { throw new IllegalStateException("A periodic watermark emitter has already been set."); } try { ClosureCleaner.clean(assigner, ExecutionConfig.ClosureCleanerLevel.RECURSIVE, true); this.punctuatedWatermarkAssigner = new SerializedValue<>(assigner); return this; } catch (Exception e) { throw new IllegalArgumentException("The given assigner is not serializable", e); } } FlinkKafkaConsumerBase(
			List<String> topics,
			Pattern topicPattern,
			KafkaDeserializationSchema<T> deserializer,
			long discoveryIntervalMillis,
			boolean useMetrics); FlinkKafkaConsumerBase<T> assignTimestampsAndWatermarks(AssignerWithPunctuatedWatermarks<T> assigner); FlinkKafkaConsumerBase<T> assignTimestampsAndWatermarks(AssignerWithPeriodicWatermarks<T> assigner); FlinkKafkaConsumerBase<T> setCommitOffsetsOnCheckpoints(boolean commitOnCheckpoints); FlinkKafkaConsumerBase<T> setStartFromEarliest(); FlinkKafkaConsumerBase<T> setStartFromLatest(); FlinkKafkaConsumerBase<T> setStartFromGroupOffsets(); FlinkKafkaConsumerBase<T> setStartFromSpecificOffsets(Map<KafkaTopicPartition, Long> specificStartupOffsets); FlinkKafkaConsumerBase<T> disableFilterRestoredPartitionsWithSubscribedTopics(); @Override void open(Configuration configuration); @Override void run(SourceContext<T> sourceContext); @Override void cancel(); @Override void close(); @Override final void initializeState(FunctionInitializationContext context); @Override final void snapshotState(FunctionSnapshotContext context); @Override final void notifyCheckpointComplete(long checkpointId); @Override TypeInformation<T> getProducedType(); static final int MAX_NUM_PENDING_CHECKPOINTS; static final long PARTITION_DISCOVERY_DISABLED; static final String KEY_DISABLE_METRICS; static final String KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS; }### Answer:
@Test @SuppressWarnings("unchecked") public void testEitherWatermarkExtractor() { try { new DummyFlinkKafkaConsumer<String>().assignTimestampsAndWatermarks((AssignerWithPeriodicWatermarks<String>) null); fail(); } catch (NullPointerException ignored) {} try { new DummyFlinkKafkaConsumer<String>().assignTimestampsAndWatermarks((AssignerWithPunctuatedWatermarks<String>) null); fail(); } catch (NullPointerException ignored) {} final AssignerWithPeriodicWatermarks<String> periodicAssigner = mock(AssignerWithPeriodicWatermarks.class); final AssignerWithPunctuatedWatermarks<String> punctuatedAssigner = mock(AssignerWithPunctuatedWatermarks.class); DummyFlinkKafkaConsumer<String> c1 = new DummyFlinkKafkaConsumer<>(); c1.assignTimestampsAndWatermarks(periodicAssigner); try { c1.assignTimestampsAndWatermarks(punctuatedAssigner); fail(); } catch (IllegalStateException ignored) {} DummyFlinkKafkaConsumer<String> c2 = new DummyFlinkKafkaConsumer<>(); c2.assignTimestampsAndWatermarks(punctuatedAssigner); try { c2.assignTimestampsAndWatermarks(periodicAssigner); fail(); } catch (IllegalStateException ignored) {} }