class KinesisDataFetcher {

	protected void emitRecordAndUpdateState(T record, long recordTimestamp, int shardStateIndex, SequenceNumber lastSequenceNumber) {
		ShardWatermarkState sws = shardWatermarks.get(shardStateIndex);
		Preconditions.checkNotNull(
			sws, "shard watermark state initialized in registerNewSubscribedShardState");
		Watermark watermark = null;
		if (sws.periodicWatermarkAssigner != null) {
			recordTimestamp =
				sws.periodicWatermarkAssigner.extractTimestamp(record, sws.lastRecordTimestamp);
			// track watermark per record since extractTimestamp has side effect
			watermark = sws.periodicWatermarkAssigner.getCurrentWatermark();
		}
		sws.lastRecordTimestamp = recordTimestamp;
		sws.lastUpdated = getCurrentTimeMillis();

		RecordWrapper<T> recordWrapper = new RecordWrapper<>(record, recordTimestamp);
		recordWrapper.shardStateIndex = shardStateIndex;
		recordWrapper.lastSequenceNumber = lastSequenceNumber;
		recordWrapper.watermark = watermark;
		try {
			sws.emitQueue.put(recordWrapper);
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}
	}

	public  KinesisDataFetcher(List<String> streams,
							SourceFunction.SourceContext<T> sourceContext,
							RuntimeContext runtimeContext,
							Properties configProps,
							KinesisDeserializationSchema<T> deserializationSchema,
							KinesisShardAssigner shardAssigner,
							AssignerWithPeriodicWatermarks<T> periodicWatermarkAssigner,
							WatermarkTracker watermarkTracker);
	@VisibleForTesting protected  KinesisDataFetcher(List<String> streams,
								SourceFunction.SourceContext<T> sourceContext,
								Object checkpointLock,
								RuntimeContext runtimeContext,
								Properties configProps,
								KinesisDeserializationSchema<T> deserializationSchema,
								KinesisShardAssigner shardAssigner,
								AssignerWithPeriodicWatermarks<T> periodicWatermarkAssigner,
								WatermarkTracker watermarkTracker,
								AtomicReference<Throwable> error,
								List<KinesisStreamShardState> subscribedShardsState,
								HashMap<String, String> subscribedStreamsToLastDiscoveredShardIds,
								FlinkKinesisProxyFactory kinesisProxyFactory);

}