library/src/main/java/com/kdgregory/aws/utils/kinesis/KinesisReader.java
/**
     *  Reloads shard iterators based on current offsets / default iterator type. This
     *  is called when the reader is first used, as well as for expired iterators. If
     *  unable to retrieve shards, silently fails (will retrieve on next iterator).
     */
Reloads shard iterators based on current offsets / default iterator type. This
is called when the reader is first used, as well as for expired iterators. If
unable to retrieve shards, silently fails (will retrieve on next iterator).
Reloads shard iterators based on current offsets / default iterator type . This is called when the reader is first used as well as for expired iterators . If unable to retrieve shards silently fails ( will retrieve on next iterator ) .
private void retrieveInitialIterators()
    {
        if (logger.isDebugEnabled())
        {
            logger.debug("retrieveInitialIterators: stream = " + streamName + ", saved sequence numbers = " + offsets);
        }
        shardIterators.clear();
        long timeoutAt = System.currentTimeMillis() + timeout;
        describeShards();
        if (shards == null)
            return;
        for (String shardId : new ArrayList<String>(offsets.keySet()))
        {
            if (! shardsById.containsKey(shardId))
            {
                offsets.remove(shardId);
            }
        }
        Map<String,ShardIteratorType> itxTypes = new HashMap<String,ShardIteratorType>();
        if (! constructMixedIteratorTree(shardsByParentId.get(null), itxTypes))
        {
            populateDefaultIterators(itxTypes);
        }
        if (logger.isDebugEnabled())
        {
            logger.debug("retrieveInitialIterators: stream = " + streamName + ", iterator types = " + itxTypes);
        }
        retrieveIterators(itxTypes, timeoutAt);
    }
private void retrieveInitialIterators ( ) { if ( logger . isDebugEnabled ( ) ) { logger . debug ( "retrieveInitialIterators: stream = " + streamName + ", saved sequence numbers = " + offsets ) ; } shardIterators . clear ( ) ; long timeoutAt = System . currentTimeMillis ( ) + timeout ; describeShards ( ) ; if ( shards == null ) return ; for ( String shardId : new ArrayList < String > ( offsets . keySet ( ) ) ) { if ( ! shardsById . containsKey ( shardId ) ) { offsets . remove ( shardId ) ; } } Map < String , ShardIteratorType > itxTypes = new HashMap < String , ShardIteratorType > ( ) ; if ( ! constructMixedIteratorTree ( shardsByParentId . get ( null ) , itxTypes ) ) { populateDefaultIterators ( itxTypes ) ; } if ( logger . isDebugEnabled ( ) ) { logger . debug ( "retrieveInitialIterators: stream = " + streamName + ", iterator types = " + itxTypes ) ; } retrieveIterators ( itxTypes , timeoutAt ) ; }
Reloads shard iterators based on current offsets / default iterator type.
Reloads shard iterators based on current offsets / default iterator type .
// this purges any expired shards from the offsets