class KinesisDataFetcher {

	protected void emitRecordAndUpdateState(T record, long recordTimestamp, int shardStateIndex, SequenceNumber lastSequenceNumber) {
		synchronized (checkpointLock) {
			if (record != null) {
				sourceContext.collectWithTimestamp(record, recordTimestamp);
			} else {
				LOG.warn("Skipping non-deserializable record at sequence number {} of shard {}.",
					lastSequenceNumber,
					subscribedShardsState.get(shardStateIndex).getStreamShardHandle());
			}

			updateState(shardStateIndex, lastSequenceNumber);
		}
	}

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

}