code: private void deserializeRecordForCollectionAndUpdateState(UserRecord record)
  		throws IOException {
  		ByteBuffer recordData = record.getData();
  
  		byte[] dataBytes = new byte[recordData.remaining()];
  		recordData.get(dataBytes);
  
  		final long approxArrivalTimestamp = record.getApproximateArrivalTimestamp().getTime();
  
  		final T value = deserializer.deserialize(
  			dataBytes,
  			record.getPartitionKey(),
  			record.getSequenceNumber(),
  			approxArrivalTimestamp,
  			subscribedShard.getStreamName(),
  			subscribedShard.getShard().getShardId());
  
  		SequenceNumber collectedSequenceNumber = (record.isAggregated())
  			? new SequenceNumber(record.getSequenceNumber(), record.getSubSequenceNumber())
  			: new SequenceNumber(record.getSequenceNumber());
  
  		fetcherRef.emitRecordAndUpdateState(
  			value,
  			approxArrivalTimestamp,
  			subscribedShardStateIndex,
  			collectedSequenceNumber);
  
  		lastSequenceNum = collectedSequenceNumber;
  	}
docs: Deserializes a record for collection, and accordingly updates the shard state in the fetcher. The last
  successfully collected sequence number in this shard consumer is also updated so that
  {@link ShardConsumer#getRecords(String, int)} may be able to use the correct sequence number to refresh shard
  iterators if necessary.
  
  <p>Note that the server-side Kinesis timestamp is attached to the record when collected. When the
  user programs uses {@link TimeCharacteristic#EventTime}, this timestamp will be used by default.
  
  @param record record to deserialize and collect
  @throws IOException