class FlinkKafkaConsumerBase extends RichParallelSourceFunction<T> implements 
		CheckpointListener,
		ResultTypeQueryable<T>,
		CheckpointedFunction {

	@Override
	public void snapshotState(FunctionSnapshotContext context) throws Exception {
		if (!running) {
			LOG.debug("snapshotState() called on closed source");
		} else {

			offsetsStateForCheckpoint.clear();

			final AbstractFetcher<?, ?> fetcher = this.kafkaFetcher;
			if (fetcher == null) {
				// the fetcher has not yet been initialized, which means we need to return the
				// originally restored offsets or the assigned partitions

				if (restoreToOffset != null) {

					for (Map.Entry<KafkaTopicPartition, Long> kafkaTopicPartitionLongEntry : restoreToOffset.entrySet()) {
						offsetsStateForCheckpoint.add(
								Tuple2.of(kafkaTopicPartitionLongEntry.getKey(), kafkaTopicPartitionLongEntry.getValue()));
					}
				} else if (subscribedPartitions != null) {
					for (KafkaTopicPartition subscribedPartition : subscribedPartitions) {
						offsetsStateForCheckpoint.add(
								Tuple2.of(subscribedPartition, KafkaTopicPartitionState.OFFSET_NOT_SET));
					}
				}

				// the map cannot be asynchronously updated, because only one checkpoint call can happen
				// on this function at a time: either snapshotState() or notifyCheckpointComplete()
				pendingOffsetsToCommit.put(context.getCheckpointId(), restoreToOffset);
			} else {
				HashMap<KafkaTopicPartition, Long> currentOffsets = fetcher.snapshotCurrentState();

				// the map cannot be asynchronously updated, because only one checkpoint call can happen
				// on this function at a time: either snapshotState() or notifyCheckpointComplete()
				pendingOffsetsToCommit.put(context.getCheckpointId(), currentOffsets);

				for (Map.Entry<KafkaTopicPartition, Long> kafkaTopicPartitionLongEntry : currentOffsets.entrySet()) {
					offsetsStateForCheckpoint.add(
							Tuple2.of(kafkaTopicPartitionLongEntry.getKey(), kafkaTopicPartitionLongEntry.getValue()));
				}
			}

			// truncate the map of pending offsets to commit, to prevent infinite growth
			while (pendingOffsetsToCommit.size() > MAX_NUM_PENDING_CHECKPOINTS) {
				pendingOffsetsToCommit.remove(0);
			}
		}
	}

	public  FlinkKafkaConsumerBase(List<String> topics, KeyedDeserializationSchema<T> deserializer);

}