### Question:
FlinkKafkaConsumerBase extends RichParallelSourceFunction<T> implements
		CheckpointListener,
		ResultTypeQueryable<T>,
		CheckpointedFunction { @VisibleForTesting OffsetCommitMode getOffsetCommitMode() { return offsetCommitMode; } 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 public void testConfigureOnCheckpointsCommitMode() throws Exception { @SuppressWarnings("unchecked") final DummyFlinkKafkaConsumer<String> consumer = new DummyFlinkKafkaConsumer<>(true); setupConsumer( consumer, false, null, true, 0, 1); assertEquals(OffsetCommitMode.ON_CHECKPOINTS, consumer.getOffsetCommitMode()); }
@Test public void testConfigureAutoCommitMode() throws Exception { @SuppressWarnings("unchecked") final DummyFlinkKafkaConsumer<String> consumer = new DummyFlinkKafkaConsumer<>(true); setupConsumer(consumer); assertEquals(OffsetCommitMode.KAFKA_PERIODIC, consumer.getOffsetCommitMode()); }
@Test public void testConfigureDisableOffsetCommitWithoutCheckpointing() throws Exception { @SuppressWarnings("unchecked") final DummyFlinkKafkaConsumer<String> consumer = new DummyFlinkKafkaConsumer<>(false); setupConsumer(consumer); assertEquals(OffsetCommitMode.DISABLED, consumer.getOffsetCommitMode()); }