flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/ShardConsumer.java
/**
	 * 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.
	 *
	 * 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
	 */
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
ShardConsumer#getRecords(String, int) may be able to use the correct sequence number to refresh shard
iterators if necessary.

Note that the server-side Kinesis timestamp is attached to the record when collected. When the
user programs uses  TimeCharacteristic#EventTime, this timestamp will be used by default.

@param record record to deserialize and collect
@throws IOException
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 ShardConsumer#getRecords ( String int ) may be able to use the correct sequence number to refresh shard iterators if necessary . Note that the server - side Kinesis timestamp is attached to the record when collected . When the user programs uses TimeCharacteristic#EventTime this timestamp will be used by default . @param record record to deserialize and collect @throws IOException
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;
	}
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 ; }
Deserializes a record for collection, and accordingly updates the shard state in the fetcher.
Deserializes a record for collection and accordingly updates the shard state in the fetcher .