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 ; } }
Register a new subscribed shard state .