class ShardConsumer implements Runnable {

	@SuppressWarnings("unchecked")
	@Override
	public void run() {
		try {
			String nextShardItr = getShardIterator(lastSequenceNum);

			long processingStartTimeNanos = System.nanoTime();

			while (isRunning()) {
				if (nextShardItr == null) {
					fetcherRef.updateState(subscribedShardStateIndex, SentinelSequenceNumber.SENTINEL_SHARD_ENDING_SEQUENCE_NUM.get());

					// we can close this consumer thread once we've reached the end of the subscribed shard
					break;
				} else {
					shardMetricsReporter.setMaxNumberOfRecordsPerFetch(maxNumberOfRecordsPerFetch);
					GetRecordsResult getRecordsResult = getRecords(nextShardItr, maxNumberOfRecordsPerFetch);

					List<Record> aggregatedRecords = getRecordsResult.getRecords();
					int numberOfAggregatedRecords = aggregatedRecords.size();
					shardMetricsReporter.setNumberOfAggregatedRecords(numberOfAggregatedRecords);

					// each of the Kinesis records may be aggregated, so we must deaggregate them before proceeding
					List<UserRecord> fetchedRecords = deaggregateRecords(
						aggregatedRecords,
						subscribedShard.getShard().getHashKeyRange().getStartingHashKey(),
						subscribedShard.getShard().getHashKeyRange().getEndingHashKey());

					long recordBatchSizeBytes = 0L;
					for (UserRecord record : fetchedRecords) {
						recordBatchSizeBytes += record.getData().remaining();
						deserializeRecordForCollectionAndUpdateState(record);
					}

					int numberOfDeaggregatedRecords = fetchedRecords.size();
					shardMetricsReporter.setNumberOfDeaggregatedRecords(numberOfDeaggregatedRecords);

					nextShardItr = getRecordsResult.getNextShardIterator();

					long adjustmentEndTimeNanos = adjustRunLoopFrequency(processingStartTimeNanos, System.nanoTime());
					long runLoopTimeNanos = adjustmentEndTimeNanos - processingStartTimeNanos;
					maxNumberOfRecordsPerFetch = adaptRecordsToRead(runLoopTimeNanos, fetchedRecords.size(), recordBatchSizeBytes, maxNumberOfRecordsPerFetch);
					shardMetricsReporter.setRunLoopTimeNanos(runLoopTimeNanos);
					processingStartTimeNanos = adjustmentEndTimeNanos; // for next time through the loop
				}
			}
		} catch (Throwable t) {
			fetcherRef.stopWithError(t);
		}
	}

	public  ShardConsumer(KinesisDataFetcher<T> fetcherRef,
						Integer subscribedShardStateIndex,
						StreamShardHandle subscribedShard,
						SequenceNumber lastSequenceNum,
						KinesisProxyInterface kinesis,
						ShardMetricsReporter shardMetricsReporter);

}