code: protected void emitRecordAndUpdateState(T record, long recordTimestamp, int shardStateIndex, SequenceNumber lastSequenceNumber) {
  		// track per shard watermarks and emit timestamps extracted from the record,
  		// when a watermark assigner was configured.
  		if (periodicWatermarkAssigner != null) {
  			ShardWatermarkState sws = shardWatermarks.get(shardStateIndex);
  			Preconditions.checkNotNull(
  				sws, "shard watermark state initialized in registerNewSubscribedShardState");
  			recordTimestamp =
  				sws.periodicWatermarkAssigner.extractTimestamp(record, sws.lastRecordTimestamp);
  			sws.lastRecordTimestamp = recordTimestamp;
  			sws.lastUpdated = getCurrentTimeMillis();
  		}
  
  		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);
  		}
  	}
docs: Atomic operation to collect a record and update state to the sequence number of the record.
  This method is called by {@link ShardConsumer}s.
  
  @param record the record to collect
  @param recordTimestamp timestamp to attach to the collected record
  @param shardStateIndex index of the shard to update in subscribedShardsState;
  this index should be the returned value from
  {@link KinesisDataFetcher#registerNewSubscribedShardState(KinesisStreamShardState)}, called
  when the shard state was registered.
  @param lastSequenceNumber the last sequence number value to update