code: public int registerNewSubscribedShardState(KinesisStreamShardState newSubscribedShardState) {
  		synchronized (checkpointLock) {
  			subscribedShardsState.add(newSubscribedShardState);
  
  			// If a registered shard has initial state that is not SENTINEL_SHARD_ENDING_SEQUENCE_NUM (will be the case
  			// if the consumer had already finished reading a shard before we failed and restored), we determine that
  			// this subtask has a new active shard
  			if (!newSubscribedShardState.getLastProcessedSequenceNum().equals(SentinelSequenceNumber.SENTINEL_SHARD_ENDING_SEQUENCE_NUM.get())) {
  				this.numberOfActiveShards.incrementAndGet();
  			}
  
  			int shardStateIndex = subscribedShardsState.size() - 1;
  
  			// track all discovered shards for watermark determination
  			ShardWatermarkState sws = shardWatermarks.get(shardStateIndex);
  			if (sws == null) {
  				sws = new ShardWatermarkState();
  				try {
  					sws.periodicWatermarkAssigner = InstantiationUtil.clone(periodicWatermarkAssigner);
  				} catch (Exception e) {
  					throw new RuntimeException("Failed to instantiate new WatermarkAssigner", e);
  				}
  				sws.lastUpdated = getCurrentTimeMillis();
  				sws.lastRecordTimestamp = Long.MIN_VALUE;
  				shardWatermarks.put(shardStateIndex, sws);
  			}
  
  			return shardStateIndex;
  		}
  	}
docs: Register a new subscribed shard state.
  
  @param newSubscribedShardState the new shard state that this fetcher is to be subscribed to